• 为你的 Flink 程序注册自定义序列化器
    • 使用 Kryo JavaSerializer 的问题

    为你的 Flink 程序注册自定义序列化器

    如果在 Flink 程序中使用了 Flink 类型序列化器无法进行序列化的用户自定义类型,Flink 会回退到通用的 Kryo 序列化器。可以使用 Kryo 注册自己的序列化器或序列化系统,比如 Google Protobuf 或 Apache Thrift。使用方法是在 Flink 程序中的 ExecutionConfig 注册类类型以及序列化器。

    1. final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    2. // 为类型注册序列化器类
    3. env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, MyCustomSerializer.class);
    4. // 为类型注册序列化器实例
    5. MySerializer mySerializer = new MySerializer();
    6. env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, mySerializer);

    需要确保你的自定义序列化器继承了 Kryo 的序列化器类。对于 Google Protobuf 或 Apache Thrift,这一点已经为你做好了:

    1. final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    2. // 使用 Kryo 注册 Google Protobuf 序列化器
    3. env.getConfig().registerTypeWithKryoSerializer(MyCustomType.class, ProtobufSerializer.class);
    4. // 注册 Apache Thrift 序列化器为标准序列化器
    5. // TBaseSerializer 需要初始化为默认的 kryo 序列化器
    6. env.getConfig().addDefaultKryoSerializer(MyCustomType.class, TBaseSerializer.class);

    为了使上面的例子正常工作,需要在 Maven 项目文件中(pom.xml)包含必要的依赖。为 Apache Thrift 添加以下依赖:

    1. <dependency>
    2. <groupId>com.twitter</groupId>
    3. <artifactId>chill-thrift</artifactId>
    4. <version>0.5.2</version>
    5. </dependency>
    6. <!-- libthrift is required by chill-thrift -->
    7. <dependency>
    8. <groupId>org.apache.thrift</groupId>
    9. <artifactId>libthrift</artifactId>
    10. <version>0.6.1</version>
    11. <exclusions>
    12. <exclusion>
    13. <groupId>javax.servlet</groupId>
    14. <artifactId>servlet-api</artifactId>
    15. </exclusion>
    16. <exclusion>
    17. <groupId>org.apache.httpcomponents</groupId>
    18. <artifactId>httpclient</artifactId>
    19. </exclusion>
    20. </exclusions>
    21. </dependency>

    对于 Google Protobuf 需要添加以下 Maven 依赖:

    1. <dependency>
    2. <groupId>com.twitter</groupId>
    3. <artifactId>chill-protobuf</artifactId>
    4. <version>0.5.2</version>
    5. </dependency>
    6. <!-- We need protobuf for chill-protobuf -->
    7. <dependency>
    8. <groupId>com.google.protobuf</groupId>
    9. <artifactId>protobuf-java</artifactId>
    10. <version>2.5.0</version>
    11. </dependency>

    请根据需要调整两个依赖库的版本。

    使用 Kryo JavaSerializer 的问题

    如果你为自定义类型注册 Kryo 的 JavaSerializer,即使你提交的 jar 中包含了自定义类型的类,也可能会遇到 ClassNotFoundException 异常。这是由于 Kryo JavaSerializer 的一个已知问题,它可能使用了错误的类加载器。

    在这种情况下,你应该使用 org.apache.flink.api.java.typeutils.runtime.kryo.JavaSerializer 来解决这个问题。这个类是在 Flink 中对 JavaSerializer 的重新实现,可以确保使用用户代码的类加载器。

    更多细节可以参考 FLINK-6025。