什么是字節(jié)碼？Java采用字節(jié)碼的好處是什么？

那我在問(wèn)問(wèn)SparkSQL的字節(jié)碼生成是怎么做的不過(guò)分吧？

代碼生成技術(shù)廣泛應(yīng)用于現(xiàn)代的數(shù)據(jù)庫(kù)系統(tǒng)中。代碼生成是將用戶輸入的表達(dá)式、查詢、存儲(chǔ)過(guò)程等現(xiàn)場(chǎng)編譯成二進(jìn)制代碼再執(zhí)行，相比解釋執(zhí)行的方式，運(yùn)行效率要高很多。尤其是對(duì)于計(jì)算密集型查詢、或頻繁重復(fù)使用的計(jì)算過(guò)程，運(yùn)用代碼生成技術(shù)能達(dá)到數(shù)十倍的性能提升。

Spark SQL在其catalyst模塊的expressions中增加了codegen模塊，對(duì)于SQL語(yǔ)句中的計(jì)算表達(dá)式，比如select num + num from t這種的sql，就可以使用動(dòng)態(tài)字節(jié)碼生成技術(shù)來(lái)優(yōu)化其性能。

spark1.x就已經(jīng)有代碼生成技術(shù)，2.0進(jìn)一步優(yōu)化。

spark1.4，使用代碼生成技術(shù)加速表達(dá)式計(jì)算。表達(dá)式代碼生成（expression codegen）表達(dá)式代碼生成主要是想解決大量虛函數(shù)調(diào)用（Virtual Function Calls），泛化的代價(jià)等。需要注意的是，上面通過(guò)表達(dá)式生成完整的類代碼只有在將 spark.sql.codegen.wholeStage 設(shè)置為 false 才會(huì)進(jìn)行的，否則只會(huì)生成一部分代碼，并且和其他代碼組成 Whole-stage Code。

spark2.0支持同一個(gè)stage的多個(gè)算子組合編譯成一段二進(jìn)制。主要就是將一串的算子，轉(zhuǎn)換成一段代碼（Spark sql轉(zhuǎn)換成java代碼），從而提高性能。一串的算子操作，可以轉(zhuǎn)換成一個(gè)java方法，這樣一來(lái)性能會(huì)有一定的提升。通過(guò)引入全階段代碼生成，大大減少了虛函數(shù)的調(diào)用，減少了 CPU 的調(diào)用，使得 SQL 的執(zhí)行速度有很大提升。

代碼編譯：通過(guò)janino實(shí)現(xiàn)的。

Janino 是一個(gè)超級(jí)小但又超級(jí)快的 Java? 編譯器. 它不僅能像 javac 工具那樣將一組源文件編譯成字節(jié)碼文件，還可以對(duì)一些 Java 表達(dá)式，代碼塊，類中的文本(class body)或者內(nèi)存中源文件進(jìn)行編譯，并把編譯后的字節(jié)碼直接加載到同一個(gè) JVM 中運(yùn)行。Janino 不是一個(gè)開(kāi)發(fā)工具, 而是作為運(yùn)行時(shí)的嵌入式編譯器，比如作為表達(dá)式求值的翻譯器或類似于 JSP 的服務(wù)端頁(yè)面引擎。通過(guò)引入了 Janino 來(lái)編譯生成的代碼，結(jié)果顯示 SQL 表達(dá)式的編譯時(shí)間減少到 5ms。在 Spark 中使用了 ClassBodyEvaluator 來(lái)編譯生成之后的代碼，參見(jiàn) org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator。

Java的IO熟悉吧？

那我問(wèn)個(gè)HDFS上傳和MapReduce讀取文件有什么區(qū)別不過(guò)分吧？

MapReduce的InputFormat常見(jiàn)子類包括:

• TextInputFormat （普通文本文件，MR框架默認(rèn)的讀取實(shí)現(xiàn)類型）

• KeyValueTextInputFormat（讀取一行文本數(shù)據(jù)按照指定分隔符，把數(shù)據(jù)封裝為kv類型）

• NLineInputFormat(讀取數(shù)據(jù)按照行數(shù)進(jìn)行劃分分片)

• CombineTextInputFormat(合并小文件，避免啟動(dòng)過(guò)多MapTask任務(wù))

HDFS讀取文件：

public static void copyBytes(InputStream in, OutputStream out, int buffSize) throws IOException {
        PrintStream ps = out instanceof PrintStream ? (PrintStream)out : null;
        byte[] buf = new byte[buffSize];

        for(int bytesRead = in.read(buf); bytesRead >= 0; bytesRead = in.read(buf)) {
            out.write(buf, 0, bytesRead);
            if (ps != null && ps.checkError()) {
                throw new IOException("Unable to write to output stream.");
            }
        }
    }

