Hadoop 之 壓縮Using snappy-java in SnappyCodechttps://issues.apache.org/jira/browse/HADOOP-17125https://github.com/apache/hadoop/pull/2201/files大數(shù)據(jù)各類(lèi)引擎關(guān)于 Snappy 的使用形式各自相應(yīng)的jiraparquet-cpp各個(gè)依賴項(xiàng)對(duì)比Hadoop NativeTask/usr/lib64/libsnappy.so.1/usr/local/hadoop-3.3.3/lib/native/libnativetask.soNativeTask JVM 啟動(dòng) │ ▼ System.loadLibrary(nativetask)[NativeRuntime.java:53]│ ▼ ld.so 解析 DT_NEEDED → 找 libsnappy.so.1│ ├── 查找順序:│1.LD_LIBRARY_PATH │2./etc/ld.so.cache由/etc/ld.so.conf*構(gòu)建 │3.默認(rèn)路徑/usr/lib64,/lib64,/usr/lib,/lib │ ├── ? 找到如/usr/lib64/libsnappy.so.1 │ → libnativetask.so 加載成功 │ →supportsCompressionCodec(SnappyCodec)JNI_TRUE │ → spill 壓縮走系統(tǒng) libsnappy 的snappy_compress()│ └── ? 找不到 → UnsatisfiedLinkError → NativeRuntime 靜態(tài)塊catchnativeLibraryLoadedfalse → NativeMapOutputCollectorDelegator.init()拋 InvalidJobConfException → createSortingCollector 回退 MapOutputBufferJavaNativeTask 回退機(jī)制static{try{System.loadLibrary(nativetask);// ← libnativetask.so 加載LOG.info(Nativetask JNI library loaded.);nativeLibraryLoadedtrue;}catch(finalThrowable t){// Ignore failuresLOG.error(Failed to load nativetask JNI library with error: t);LOG.info(java.library.pathSystem.getProperty(java.library.path));LOG.info(LD_LIBRARY_PATHSystem.getenv(LD_LIBRARY_PATH));// nativeLibraryLoaded 保持 false不拋異常}}publicvoidinit(Context context)throws IOException,ClassNotFoundException{this.contextcontext;this.jobcontext.getJobConf();Platforms.init(job);if(job.getNumReduceTasks()0){String messageThere is no reducer, no need to use native output collector;LOG.error(message);thrownewInvalidJobConfException(message);}Class?comparatorClassjob.getClass(MRJobConfig.KEY_COMPARATOR,null,RawComparator.class);if(comparatorClass!null!Platforms.define(comparatorClass)){String messageNative output collector doesnt support customized java comparator job.get(MRJobConfig.KEY_COMPARATOR);LOG.error(message);thrownewInvalidJobConfException(message);}if(!QuickSort.class.getName().equals(job.get(Constants.MAP_SORT_CLASS))){String messageNative-Task doesnt support sort class job.get(Constants.MAP_SORT_CLASS);LOG.error(message);thrownewInvalidJobConfException(message);}if(job.getBoolean(MRConfig.SHUFFLE_SSL_ENABLED_KEY,false)true){String messageNative-Task doesnt support secure shuffle;LOG.error(message);thrownewInvalidJobConfException(message);}finalClass?keyClsjob.getMapOutputKeyClass();try{SuppressWarnings(rawtypes)finalINativeSerializer serializerNativeSerialization.getInstance().getSerializer(keyCls);if(nullserializer){String messageKey type not supported. Cannot find serializer for keyCls.getName();LOG.error(message);thrownewInvalidJobConfException(message);}elseif(!Platforms.support(keyCls.getName(),serializer,job)){String messageNative output collector doesnt support this key, this key is not comparable in native: keyCls.getName();LOG.error(message);thrownewInvalidJobConfException(message);}}catch(finalIOException e){String messageCannot find serializer for keyCls.getName();LOG.error(message);thrownewIOException(message);}finalboolean retNativeRuntime.isNativeLibraryLoaded();if(ret){if(job.getBoolean(MRJobConfig.MAP_OUTPUT_COMPRESS,false)){String codecjob.get(MRJobConfig.MAP_OUTPUT_COMPRESS_CODEC);if(!NativeRuntime.supportsCompressionCodec(codec.getBytes(Charsets.UTF_8))){String messageNative output collector doesnt support compression codec codec;LOG.error(message);thrownewInvalidJobConfException(message);}}NativeRuntime.configure(job);finallongupdateIntervaljob.getLong(Constants.NATIVE_STATUS_UPDATE_INTERVAL,Constants.NATIVE_STATUS_UPDATE_INTERVAL_DEFVAL);updaternewStatusReportChecker(context.getReporter(),updateInterval);updater.start();}else{String messageNativeRuntime cannot be loaded, please check that libnativetask.so is in hadoop library dir;LOG.error(message);thrownewInvalidJobConfException(message);}this.handlernull;try{finalClassKoKClass(ClassK)job.getMapOutputKeyClass();finalClassKoVClass(ClassK)job.getMapOutputValueClass();finalTaskAttemptID idcontext.getMapTask().getTaskID();finalTaskContext taskContextnewTaskContext(job,null,null,oKClass,oVClass,context.getReporter(),id);handlerNativeCollectorOnlyHandler.create(taskContext);}catch(finalIOException e){String messageNative output collector cannot be loaded;;LOG.error(message);thrownewIOException(message,e);}LOG.info(Native output collector can be successfully enabled!);}回退點(diǎn)位出現(xiàn)了hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/MapTask.javaClass?[]collectorClassesjob.getClasses(JobContext.MAP_OUTPUT_COLLECTOR_CLASS_ATTR,MapOutputBuffer.class);// ↑ 讀取 mapreduce.job.map.output.collector.class// 你的配置: NativeMapOutputCollectorDelegator, MapTask$MapOutputBuffer// 逗號(hào)分隔 → 數(shù)組 [Delegator, MapOutputBuffer]intremainingCollectorscollectorClasses.length;Exception lastExceptionnull;for(Class clazz:collectorClasses){try{// 校驗(yàn)實(shí)現(xiàn)接口if(!MapOutputCollector.class.isAssignableFrom(clazz)){throw...;}// 實(shí)例化 初始化MapOutputCollectorKEY,VALUEcollectorReflectionUtils.newInstance(subclazz,job);collector.init(context);// ← ①失敗就在這里拋LOG.info(Map output collector class collector.getClass().getName());returncollector;// ← 成功即返回}catch(Exception e){String msgUnable to initialize MapOutputCollector clazz.getName();if(--remainingCollectors0){msg (remainingCollectors more collector(s) to try);}lastExceptione;LOG.warn(msg,e);// ← 打 WARN 繼續(xù)試下一個(gè)}}// 全部失敗才拋 IOExceptionOS /usr/lib64/libsnappy.so.1#rpm-qi snappyName:snappy Version:1.1.9Release:2.oe2203sp3 Architecture:x86_64 Install Date:Wed17Dec202503:03:47PM CST Group:Unspecified Size:50212License:BSD Signature:RSA/SHA256,Fri05Jan202410:30:58AM CST,Key ID b943e4ef06bfb671 Source RPM:snappy-1.1.9-2.oe2203sp3.src.rpm Build Date:Thu28Dec202301:30:01PM CST Build Host:dc-64g.compass-ci Packager:http://openeuler.orgURL:https://github.com/google/snappySummary:A fast compressor/decompressor Description:Snappy is a compression/decompression library.It doesnotaimformaximum compression,orcompatibility with any other compression library;instead,it aimsforvery high speedsandreasonable compression./usr/local/hadoop-3.3.3/share/hadoop/hdfs/lib/snappy-java-1.1.8.2.jarhttps://github.com/xerial/snappy-javaunzip snappy-java-1.1.8.2.jar-d snappy-java-unzip:/tmp/snappy-java-unzip# find./-name*.so./org/xerial/snappy/native/Linux/x86_64/libsnappyjava.so./org/xerial/snappy/native/Linux/x86/libsnappyjava.so./org/xerial/snappy/native/Linux/android-arm/libsnappyjava.so./org/xerial/snappy/native/Linux/aarch64/libsnappyjava.so./org/xerial/snappy/native/Linux/ppc64le/libsnappyjava.so./org/xerial/snappy/native/Linux/ppc/libsnappyjava.so./org/xerial/snappy/native/Linux/ppc64/libsnappyjava.so./org/xerial/snappy/native/Linux/s390x/libsnappyjava.so./org/xerial/snappy/native/Linux/armv7/libsnappyjava.so./org/xerial/snappy/native/Linux/arm/libsnappyjava.so./org/xerial/snappy/native/Linux/armv6/libsnappyjava.so./org/xerial/snappy/native/FreeBSD/x86_64/libsnappyjava.so./org/xerial/snappy/native/SunOS/sparc/libsnappyjava.so./org/xerial/snappy/native/SunOS/x86_64/libsnappyjava.so./org/xerial/snappy/native/SunOS/x86/libsnappyjava.solibsnappyjava.so 來(lái)源不同hadoop version 對(duì) snappy lib 的使用方式snappy-java 跨平臺(tái)設(shè)計(jì)完整流程圖