Flink處理數據介紹

Flink流應用程序處理是以數據對象表示的事件流。在Flink內部，處理數據對象，通過被序列化和反序列化進行網絡傳送，從狀態后端、檢查點和保存點讀取它們。

為了有效地做到這一點，Flink需要明確知道應用程序所處理的數據類型。Flink使用類型信息的概念來表示數據類型，并為每個數據類型生成特定的序列化器、反序列化器和比較器。

Flink還具有一個類型提取系統，該系統分析函數的輸入和返回類型，以自動獲取類型信息，從而獲得序列化器和反序列化器。

Flink支持Java和Scala中所有常見數據類型，在某些情況下，例如lambda函數或泛型類型，需要顯式地聲明類型信息，應用程序才能正常工作。

使用最廣泛的類型有如下幾種：

基礎數據類型

Flink支持Java、Scala所有基本數據類型，如Integer、String、Double、Long 等

java編寫示例：

        DataStream<Long> stream1 = env.fromElements(1L, 2L, 3L, 4L);SingleOutputStreamOperator<Object> streamMap = stream1.map(new MapFunction<Long, Object>() {@Overridepublic Object map(Long n) throws Exception {return n+1;}});

Java 對象（POJOs）

java編寫示例：

        DataStream<Persion> stream2 = env.fromElements(new Persion("王五",25),new Persion("趙六",26));DataStream<Persion> stream2Filter = stream2.filter(new FilterFunction<Persion>() {@Overridepublic boolean filter(Persion persion) throws Exception {if(persion.age>25){return true;}return false;}});

Java和Scala元組（Tuples）

java編寫示例：

        DataStream<Tuple2<String, Integer>> stream3 =  stream2.flatMap(new FlatMapFunction<Persion, Tuple2<String, Integer>>() {@Overridepublic void flatMap(Persion persion, Collector<Tuple2<String, Integer>> collector) throws Exception {//形成一個元組Tuple2<String, Integer> tp = Tuple2.of(persion.name, persion.age+1);//將組成的Tuple放入到 Collector 集合，并輸出collector.collect(tp);}});

Scala 樣例類（case classes）

示例：

case class Person(name: String, age: Int) 
val persons: DataStream[Person] = env.fromElements(
Person("王五", 25), 
Person("趙六", 26) )
persons.filter(p => p.age > 25)

其它數據類型 Arrays, Lists, Maps, Enums等

Flink 也可支持Java 和Scala 語言中的其它數據類型，如Java語言的數據類型：ArrayList，HashMap，Enum等。

總結

以上是生活随笔為你收集整理的java 汇率使用的数据类型_Flink计算支持的数据类型的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。