Java中的Serializable熟悉吧？

那我問(wèn)問(wèn)你知道的任何一個(gè)框架的序列化是怎么做的？為什么這些框架不用Java原生的序列化不過(guò)分吧？

Flink為什么要自己實(shí)現(xiàn)序列化框架？

目前，絕大多數(shù)的大數(shù)據(jù)計(jì)算框架都是基于JVM實(shí)現(xiàn)的，為了快速地計(jì)算數(shù)據(jù)，需要將數(shù)據(jù)加載到內(nèi)存中進(jìn)行處理。當(dāng)大量數(shù)據(jù)需要加載到內(nèi)存中時(shí)，如果使用Java序列化方式來(lái)存儲(chǔ)對(duì)象，占用的空間會(huì)較大降低存儲(chǔ)傳輸效率。

例如：一個(gè)只包含布爾類型的對(duì)象需要占用16個(gè)字節(jié)的內(nèi)存：對(duì)象頭要占8個(gè)字節(jié)、boolean屬性占用1個(gè)字節(jié)、對(duì)齊填充還要占用7個(gè)字節(jié)。

Java序列化方式存儲(chǔ)對(duì)象存儲(chǔ)密度是很低的。也是基于此，F(xiàn)link框架實(shí)現(xiàn)了自己的內(nèi)存管理系統(tǒng)，在Flink自定義內(nèi)存池分配和回收內(nèi)存，然后將自己實(shí)現(xiàn)的序列化對(duì)象存儲(chǔ)在內(nèi)存塊中。

Java生態(tài)系統(tǒng)中有挺多的序列化框架，例如：Kryo、Avro、ProtoBuf等。Flink自己實(shí)現(xiàn)了一套序列化系統(tǒng)可以讓我們編寫程序的時(shí)候，盡快地發(fā)現(xiàn)問(wèn)題，更加節(jié)省內(nèi)存空間，并直接進(jìn)行二進(jìn)制數(shù)據(jù)的處理。

例如：POJO類型對(duì)應(yīng)的是PojoTypeInfo、基礎(chǔ)數(shù)據(jù)類型數(shù)組對(duì)應(yīng)的是BasicArrayTypeInfo、Map類型對(duì)應(yīng)的是MapTypeInfo、值類型對(duì)應(yīng)的是ValueTypeInfo。

除了對(duì)類型地描述之外，TypeInformation還提供了序列化的支撐。在TypeInformation中有一個(gè)方法：createSerializer方法，它用來(lái)創(chuàng)建序列化器，序列化器中定義了一系列的方法。其中，通過(guò)serialize和deserialize方法，可以將指定類型進(jìn)行序列化。并且，F(xiàn)link的這些序列化器會(huì)以稠密的方式來(lái)將對(duì)象寫入到內(nèi)存中。

Spark的目標(biāo)是在便利與性能中取得平衡，所以提供2種序列化的選擇。

• Java serialization

在默認(rèn)情況下，Spark會(huì)使用Java的ObjectOutputStream框架對(duì)對(duì)象進(jìn)行序列化，并且可以與任何實(shí)現(xiàn)java.io.Serializable的類一起工作。您還可以通過(guò)擴(kuò)展java.io.Externalizable來(lái)更緊密地控制序列化的性能。Java序列化是靈活的，但通常相當(dāng)慢，并且會(huì)導(dǎo)致許多類的大型序列化格式。

• Kryo serialization

Spark還可以使用Kryo庫(kù)（版本2）來(lái)更快地序列化對(duì)象。Kryo比Java串行化（通常多達(dá)10倍）要快得多，也更緊湊，但是不支持所有可串行化類型，并且要求您提前注冊(cè)您將在程序中使用的類，以獲得最佳性能

Kryo serialization 性能和序列化大小都比默認(rèn)提供的 Java serialization 要好，但是使用Kryo需要將自定義的類先注冊(cè)進(jìn)去，使用起來(lái)比Java serialization麻煩。自從Spark 2.0.0以來(lái)，我們?cè)谑褂煤?jiǎn)單類型、簡(jiǎn)單類型數(shù)組或字符串類型的簡(jiǎn)單類型來(lái)調(diào)整RDDs時(shí)，在內(nèi)部使用Kryo序列化器。

Java中的反射了解吧？

