flink.12 序列化

flink.12 序列化

一.元组(Tuples and Case Classes )

对java来说Tuples是flink自带的一种类, 对于scala来说flink没有提供类似Tuples的类, 因为scala天生自带了一种特殊类 case class.

主要说说java版的Tuples, Java API 提供从Tuple1最高到Tuple25. 元组的每个字段都可以是任意 Flink 类型, 1 25这个数字的意思是参数的个数.
Tuple1 t1;
Tuple2<String,String> t2;
Tuple3<String,Integer,Long> t3;
访问Tuple中的数据flink提供了便捷的方法,比如:Field(int position)。字段索引从 0 开始, 或者tuple.f格式, f后面跟数字,也是从0开始.

比如 Tuple3<String, String,Integer> t3=new Tuple3<String,
String,Integer>(“张三”, “男”,20)
如果要访问年龄有下面两种方法:

1. t3.f2

case class WordCount(word: String, count: Int)
val input = env.fromElements(WordCount("hello", 1),WordCount("world", 2)) // Case Class Data Set

2.java版

DataStream<Tuple2<String, Integer>> wordCounts = env.fromElements(new Tuple2<String, Integer>("hello", 1),new Tuple2<String, Integer>("world", 2));wordCounts.map(new MapFunction<Tuple2<String, Integer>, Integer>() {@Overridepublic Integer map(Tuple2<String, Integer> value) throws Exception {return value.f1;}
});

我们来看看java的Tuple,Tuple是一个实现了java序列化接口的一个顶层接口,Tuple2 Tupe3 …Tuple25是实现了Tuple接口的具体的类
Tuple接口实现了java的序列化接口,public abstract class Tuple implements java.io.Serializable
下面是Tuple2的源码:

/** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements.  See the NOTICE file* distributed with this work for additional information* regarding copyright ownership.  The ASF licenses this file* to you under the Apache License, Version 2.0 (the* "License"); you may not use this file except in compliance* with the License.  You may obtain a copy of the License at**     .0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/// --------------------------------------------------------------
//  THIS IS A GENERATED SOURCE FILE. DO NOT EDIT!
//  GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
// --------------------------------------------------------------package org.apache.flink.api.java.tuple;import org.apache.flink.annotation.Public;
import org.apache.flink.util.StringUtils;/*** A tuple with 2 fields. Tuples are strongly typed; each field may be of a separate type. The* fields of the tuple can be accessed directly as public fields (f0, f1, ...) or via their position* through the {@link #getField(int)} method. The tuple field positions start at zero.** <p>Tuples are mutable types, meaning that their fields can be re-assigned. This allows functions* that work with Tuples to reuse objects in order to reduce pressure on the garbage collector.** <p>Warning: If you subclass Tuple2, then be sure to either** <ul>*   <li>not add any new fields, or*   <li>make it a POJO, and always declare the element type of your DataStreams/DataSets to your*       descendant type. (That is, if you have a "class Foo extends Tuple2", then don't use*       instances of Foo in a DataStream&lt;Tuple2&gt; / DataSet&lt;Tuple2&gt;, but declare it as*       DataStream&lt;Foo&gt; / DataSet&lt;Foo&gt;.)* </ul>** @see Tuple* @param <T0> The type of field 0* @param <T1> The type of field 1*/
@Public
public class Tuple2<T0, T1> extends Tuple {private static final long serialVersionUID = 1L;/** Field 0 of the tuple. */public T0 f0;/** Field 1 of the tuple. */public T1 f1;/** Creates a new tuple where all fields are null. */public Tuple2() {}/*** Creates a new tuple and assigns the given values to the tuple's fields.** @param f0 The value for field 0* @param f1 The value for field 1*/public Tuple2(T0 f0, T1 f1) {this.f0 = f0;this.f1 = f1;}@Overridepublic int getArity() {return 2;}@Override@SuppressWarnings("unchecked")public <T> T getField(int pos) {switch (pos) {case 0:return (T) this.f0;case 1:return (T) this.f1;default:throw new IndexOutOfBoundsException(String.valueOf(pos));}}@Override@SuppressWarnings("unchecked")public <T> void setField(T value, int pos) {switch (pos) {case 0:this.f0 = (T0) value;break;case 1:this.f1 = (T1) value;break;default:throw new IndexOutOfBoundsException(String.valueOf(pos));}}/*** Sets new values to all fields of the tuple.** @param f0 The value for field 0* @param f1 The value for field 1*/public void setFields(T0 f0, T1 f1) {this.f0 = f0;this.f1 = f1;}/*** Returns a shallow copy of the tuple with swapped values.** @return shallow copy of the tuple with swapped values*/public Tuple2<T1, T0> swap() {return new Tuple2<T1, T0>(f1, f0);}// -------------------------------------------------------------------------------------------------// standard utilities// -------------------------------------------------------------------------------------------------/*** Creates a string representation of the tuple in the form (f0, f1), where the individual* fields are the value returned by calling {@link Object#toString} on that field.** @return The string representation of the tuple.*/@Overridepublic String toString() {return "("+ StringUtils.arrayAwareToString(this.f0)+ ","+ StringUtils.arrayAwareToString(this.f1)+ ")";}/*** Deep equality for tuples by calling equals() on the tuple members.** @param o the object checked for equality* @return true if this is equal to o.*/@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (!(o instanceof Tuple2)) {return false;}@SuppressWarnings("rawtypes")Tuple2 tuple = (Tuple2) o;if (f0 != null ? !f0.equals(tuple.f0) : tuple.f0 != null) {return false;}if (f1 != null ? !f1.equals(tuple.f1) : tuple.f1 != null) {return false;}return true;}@Overridepublic int hashCode() {int result = f0 != null ? f0.hashCode() : 0;result = 31 * result + (f1 != null ? f1.hashCode() : 0);return result;}/*** Shallow tuple copy.** @return A new Tuple with the same fields as this.*/@Override@SuppressWarnings("unchecked")public Tuple2<T0, T1> copy() {return new Tuple2<>(this.f0, this.f1);}/*** Creates a new tuple and assigns the given values to the tuple's fields. This is more* convenient than using the constructor, because the compiler can infer the generic type* arguments implicitly. For example: {@code Tuple3.of(n, x, s)} instead of {@code new* Tuple3<Integer, Double, String>(n, x, s)}*/public static <T0, T1> Tuple2<T0, T1> of(T0 f0, T1 f1) {return new Tuple2<>(f0, f1);}
}

