写给大忙人的Flink的Data Types
- 2020 年 3 月 18 日
- 筆記
一.Flink 中 Data Type 组成
- 基本数据类型：java 的 8 中基本数据类型加上它们各自的包装类型，在加上 void , String, Date,BigDecimal, BigInteger.
- 基本数据类型的数据和 Object 类型的数组
- 复合类型 1.Flink Java Tuples 2. scala case classes 3. Row 4. POJOs: 如果要被 Flink 识别的也允许按 name 引用的话，需要复符合一定的规则(否则的话，会被当做泛型处理) 1). 这个类是 pulic 的并且没有非静态的内部类。 2). 得有一个没有参数的 pulic 构造器 3).所有非静态的非 transient 的属性(包括所有的父类)都必须是 pulic 或者符合 java beans 命名规范的 getter setter 方法。
- 辅助类型 (集合类、Option、Either 等)
- 泛型：不会被 Flink 自带的序列化器序列化，而被是 Kryo
二、Flink 是如何处理 Data Type 的 首先Flink会根据自身的序列化器进行序列化，如果不行，则默认回退到 Kryo 序列化器进行序列化。
可能碰到的问题，如下：
- Registering subtypes 如果方法签名是父类，而返回或者使用的是子类，也就是所谓的协变返回类型关于协变返回类型。让 Flink 知道所有的子类可以在一定的程度上提高性能。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.registerType(KuduTableDesc.class);
- Registering custom serializers 虽然 Flink 自己序列化不了的会给 Kryo，但是 Kryo 也不能很好的处理掉所有的类型，这个时候就要自定义序列化器了。
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, MyCustomSerializer.class);
- Adding Type Hints Flink 可能无法推断出泛型的类型，仅仅在 Java Api 中时必要的。
DataSet<SomeType> result = dataSet .map(new MyGenericNonInferrableFunction<Long, SomeType>()) .returns(SomeType.class); DataSet<SomeType> result = dataSet .map(new MyGenericNonInferrableFunction<Long, SomeType>()) .returns(new TypeHint<SomeType.class});
- Manually creating a TypeInformation Flink 可能无法推断出泛型的类型时
TypeInformation<String> info = TypeInformation.of(String.class); TypeInformation<Tuple2<String, Double>> info = TypeInformation.of(new TypeHint<Tuple2<String, Double>>(){});
三、常见的 returns 的使用
.returns(Types.TUPLE(Types.INT,Types.INT)) .returns(Types.STRING) .returns(TypeInformation.of(String.class)) .returns(new TypeHint<Tuple2<String, String>>(){}) .returns(TypeInformation.of(new TypeHint<Tuple2<ConsumerRecord, String>>() {})) .returns(SomeType.class)
