2.Flink序列化机制_flink pojo 序列化器

概述

  在两个进程进行远程通信时，它们需要将各种类型的数据以二进制序列的形式在网络上传输，数据发送方需要将对象转换为字节序列，进行序列化，而接收方则将字节序列恢复为各种对象，进行反序列化。对象的序列化有两个主要用途：一是将对象的字节序列永久保存到硬盘上，通常存放在文件中；二是在网络上传输对象的字节序列。序列化的好处包括减少数据在内存和硬盘中的占用空间，减少网络传输开销，精确推算内存使用情况，降低垃圾回收的频率。

注意

  Flink序列化机制负责在节点之间传输数据时对数据对象进行序列化和反序列化，确保数据的正确性和一致性。Flink提供了多种序列化器，包括Kryo、Avro和Java序列化器等，大多数情况下，用户不用担心flink的序列化框架，Flink会通过TypeInfomation在数据处理之前推断数据类型，进而使用对应的序列化器，例如：针对标准类型（int,double,long,string）直接由Flink自带的序列化器处理，其他类型默认会交给Kryo处理。但是对于Kryo仍然无法处理的类型，可以采取以下两种解决方案：

强制使用Avro替代Kryo序列化

//设置flink序列化方式为avro
env.getConfig().enableForceAvro();
1
2

自定义注册Kryo序列化

//注册kryo 自定义序列化器
env.getConfig().registerTypeWithKryoSerializer(Class<?> type, Class<? extends Serializer> serializerClass)
1
2

实践

自定义pojo

public class SerializerTest {
    public static void main(String[] args) throws Exception {
        //准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Flink会自动识别数据类型，当符合要求时，会优先使用其定制的序列化框架，否则，将默认使用Kryo完成序列化
        // 关闭使用 kryo
        env.getConfig().disableGenericTypes();
        //source ,socket:1,zs,18
        DataStreamSource<String> ds = env.socketTextStream("localhost", 9999);

        ds.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String s) throws Exception {
                String[] arr = s.split(",");
                return new Student(Integer.valueOf(arr[0]), arr[1], Integer.valueOf(arr[2]));
            }
        }).print();
        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

public class Student {
    private Integer id;
    private String name;
    private Integer age;

    public Student() {
    }

    public Student(Integer id, String name, Integer age) {
        this.id = id;
        this.name = name;
        this.age = age;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public void setName(String name) {
        this.name = name;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public Integer getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    public Integer getAge() {
        return age;
    }

    @Override
    public String toString() {
        return "Student{" +
                "id=" + id +
                ", name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

没有序例化时，也是可以正常使用的。

自定义pojo kryo序例化

public class SerializerTestT {
    public static void main(String[] args) throws Exception {
        //准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().registerTypeWithKryoSerializer(Student.class, StudentSerializer.class);

        //source ,socket:1,zs,18
        DataStreamSource<String> ds = env.socketTextStream("localhost", 9999);

        ds.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String s) throws Exception {
                String[] arr = s.split(",");
                return new Student(Integer.valueOf(arr[0]), arr[1], Integer.valueOf(arr[2]));
            }
        }).print();
        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

/**
 * 顺序不要变
 */
public class StudentSerializer extends Serializer {

    @Override
    public void write(Kryo kryo, Output output, Object o) {
        Student student = (Student) o;
        output.writeInt(student.getId());
        output.writeString(student.getName());
        output.writeInt(student.getAge());

    }

    @Override
    public Object read(Kryo kryo, Input input, Class aClass) {
        Student student = new Student();
        student.setId(input.readInt());
        student.setName(input.readString());
        student.setAge(input.readInt());
        return student;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

结束

Flink序列化机制