Flink CDC系列之：Kafka表结构信息管理类TableSchemaInfo

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

TableSchemaInfo：Kafka表结构信息管理类详解

该类为Flink CDC中用于管理表结构信息的核心组件，主要负责维护表的元数据、字段访问器以及序列化机制，并提供关键的数据格式转换能力。作为支持Kafka JSON/CSV序列化功能的重要基础模块，其设计兼顾了灵活性与性能。

类定义

public class TableSchemaInfo

此类整合了表标识、Schema定义、主键索引映射、字段提取逻辑和序列化策略，是实现高效数据写入与解析的关键支撑结构。

核心字段说明

• tableId：表示当前表的唯一标识（TableId类型）
• schema：描述表的完整结构（Schema对象）
• primaryKeyColumnIndexes：记录所有主键字段在列列表中的位置索引（List<Integer>）
• fieldGetters：字段值提取器集合，用于从原始记录中获取各列数据（List<RecordData.FieldGetter>）
• serializationSchema：行数据的序列化处理器（SerializationSchema<RowData>）

[此处为图片1]

构造函数实现

public TableSchemaInfo(
    TableId tableId,
    Schema schema,
    SerializationSchema<RowData> serializationSchema,
    ZoneId zoneId) {
  this.tableId = tableId;
  this.schema = schema;
  this.serializationSchema = serializationSchema;
  this.fieldGetters = createFieldGetters(schema, zoneId);
  this.primaryKeyColumnIndexes = new ArrayList<>();

  // 遍历主键名称，查找对应列索引并保存
  for (String primaryKey : schema.primaryKeys()) {
    for (int columnIndex = 0; columnIndex < schema.getColumnCount(); columnIndex++) {
      if (schema.getColumns().get(columnIndex).getName().equals(primaryKey)) {
        primaryKeyColumnIndexes.add(columnIndex);
        break;
      }
    }
  }
}

主键索引构建流程

1. 遍历Schema中声明的所有主键名称
2. 对每个主键名，在列集合中搜索匹配的列名
3. 将匹配到的列索引存入primaryKeyColumnIndexes列表
4. 最终形成主键到物理列位置的映射关系

主要方法解析

getRowDataFromRecordData —— 数据转换入口

public RowData getRowDataFromRecordData(RecordData recordData, boolean primaryKeyOnly) {
  if (primaryKeyOnly) {
    return createPrimaryKeyRowData(recordData);   // 仅输出主键字段
  } else {
    return createFullRowData(recordData);         // 输出全部字段
  }
}

根据参数决定是否只提取主键部分，从而支持不同场景下的数据投递需求。

createPrimaryKeyRowData —— 主键数据构造

private RowData createPrimaryKeyRowData(RecordData recordData) {
  // 创建一个包含表名 + 所有主键值的GenericRowData实例
  GenericRowData genericRowData = new GenericRowData(primaryKeyColumnIndexes.size() + 1);

  // 第一个字段设置为表标识
  genericRowData.setField(0, StringData.fromString(tableId.toString()));

  // 接下来的字段依次填充各个主键的实际值
  for (int i = 0; i < primaryKeyColumnIndexes.size(); i++) {
    int columnIndex = primaryKeyColumnIndexes.get(i);
    RecordData.FieldGetter fieldGetter = fieldGetters.get(columnIndex);
    genericRowData.setField(i + 1, fieldGetter.getFieldOrNull(recordData));
  }

  return genericRowData;
}

生成的结果结构如下：

• 字段0：表的全路径标识（字符串形式）
• 字段1~n：依次为主键列的值

[此处为图片1]

在数据处理流程中，为了实现不同数据格式之间的转换，通常需要将原始记录数据（RecordData）转化为通用的行数据结构（RowData）。以下是相关方法与逻辑的详细说明。

createFullRowData - 构建完整行数据

