flink15里DataStream<Raw> 转table 为什么只有一个字段，有遇到过的吗？

flink15里DataStream 转table 为什么只有一个字段，有遇到过的吗 f0 RAW('org.apache.flink.types.Row', '...')

展开

收起

真的很搞笑 2023-05-14 21:04:47 134 版权

阿里云 AI 助理回答

在使用 Flink 1.15 的过程中，如果将 DataStream 转换为 Table 时发现生成的表只有一个字段（例如 f0 RAW('org.apache.flink.types.Row', '...')），这通常是由于类型推导或转换配置不正确导致的。以下是可能的原因及解决方法：

原因分析

未正确指定 Schema 或类型信息
- 在将 DataStream 转换为 Table 时，如果没有明确指定字段名和字段类型，Flink 默认会将整个 Row 对象视为一个字段，并将其类型标记为 RAW 类型。
- 这种情况通常发生在使用 toTable 或 fromDataStream 方法时，未通过 Schema 或 @DataTypeHint 注解提供明确的字段定义。
POJO 类字段推导问题
- 如果 DataStream 中的数据是自定义的 POJO 类对象，而该类缺少有参构造函数，Flink 会按照字段名的字典序重新排列字段顺序，可能导致字段映射错误或无法正确推导字段类型。
- 此外，如果 POJO 类的字段没有正确的 getter 和 setter 方法，也可能导致字段无法被正确识别。
依赖版本冲突
- 如果使用的 Flink 版本与连接器或其他依赖库的版本不匹配，可能会导致类型推导失败，进而生成默认的 RAW 类型字段。
未启用宽容模式
- 在某些场景下（如同步到 Hologres 表时），如果未启用字段类型宽容模式（enableTypeNormalization），可能会导致字段类型不兼容，从而生成默认的 RAW 类型字段。

解决方法

1. 明确指定字段和类型

在将 DataStream 转换为 Table 时，建议显式指定字段名和字段类型。可以通过以下方式实现：

使用 Schema 定义字段

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

DataStream<Row> dataStream = env.fromElements(
  Row.of(1, "Alice"),
  Row.of(2, "Bob")
);

Table table = tEnv.fromDataStream(
  dataStream,
  Schema.newBuilder()
      .column("id", DataTypes.INT())
      .column("name", DataTypes.STRING())
      .build()
);

使用 @DataTypeHint 注解 如果数据流中的对象是自定义类，可以使用 @DataTypeHint 注解来指定字段类型：

public static class MyPojo {
  public int id;
  @DataTypeHint("STRING")
  public String name;
}

DataStream<MyPojo> dataStream = env.fromElements(
  new MyPojo(1, "Alice"),
  new MyPojo(2, "Bob")
);

Table table = tEnv.fromDataStream(dataStream);

2. 检查 POJO 类定义

确保 POJO 类满足以下条件： - 提供无参构造函数和有参构造函数。 - 为每个字段提供标准的 getter 和 setter 方法。 - 字段名与 SQL 查询中定义的字段名一致。

示例：

public static class MyPojo {
    public int id;
    public String name;

    public MyPojo() {}

    public MyPojo(int id, String name) {
        this.id = id;
        this.name = name;
    }

    public int getId() { return id; }
    public void setId(int id) { this.id = id; }

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

3. 检查依赖版本

确保使用的 Flink 版本与连接器版本兼容。例如，如果您使用的是 Flink 1.15，则需要选择对应版本的连接器 JAR 文件。

4. 启用宽容模式

如果目标表支持字段类型宽容模式（如 Hologres 表），可以在创建表时启用 enableTypeNormalization 参数：

CREATE TABLE IF NOT EXISTS holo_table (
    id BIGINT,
    name STRING
) WITH (
    'connector' = 'hologres',
    'enableTypeNormalization' = 'true'
);