无需编码：通过KSQLDB编辑Debezium同步的Kafka Topic消息并建表

1. 注册Debezium生成的原始Kafka Topic为KSQLDB流 #

Debezium输出的消息包含before/after/source等结构化字段，先将目标Topic注册为KSQLDB流，方便后续操作。如果用Avro格式，KSQL可自动推断Schema，无需手动定义字段：

-- Avro格式（需KSQL已连接Schema Registry）
CREATE STREAM original_postgres_stream
WITH (
    KAFKA_TOPIC='dbserver1.public.your_table', -- 替换为你的Debezium Topic名
    VALUE_FORMAT='AVRO'
);

-- JSON格式（需手动匹配PostgreSQL表字段）
CREATE STREAM original_postgres_stream (
    after STRUCT<
        id INT,
        name VARCHAR,
        create_time TIMESTAMP
    >,
    source STRUCT<ts_ms BIGINT>
) WITH (
    KAFKA_TOPIC='dbserver1.public.your_table',
    VALUE_FORMAT='JSON'
);

可通过DESCRIBE original_postgres_stream;验证流结构是否正确。

2. 创建带自定义列的转换流 #

基于原始流，添加自定义列（支持固定值、字段计算、内置函数生成等逻辑），并输出到新Topic：

CREATE STREAM transformed_stream
WITH (
    KAFKA_TOPIC='transformed_your_table', -- 自定义转换后的Topic名
    VALUE_FORMAT='AVRO' -- 或JSON，和原始流格式保持一致
) AS
SELECT
    after->id,
    after->name,
    after->create_time,
    -- 自定义列示例
    'default_tag' AS custom_fixed_col, -- 固定值列
    UPPER(after->name) AS custom_upper_name, -- 字符串处理列
    CURRENT_TIMESTAMP() AS custom_process_time, -- 处理时间戳
    source->ts_ms AS original_source_time -- 从Debezium元数据取原数据时间
FROM original_postgres_stream
EMIT CHANGES;

3. 基于转换流创建KSQLDB表 #

如果需要做聚合、状态查询等操作，基于转换流创建表（需指定主键，对应PostgreSQL表的主键）：

-- 方式1：直接绑定转换后的Topic
CREATE TABLE transformed_postgres_table (
    id INT PRIMARY KEY,
    name VARCHAR,
    create_time TIMESTAMP,
    custom_fixed_col VARCHAR,
    custom_upper_name VARCHAR,
    custom_process_time TIMESTAMP,
    original_source_time BIGINT
) WITH (
    KAFKA_TOPIC='transformed_your_table',
    VALUE_FORMAT='AVRO'
);

-- 方式2：通过流聚合生成表（适合需要维护最新状态的场景）
CREATE TABLE transformed_postgres_table AS
SELECT
    id,
    LATEST_BY_OFFSET(name) AS name,
    LATEST_BY_OFFSET(custom_upper_name) AS custom_upper_name
FROM transformed_stream
GROUP BY id
EMIT CHANGES;

关键注意事项 #

• 确保KSQLDB对目标Kafka Topic有读写权限。
• 若用Avro格式，需提前配置KSQLDB连接Schema Registry。
• 自定义列可灵活使用KSQL内置函数（如日期函数、数学运算、字符串处理等）实现复杂逻辑。

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