如何定义支持递归自引用的Flink POJO类型？

以下是三种经过验证的可行方案，可根据业务场景选择：

• 方案1：自定义TypeSerializer与TypeInformation（最贴合原生POJO场景，性能最优）
  Flink默认的PojoTypeExtractor无法识别递归自引用的泛型边界，会错误标记为未知泛型，直接自定义类型信息和序列化器即可绕过该限制。
  核心逻辑是在序列化器中通过自引用实现递归处理，示例代码如下：

  // 自定义Node序列化器
  public class NodeSerializer extends TypeSerializer<Node> {
      private final MapSerializer<String, Node> innerMapSerializer;
  
      public NodeSerializer() {
          // 传入当前序列化器作为Map value的序列化器，实现递归支持
          this.innerMapSerializer = new MapSerializer<>(
              StringSerializer.INSTANCE,
              this
          );
      }
  
      @Override
      public void serialize(Node node, DataOutputView out) throws IOException {
          innerMapSerializer.serialize(node.getNodeMap(), out);
      }
  
      @Override
      public Node deserialize(DataInputView in) throws IOException {
          Map<String, Node> nodeMap = innerMapSerializer.deserialize(in);
          Node res = new Node();
          res.setNodeMap(nodeMap);
          return res;
      }
  
      // 剩余方法如duplicate、copy、getLength、equals、hashcode等按Flink要求实现即可
  }
  
  // 自定义Node类型信息
  public class NodeTypeInfo extends TypeInformation<Node> {
      @Override
      public TypeSerializer<Node> createSerializer(ExecutionConfig config) {
          return new NodeSerializer();
      }
  
      // 剩余方法按需实现即可，比如isBasicType返回false，getArity返回1等
  }
  

  使用时直接在状态描述符中指定自定义类型信息即可：

  ValueStateDescriptor<Node> desc = new ValueStateDescriptor<>("tree-state", new NodeTypeInfo());
  

• 方案2：结构扁平化转通用类型存储（开发成本最低）
  不需要修改类型系统，直接将Node树序列化/反序列化为通用类型存储：

  • 可以用Jackson等序列化库将Node转为Json字符串，存储为String类型状态
  • 也可以直接序列化为byte数组，存储为ByteArray类型状态
    该方案适合QPS不高的场景，不需要适配Flink类型系统，开发速度最快，注意配置序列化工具的递归深度限制，避免栈溢出即可。

• 方案3：复用Avro递归支持能力（适配成本中等）
  你已经知道Flink对Avro的递归类型有原生支持，可以直接用Avro定义Node的递归Schema，用Avro生成的类做状态存储，业务逻辑层如果需要用原生POJO，补充一层Avro类和POJO的转换逻辑即可。

注意：如果树的深度极高，建议将序列化/反序列化的递归逻辑改为迭代实现，避免栈溢出问题。

内容的提问来源于stack exchange，提问作者Ken