那我問(wèn)問(wèn)Spark SQL將RDD轉(zhuǎn)換為DataFrame如何實(shí)現(xiàn)的不過(guò)分吧？

Spark SQL支持將現(xiàn)有RDDS轉(zhuǎn)換為DataFrame的兩種不同方法，其實(shí)也就是隱式推斷或者顯式指定DataFrame對(duì)象的Schema。

• 1.使用反射機(jī)制( Reflection )推理出schema (結(jié)構(gòu)信息)

第一種將RDDS轉(zhuǎn)化為DataFrame的方法是使用Spark SQL內(nèi)部反射機(jī)制來(lái)自動(dòng)推斷包含特定類型對(duì)象的RDD的schema

(RDD的結(jié)構(gòu)信息)進(jìn)行隱式轉(zhuǎn)化。采用這種方式轉(zhuǎn)化為DataFrame對(duì)象，往往是因?yàn)楸晦D(zhuǎn)化的RDD[T]所包含的T對(duì)象本身就是具有典型-一維表嚴(yán)格的字段結(jié)構(gòu)的對(duì)象，因此Spark SQL很容易就可以自動(dòng)推斷出合理的Schema這種基于反射機(jī)制隱式地創(chuàng)建DataFrame的方法往往僅需更簡(jiǎn)潔的代碼即可完成轉(zhuǎn)化，并且運(yùn)行效果良好。

Spark SQL的Scala接口支持自動(dòng)將包含樣例類( case class對(duì)象的RDD轉(zhuǎn)換為DataFrame對(duì)象。在樣例類的聲明中 已預(yù)先定義了表的結(jié)構(gòu)信息，內(nèi)部通過(guò)反射機(jī)制即可讀取樣例類的參數(shù)的名稱、類型，轉(zhuǎn)化為DataFrame對(duì)象的Schema.樣例類不僅可以包含Int、Double、String這樣的簡(jiǎn)單數(shù)據(jù)類型，也可以嵌套或包含復(fù)雜類型，例如Seq或Arrays。

• 2.由開(kāi)發(fā)者指定Schema

RDD轉(zhuǎn)化DataFrame的第二種方法是通過(guò)編程接口，允許先構(gòu)建個(gè)schema,然后將其應(yīng)用到現(xiàn)有的RDD(Row),較前一種方法由樣例類或基本數(shù)據(jù)類型 (Int、String) 對(duì)象組成的RDD加過(guò)toDF ()直接隱式轉(zhuǎn)化為DataFrame不同，不僅需要根據(jù)需求、以及數(shù)據(jù)結(jié)構(gòu)構(gòu)建Schema,而且需要將RDD[T]轉(zhuǎn)化為Row對(duì)象組成的RDD (RDD[Row]),這種方法雖然代碼量一些，但也提供了更高的自由度，更加靈活。

LinkedHashMap知道吧？

那我問(wèn)問(wèn)Spark做內(nèi)存管理有一種叫"存儲(chǔ)內(nèi)存"，用了什么數(shù)據(jù)結(jié)構(gòu)?淘汰策略是什么不過(guò)分吧？

由于同一個(gè) Executor 的所有的計(jì)算任務(wù)共享有限的存儲(chǔ)內(nèi)存空間，當(dāng)有新的 Block 需要緩存但是剩余空間不足且無(wú)法動(dòng)態(tài)占用時(shí)，就要對(duì) LinkedHashMap 中的舊 Block 進(jìn)行淘汰（Eviction），而被淘汰的 Block 如果其存儲(chǔ)級(jí)別中同時(shí)包含存儲(chǔ)到磁盤的要求，則要對(duì)其進(jìn)行落盤（Drop），否則直接刪除該 Block

遍歷 LinkedHashMap 中 Block，按照最近最少使用（LRU）的順序淘汰，直到滿足新 Block 所需的空間。其中LRU 是 LinkedHashMap 的特性。

快速失?。╢ail-fast）和安全失敗（fail-safe）聽(tīng)過(guò)吧？

Flink哪里的設(shè)計(jì)用到了fail-fast理念？

In case of a program failure (due to machine-, network-, or software failure), Flink stops the distributed streaming dataflow. The system then restarts the operators and resets them to the latest successful checkpoint. The input streams are reset to the point of the state snapshot. Any records that are processed as part of the restarted parallel dataflow are guaranteed to not have been part of the previously checkpointed state.

你很懂ConcurrentHashMap？

### 那我問(wèn)問(wèn)Spark/Flink中哪里用到了ConcurrentHashMap？

友情提示：Spark中的所有Settings，F(xiàn)link中的ParameterUtil，太多了。