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

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

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

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

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

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

  2. <dependency>
  3.     <groupId>com.twitter</groupId>
  4.     <artifactId>chill-thrift</artifactId>
  5.     <version>0.7.6</version>
  6.     <!-- exclusions for dependency conversion -->
  7.     <exclusions>
  8.         <exclusion>
  9.             <groupId>com.esotericsoftware.kryo</groupId>
  10.             <artifactId>kryo</artifactId>
  11.         </exclusion>
  12.     </exclusions>
  13. </dependency>
  14. <!-- libthrift is required by chill-thrift -->
  15. <dependency>
  16.     <groupId>org.apache.thrift</groupId>
  17.     <artifactId>libthrift</artifactId>
  18.     <version>0.11.0</version>
  19.     <exclusions>
  20.         <exclusion>
  21.             <groupId>javax.servlet</groupId>
  22.             <artifactId>servlet-api</artifactId>
  23.         </exclusion>
  24.         <exclusion>
  25.             <groupId>org.apache.httpcomponents</groupId>
  26.             <artifactId>httpclient</artifactId>
  27.         </exclusion>
  28.     </exclusions>
  29. </dependency>

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

  2. <dependency>
  3.     <groupId>com.twitter</groupId>
  4.     <artifactId>chill-protobuf</artifactId>
  5.     <version>0.7.6</version>
  6.     <!-- exclusions for dependency conversion -->
  7.     <exclusions>
  8.         <exclusion>
  9.             <groupId>com.esotericsoftware.kryo</groupId>
  10.             <artifactId>kryo</artifactId>
  11.         </exclusion>
  12.     </exclusions>
  13. </dependency>
  14. <!-- We need protobuf for chill-protobuf -->
  15. <dependency>
  16.     <groupId>com.google.protobuf</groupId>
  17.     <artifactId>protobuf-java</artifactId>
  18.     <version>3.7.0</version>
  19. </dependency>

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

使用 Kryo JavaSerializer 的问题

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

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

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