flink.12 序列化

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

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

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

t3.f2t3.getField(2)

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

2.java版

DataStream> wordCounts = env.fromElements(
    new Tuple2("hello", 1),
    new Tuple2("world", 2));

wordCounts.map(new MapFunction, Integer>() {
    @Override
    public Integer map(Tuple2 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的源码:

// --------------------------------------------------------------
//  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;


@Public
public class Tuple2 extends Tuple {

    private static final long serialVersionUID = 1L;

    
    public T0 f0;
    
    public T1 f1;

    
    public Tuple2() {}

    
    public Tuple2(T0 f0, T1 f1) {
        this.f0 = f0;
        this.f1 = f1;
    }

    @Override
    public int getArity() {
        return 2;
    }

    @Override
    @SuppressWarnings("unchecked")
    public  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  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));
        }
    }

    
    public void setFields(T0 f0, T1 f1) {
        this.f0 = f0;
        this.f1 = f1;
    }

    
    public Tuple2 swap() {
        return new Tuple2(f1, f0);
    }

    // -------------------------------------------------------------------------------------------------
    // standard utilities
    // -------------------------------------------------------------------------------------------------

    
    @Override
    public String toString() {
        return "("
                + StringUtils.arrayAwareToString(this.f0)
                + ","
                + StringUtils.arrayAwareToString(this.f1)
                + ")";
    }

    
    @Override
    public 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;
    }

    @Override
    public int hashCode() {
        int result = f0 != null ? f0.hashCode() : 0;
        result = 31 * result + (f1 != null ? f1.hashCode() : 0);
        return result;
    }

    
    @Override
    @SuppressWarnings("unchecked")
    public Tuple2 copy() {
        return new Tuple2<>(this.f0, this.f1);
    }

    
    public static  Tuple2 of(T0 f0, T1 f1) {
        return new Tuple2<>(f0, f1);
    }
}

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

普通类有以下要求:

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

下面是例子代码:

public class WordWithCount {

    public String word;
    public int count;

    public WordWithCount() {}

    public WordWithCount(String word, int count) {
        this.word = word;
        this.count = count;
    }
}

DataStream 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 Set

input.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 (https://github.com/EsotericSoftware/kryo).
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 {…代码省略}

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

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

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

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

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

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

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

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

注册子类型：如果函数签名只描述超类型，但在执行过程中实际上使用了超类型的子类型，那么让 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_TYPE_INFO =
        new BasicTypeInfo<>(
                Boolean.class,
                new Class[] {},
                BooleanSerializer.INSTANCE,
                BooleanComparator.class);
注意参数:BooleanSerializer  这个序列化器其实就是实现了TypeSerializer.

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