spark 源碼分析之十三 -- SerializerManager剖析


對SerializerManager的說明:

它是為各種Spark組件配置序列化,壓縮和加密的組件,包括自動選擇用於shuffle的Serializer。spark中的數據在network IO 或 local disk IO傳輸過程中。都需要序列化。其默認的 Serializer 是 org.apache.spark.serializer.JavaSerializer,在一定條件下,可以使用kryo,即org.apache.spark.serializer.KryoSerializer。

 

支持的兩種序列化方式

即值的類型是八種基本類型中一種或null或String,都會使用kryo,否則使用默認序列化方式,即java序列化方式。

 

它還負責讀寫Block流是否使用壓縮:

 

數據流是否支持壓縮

默認情況下:

其中,如果使用壓縮,默認的壓縮是 lz4, 可以通過參數 spark.io.compression.codec 來配置。它支持的所有壓縮類型如下:

讀寫數據流如何支持壓縮

其中,支持壓縮的InputStream和OutputStream是對原來的InputStream和OutputStream做了包裝。我們以LZ4BlockOutputStream為例說明。

調用如下函數返回支持壓縮的OutputStream:

首先,LZ4BlockOutputStream的繼承關系如下:

 

被包裝的類被放到了FilterOutputStream類的out 字段中,如下:

outputStream核心方法就是write。直接來看LZ4BlockOutputStream的write方法:

 

其中buffer是一個byte 數組,默認是 32k,可以通過spark.io.compression.lz4.blockSize 參數來指定,在LZ4BlockOutputStream類中用blockSize保存。

 

重點看flushBufferedData方法:

 

方法內部實現思路如下:

外部寫入到buffer中的數據經過compressor壓縮到compressorBuffer中,然后再寫入一些magic,最終將壓縮的buffer寫入到out中,write操作結束。

 

可見,數據的壓縮是由 LZ4BlockOutputStream 負責的,壓縮之后的數據被寫入到目標outputStream中。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM