3.1.1　流數(shù)據(jù)類型

Flink以一種獨特的方式處理數(shù)據(jù)類型和序列化，它包含自己的類型描述符、泛型類型提取和類型序列化框架。基于Java和Scala語言，F(xiàn)link實現(xiàn)了一套自己的類型系統(tǒng)，它支持很多類型，包括：

（1）基本類型。

（2）數(shù)組類型。

（3）復(fù)合類型。

（4）輔助類型。

（5）通用類型。

詳細(xì)的Flink類型系統(tǒng)如圖3-1所示。

Flink針對Java和Scala的DataStream API要求流數(shù)據(jù)的內(nèi)容必須是可序列化的。Flink內(nèi)置了以下類型數(shù)據(jù)的序列化器。

（1）基本數(shù)據(jù)類型：String、Long、Integer、Boolean、Array。

（2）復(fù)合數(shù)據(jù)類型：Tuple、POJO、Scala case class。

對于其他類型，F(xiàn)link會返回Kryo。也可以在Flink中使用其他序列化器。Avro尤其得到了很好的支持。

1．Java DataStream API使用的流數(shù)據(jù)類型

對于Java API，F(xiàn)link定義了自己的Tuple1~Tuple25類型來表示元組類型，代碼如下：

在Java中，POJO（Plain Old Java Object）是這樣的Java類：

（1）有一個無參的默認(rèn)構(gòu)造器。

（2）所有的字段要么是public的，要么有一個默認(rèn)的getter和setter。

例如，定義一個名為Person的POJO類，代碼如下：

2．Scala DataStream API使用的流數(shù)據(jù)類型

對于元組，可以使用Scala自己的Tuple類型，代碼如下：

對于對象類型，使用case class（相當(dāng)于Java中的JavaBean），代碼如下：

3．Flink類型系統(tǒng)

對于創(chuàng)建的任意一個POJO類型，看起來它是一個普通的Java Bean，在Java中，可以使用Class來描述該類型，但其實在Flink引擎中，它被描述為PojoTypeInfo，而PojoTypeInfo是TypeInformation的子類。

TypeInformation是Flink類型系統(tǒng)的核心類。Flink使用TypeInformation來描述所有Flink支持的數(shù)據(jù)類型，就像Java中的Class類型一樣。每種Flink支持的數(shù)據(jù)類型都對應(yīng)于TypeInformation的子類。例如POJO類型對應(yīng)的是PojoTypeInfo、基礎(chǔ)數(shù)據(jù)類型數(shù)組對應(yīng)的是BasicArrayTypeInfo、Map類型對應(yīng)的是MapTypeInfo、值類型對應(yīng)的是ValueTypeInfo。

除了對類型的描述，TypeInformation還提供了對序列化的支持。在TypeInformation中有一種方法：createSerializer方法，它用來創(chuàng)建序列化器，在序列化器中定義了一系列的方法，其中，通過serialize和deserialize方法可以將指定類型進行序列化，并且Flink的這些序列化器會以稠密的方式將對象寫入內(nèi)存中。Flink中也提供了非常豐富的序列化器。在基于Flink類型系統(tǒng)支持的數(shù)據(jù)類型進行編程時，F(xiàn)link在運行時會推斷出數(shù)據(jù)類型的信息，程序員在基于Flink編程時，幾乎不需要關(guān)心類型和序列化。

4．類型與Lambda表達式支持

在編譯時，編譯器能夠從Java源代碼中讀取完整的類型信息，并強制執(zhí)行類型的約束，但在生成class字節(jié)碼時會將參數(shù)化類型信息刪除，這就是類型擦除。類型擦除可以確保不會為泛型創(chuàng)建新的Java類，泛型是不會產(chǎn)生額外的開銷的。也就是說，泛型只是在編譯器編譯時能夠理解該類型，但編譯后在執(zhí)行時泛型會被擦除。

為了便于說明，參看下面的代碼：

以上是一段Java的泛型方法，但在編譯后，編譯器會將未綁定類型的T擦除，替換為Object，也就是編譯之后的代碼如下：

泛型只能防止在運行時出現(xiàn)類型錯誤，但運行時會出現(xiàn)以下異常，而且Flink以非常友好的方式提示：

因為Java編譯器可將類型擦除，所以Flink根本無法推斷算子（例如flatMap）要輸出的類型是什么，所以在Flink中使用Lambda表達式時，為了防止因類型擦除而出現(xiàn)運行時錯誤，需要指定TypeInformation或者TypeHint。

創(chuàng)建TypeInformation，代碼如下：

創(chuàng)建TypeHint，代碼如下：