所以flink针对Tuple的序列化,底层还是用的java的序列化,并没有用其他的序列化框架.

二.java或者scala 遵循下述规范的类(POJOs )

普通类有以下要求:

1. 必须是public 类
2. 必须有一个不带参数的默认构造函数
3. 字段必须也是公共的,或者提供get/set方法
4. 字段的类型必须被注册的序列化器支持

下面是例子代码:

public class WordWithCount {public String word;public int count;public WordWithCount() {}public WordWithCount(String word, int count) {this.word = unt = count;}
}DataStream<WordWithCount> wordCounts = env.fromElements(new WordWithCount("hello", 1),new WordWithCount("world", 2));wordCounts.keyBy(value -> value.word);下面是scala
class WordWithCount(var word: String, var count: Int) {
//无参辅助构造器def this() {this(null, -1)}
}val input = env.fromElements(
//下面这种是直接调用的主构造器,关于scala构造器请参考我的其他文章new WordWithCount("hello", 1),new WordWithCount("world", 2)) // Case Class Data Setinput.keyBy(_.word)

下面来说说工作原理:对于你自己定义的普通类,flink首先会对你的这个类做类的检测,比如针对第一条检测是否是public 修饰的类–>Modifier.isPublic([类].getModifiers()), 检测完了之后发现符合上述四条规则,那么就会对当前类调用PojoSerializer 序列化器进行封装,下面是继承关系:

public final class PojoSerializer extends TypeSerializer {…}
public abstract class TypeSerializer implements Serializable{…}

可以看出最后用的序列化还是java的序列化. TypeSerializer是一个顶层接口,基本上所有的序列化的类都是TypeSerializer的一种实现包括PojoSerializer,下面是一些实现了TypeSerializer的类.

如果检测不符合上述四条规则,那么flink默认的序列化器是上图中的:KryoSerializer ,这个序列化器就是用的 Kryo框架.打开KryoSerializer 类发现有下面的注释:

A type serializer that serializes its type using the Kryo serialization framework ().
This serializer is intended as a fallback serializer for the cases that are not covered by the basic types, tuples, and POJOs.
Type parameters:
– The type to be serialized.
public class KryoSerializer extends TypeSerializer {…代码省略}

三.原始类型(Primitive Types )

flink支持所有scala/java 的所有原始类型:Integer String Double

四.通用类(General Class Types)

java/scala 不遵守二中所说的规范,那么scala会将此类按照统一的序列化标准进行序列化,这个序列化标准采用的序列化框架是Kryo

五.flink内置的Values类型

你需要实现org.pes.Value 接口的 read 和write方法. 和通用类(General Class Types) 相比Values接口序列化方式更加高效.
Flink提供了预定义的Value类型，与基本数据类型相对应。(ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue)。这些Value类型充当基本数据类型的可变变量:它们的值可以更改，从而允许程序员重用对象，减轻垃圾收集器的压力。
下图展示了预定义的类型和接口Value的关系:

六.Hadoop Writables

您可以使用实现org.apache.hadoop.Writable接口的类型。write()和方法中定义的序列化逻辑readFields()将用于序列化。

七.特殊类型

您可以使用特殊类型，包括 Scala 的Either、Option和Try. Java API 有自己的自定义实现Either。与 Scala 类似Either，它表示两种可能类型的值，左或右。 Either对于需要输出两种不同类型记录的错误处理或运算符非常有用。

关于java泛型类型擦除的问题’

Java 编译器在编译后丢弃了很多泛型类型信息。这在 Java中称为类型擦除。这意味着在运行时，对象的实例不再知道其泛型类型。例如，在 JVM 中DataStream和的实例DataStream看起来相同。

Flink 在准备程序执行时（调用程序的 main 方法时）需要类型信息。Flink Java API 试图以各种方式重构被丢弃的类型信息，并将其显式存储在数据集和运算符中。您可以通过检索类型Type()。该方法返回一个 的实例TypeInformation，这是 Flink 内部表示类型的方式。

类型推断有其局限性，在某些情况下需要程序员代码来“配合”。，例如 ExecutionEnvironment.fromCollection(),您可以在其中传递描述类型的参数。
但是像MapFunction<I, O>这样的泛型函数可能需要额外的类型信息。

常见的问题

• 注册子类型：如果函数签名只描述超类型，但在执行过程中实际上使用了超类型的子类型，那么让 Flink
  了解这些子类型可能会大大提高性能。为此，请.registerType(clazz)为每个子类型调用StreamExecutionEnvironmentor
  ExecutionEnvironment。
• 注册自定义序列化器： Flink 回退到Kryo来处理它自己不能透明处理的类型。并非所有类型都由 Kryo（因此也由
  Flink）无缝处理。例如，许多 Google Guava
  集合类型在默认情况下无法正常工作。解决方案是为导致问题的类型注册额外的序列化程序。这句话是说有个特殊的类型既不是Integer也不是String,而是特殊的自定义的类型比如MyInt此时我们可以针对MyInt注册一个自定义的序列化器.调用。.getConfig().addDefaultKryoSerializer(clazz,
  serializer)-许多库中提供了其他 Kryo
  序列化程序。有关使用自定义序列化程序的更多详细信息，请参阅自定义序列化程序。此功能其实不常用,一般来说我们只会自定义对象,很少会自定义类型.
• 添加类型提示：有时，尽管 Flink 无法推断出泛型类型，但用户必须传递类型提示。这通常只在 Java API
  中是必需的。类型提示部分更详细地描述了这一点。
• 手动创建TypeInformation:这对于某些 API 调用可能是必要的，因为 Java 的泛型类型擦除，Flink
  无法推断数据类型。有关详细信息，请参阅创建 TypeInformation 或 TypeSerializer 。

关于TypeInfomation

flink实现了自己的类型系统,方便做类型检查.以及扁平化对象到schem ,TypeInfomation就是所有类型的基类,TypeInfomation描述了当前类型的类型是什么,以及它序列化器用的什么,下图是是实现此接口的基类:

我们来看一个BasicTypeInfo:
BasicTypeInfo for primitive types (int, long, double, byte, …), String, Date, Void, BigInteger, and BigDecimal.(BasicTypeInfo 是java基础类型的封装类,比如 int long double byte)

public static final BasicTypeInfo<Boolean> BOOLEAN_TYPE_INFO =new BasicTypeInfo<>(Boolean.class,new Class<?>[] {},BooleanSerializer.INSTANCE,BooleanComparator.class);
注意参数:BooleanSerializer  这个序列化器其实就是实现了TypeSerializer.

上面说过: TypeSerializer是一个顶层接口,基本上所有的序列化的类都是TypeSerializer的一种实现.