flink waljson

### 什么是 WalJson 及其在 Flink 中的应用

WalJson 是一种用于表示逻辑复制槽输出插件的 JSON 格式的协议，在 PostgreSQL 数据库中广泛应用于变更数据捕获 (CDC) 场景。当与 Apache Flink 结合使用时，可以通过 Debezium 或其他 CDC 工具连接到数据库，并利用 WalJson 来捕捉表结构变化和记录更新事件。

为了使 Flink 能够处理来自 PostgreSQL 的增量更改日志，通常会采用如下方式之一：

#### 方法一：通过 Debezium Connector 实现

Debezium 是一款开源项目，旨在简化从各种数据库系统获取变更日志的过程。对于 PostgreSQL 数据源来说，可以部署 Debezium 并将其配置为以 WalJson 格式发送消息给 Kafka 主题；之后再由 Flink 应用程序订阅这些主题来进行进一步的数据转换或存储操作。

安装并运行 Debezium Postgres Connector 后，可以在 `connect-debezium-postgres-source` 文件夹下找到样例配置文件 `postgresql.properties`[^1]。

name=mysql-connector
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=localhost
database.port=5432
database.user=myuser
database.password=mypassword
database.dbname=mydb
plugin.name=wal2json
slot.name=my_slot_name
publication.autocreate.mode=eager
table.include.list=public.my_table
topic.prefix=my_topic_prefix

此配置指定了要监听哪个数据库实例 (`localhost`) 和特定模式下的哪些表格 (`my_table`) ，并将所有变更发布至名为 `my_topic_prefix.public.my_table` 的 Kafka 主题上。

接着就可以创建一个简单的 Flink 流应用程序来消费这个 Kafka 主题中的数据了:

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkPostgresCdcExample {
    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        var kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("kafka-broker:9092")
                .setTopics("my_topic_prefix.public.my_table")
                .setGroupId("test-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
           .print(); // Replace with actual processing logic.

        env.execute("Flink PostgreSQL CDC Example");
    }
}

这段 Java 代码展示了如何设置 Kafka 源作为输入端点，并指定相应的反序列化器以便于后续解析 Json 字符串格式的消息体内容[^2]。

#### 方法二：直接读取 WAL 日志

另一种更底层的方法是让 Flink 直接访问 PostgreSQL 的 Write-Ahead Logging(WAL)，但这涉及到复杂的权限管理和网络通信细节，因此一般不推荐这样做除非有特殊需求。而且官方也不支持这种方式，因为这可能会破坏数据库的一致性和稳定性。

---