報錯日志描述

---

• 報錯日志：

The generic type parameters of 'Collector' are missing. In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved
// 缺少“Collector”的泛型類型參數(shù)。在許多情況下，當涉及Java泛型時，lambda方法不能為自動類型提取提供足夠的信息

• 建議日志：

The return type of function 'main(TypeErasure.java:23)' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.
// 由于類型擦除，無法自動確定函數(shù)“main（TypeErasure.java:23）”的返回類型
// 您可以通過對轉(zhuǎn)換調(diào)用的結(jié)果使用returns（…）方法，
// 或者通過讓函數(shù)實現(xiàn)“ResultTypeQueryable”接口來提供類型信息提示

問題描述

---

• Java 8 自身存在的問題：
  • 在使用Java API 寫 Lambda 的時候，JVM 運行時會擦除類型（泛型類型）
    • Flink 無法準確獲取到數(shù)據(jù)類型
    • 此時就需要我們手動指定類型
  • Scala 就很好的解決的這個問題，無需指定類型
• 正常情況下編寫Java： 會需要手動設(shè)置輸入格式，和輸出格式

source.flatMap(new FlatMapFunction<String, Object>() {})    // 輸入格式 String，輸出格式 Object

• Java 使用 Lambda 表達式：

source.flatMap(()->{// 輸入和輸出格式都沒有指定，java 8 無法做自動類型推斷})

• 需要手動指定類型：

source.flatMap(()->{// 所以需要手動指定類型},Types.類型)

報錯解決

---

• 解決方案：
  • 在Flink中經(jīng)常使用的類型已經(jīng)預定義在了 Types 中它們的 serializer/deserializer 和 Comparator 已經(jīng)定義好了
  • Tuple 類型既可以使用 TypeHint 指定又可以使用 Types 指定
• 問題代碼：

SingleOutputStreamOperator<String> flatMap = source.flatMap((line, collect) -> {String[] words = line.split(" ");for (String word : words) {collect.collect(word);}});

• 修改后： Types 方法

SingleOutputStreamOperator<String> flatMap = source.flatMap((line, collect) -> {String[] words = line.split(" ");for (String word : words) {collect.collect(word);}},Types.STRING);      // 指定類型

• 其他案例：

SingleOutputStreamOperator<Tuple2<Object, Integer>> map = flatMap.map(word -> Tuple2.of(word, 1),Types.TUPLE(Types.STRING, Types.INT));

• ps：簡單易懂，我比較喜歡用這種

---

其他方法

---

方法一：TypeInformation

---

• TypeInformation 是Flink類型系統(tǒng)的核心，是生成序列化/反序列化工具和 Comparator 的工具類
  • 同時它還是連接schema和編程語言內(nèi)部類型系統(tǒng)的橋梁
• 可以使用 of 方法創(chuàng)建 TypeInformation ：
  • of(Class typeClass)：從 Class 創(chuàng)建
  • of(TypeHint typeHint)：從 TypeHint 創(chuàng)建

方法二：TypeHint

---

• 由于泛型類型在運行時會被JVM擦除，所以說我們無法使用
  • TypeInformation.of(XXX.class) 方式指定帶有泛型的類型
• 為了可以支持泛型類型，Flink引入了 TypeHint
  • 例如我們需要獲取 Tuple2<String, Long> 的類型信息，可以使用如下方式：

TypeInformation<Tuple2<String, Long>> info = TypeInformation.of(new TypeHint<Tuple2<String, Long>>(){});
// 或者
TypeInformation<Tuple2<String, Long>> info = new TypeHint<Tuple2<String, Long>>(){}.getTypeInfo();

---

下班…