private RowData createFullRowData(RecordData recordData) {
    // 初始化一个包含所有字段的GenericRowData
    GenericRowData genericRowData = new GenericRowData(recordData.getArity());
    // 遍历并设置每个字段的值
    for (int i = 0; i < recordData.getArity(); i++) {
        genericRowData.setField(i, fieldGetters.get(i).getFieldOrNull(recordData));
    }
    return genericRowData;
}

该方法的主要功能是基于输入的recordData生成一个完整的RowData对象，确保所有字段均被正确提取和封装。

核心组成要素包括：

• GenericRowData：来自 Apache Flink 的通用行数据容器，用于存储一行中的多个字段值。
• recordData.getArity()：返回当前记录所包含的字段总数。
• fieldGetters：一个按字段顺序排列的获取器列表，每个元素负责从recordData中安全地提取对应位置的字段值，支持空值处理。

[此处为图片1]

执行流程如下：

1. 根据字段数量初始化一个新的GenericRowData实例。
2. 通过循环遍历每一个字段索引。
3. 使用对应的fieldGetter从源数据中读取值，并写入目标行结构。
4. 最终返回已填充数据的RowData对象。

此模式广泛应用于流式或批处理任务中，保证数据在格式映射过程中不丢失任何字段信息。

createFieldGetters - 字段获取器集合构建

private static List<RecordData.FieldGetter> createFieldGetters(Schema schema, ZoneId zoneId) {
    List<RecordData.FieldGetter> fieldGetters = new ArrayList<>(schema.getColumns().size());
    for (int i = 0; i < schema.getColumns().size(); i++) {
        fieldGetters.add(createFieldGetter(schema.getColumns().get(i).getType(), i, zoneId));
    }
    return fieldGetters;
}

该方法的作用是依据给定的数据表结构（Schema），为每一列创建一个类型匹配的字段值提取器。这些提取器后续可用于高效、类型安全地访问记录中的各个字段。

createFieldGetter - 按类型生成字段获取器

这是整个机制中最关键的部分，针对不同的数据类型动态生成相应的取值逻辑：

字符串类型处理

对于字符型字段（如 CHAR 和 VARCHAR），其值会被转换为二进制字符串形式：

case CHAR:
    case VARCHAR:
        fieldGetter = record -> BinaryStringData.fromString(record.getString(fieldPos).toString());

基础数值类型映射

以下类型直接调用 RecordData 提供的原生访问方法：

case BOOLEAN: fieldGetter = record -> record.getBoolean(fieldPos);
case TINYINT: fieldGetter = record -> record.getByte(fieldPos);
case SMALLINT: fieldGetter = record -> record.getShort(fieldPos);
case INTEGER: fieldGetter = record -> record.getInt(fieldPos);
case BIGINT: fieldGetter = record -> record.getLong(fieldPos);
case FLOAT:   fieldGetter = record -> record.getFloat(fieldPos);
case DOUBLE:  fieldGetter = record -> record.getDouble(fieldPos);

高精度小数类型（DECIMAL）支持

针对 DECIMAL 类型，需额外考虑精度与标度参数，通过工具方法提取后构造对应的 DecimalData 对象：

case DECIMAL:
    final int decimalPrecision = getPrecision(fieldType);
    final int decimalScale = getScale(fieldType);
    fieldGetter = record -> DecimalData.fromBigDecimal(
        record.getDecimal(fieldPos, decimalPrecision, decimalScale).toBigDecimal()
    );

这种设计保障了在处理金融级精度数据时的准确性与一致性。

最终输出的数据结构形式为：

RowData[Col1, Col2, Col3, ...]

该结构代表了一个标准化的行数据表示方式，适用于下游系统的进一步处理与分析。

decimalPrecision, decimalScale);

record.getDecimal(fieldPos, decimalPrecision, decimalScale).toBigDecimal(),

日期与时间类型处理

case DATE:
fieldGetter = record -> (int) record.getDate(fieldPos).toEpochDay(); // 转换为从纪元起的天数

case TIME_WITHOUT_TIME_ZONE:
fieldGetter = record -> (int) record.getTime(fieldPos).toMillisOfDay(); // 转换为当日毫秒数

case TIMESTAMP_WITHOUT_TIME_ZONE:
fieldGetter =
record ->
TimestampData.fromTimestamp(
record.getTimestamp(fieldPos, getPrecision(fieldType)).toTimestamp());

case TIMESTAMP_WITH_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
fieldGetter =
record ->
TimestampData.fromInstant(
record.getLocalZonedTimestampData(
fieldPos,
DataTypeChecks.getPrecision(fieldType)
).toInstant());

[此处为图片1]

二进制数据类型的处理

case BINARY:
case VARBINARY:
fieldGetter = record -> record.getBinary(fieldPos);

空值安全的字段获取器包装

当字段非空时，直接返回原始获取器：

if (!fieldType.isNullable()) {
    return fieldGetter;
}

对于可为空的字段，则添加空值判断逻辑：

return row -> {
    if (row.isNullAt(fieldPos)) {
        return null;
    }
    return fieldGetter.getFieldOrNull(row);
};

[此处为图片2]

设计模式结合使用：访问者模式与策略模式

访问者模式的应用

通过定义统一接口实现对不同数据字段的访问：

interface FieldGetter {
    Object getFieldOrNull(RecordData record);
}

针对各类数据类型构建具体实现：

FieldGetter stringGetter = record -> ...;
FieldGetter intGetter = record -> ...;
FieldGetter timestampGetter = record -> ...;

策略模式的应用

根据字段类型动态选择对应的获取策略：

switch (fieldType.getTypeRoot()) {
    case CHAR: return stringGetter;
    case INTEGER: return intGetter;
    case TIMESTAMP: return timestampGetter;
    // 其他类型... }

实际应用场景示例：JSON序列化过程

在 JSON 序列化组件中的实现方式如下：

public class JsonSerializationSchema {
    private final Map<TableId, TableSchemaInfo> jsonSerializers;

    public byte[] serialize(Event event) {
        DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
        RecordData recordData = getRecordData(dataChangeEvent);
        TableSchemaInfo tableSchemaInfo = jsonSerializers.get(dataChangeEvent.tableId());

        // 提取仅含主键的 RowData 并序列化为 JSON
        RowData rowData = tableSchemaInfo.getRowDataFromRecordData(recordData, true);
        return tableSchemaInfo.getSerializationSchema().serialize(rowData);
    }
}

数据格式转换流程说明

从 Flink CDC 内部使用的 RecordData 格式开始，经过一系列类型适配和空值处理后，最终输出可用于外部系统的标准结构（如 JSON）。整个流程涵盖了基本类型转换、时区处理、二进制支持以及可空性封装。

TableSchemaInfo 类作为表结构管理与数据转换的核心组件，在 Flink CDC 到 Kafka 的数据流转中扮演着关键角色。它实现了从 Flink Table 内部格式到序列化所需格式的高效、类型安全的转换。

该类通过以下机制保障数据处理的可靠性与性能：

• 类型安全转换：针对不同数据类型提供专用的数据提取器，确保类型匹配与正确解析。
• 性能优化策略：预先计算字段获取器及主键索引位置，减少运行时开销，提升访问效率。
• 空值安全性：内置对可为空字段的判空处理，避免因 null 值引发异常。
• 灵活输出支持：兼容主键模式与全字段模式两种输出方式，适配多种业务场景需求。
• 良好的扩展性：设计上支持新增数据类型和自定义输出模式，便于后续功能拓展。
• 统一接口封装：将复杂的类型映射与转换逻辑隐藏于简洁 API 之后，降低使用复杂度。

在实际数据流中，RowData 作为 Flink Table 的内部数据表示形式，首先由 TableSchemaInfo.getRowDataFromRecordData() 方法进行解析与提取 [此处为图片1]；随后，经由 SerializationSchema.serialize() 转换为字节流（如 JSON 或 CSV 格式），最终写入 Kafka。

综上所述，TableSchemaInfo 不仅是连接 Flink CDC 与外部消息系统的重要桥梁，更通过其结构化设计保障了数据序列化过程中的高效性、安全性与可维护性。

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Schema tables Table Esch Info