开发者社区大数据与机器学习实时计算 Flink 正文

Flink CDC 有没有DEMO看下依赖包？

展开

收起

真的很搞笑 2023-09-19 20:29:27 60 版权

阿里云 AI 助理回答

Flink CDC 提供了多种使用场景的 DEMO 和依赖包配置方法，以下为具体的参考内容和操作步骤：

1. Flink CDC 的依赖包配置

在使用 Flink CDC 时，您需要根据作业类型（SQL 或 DataStream）选择合适的依赖包。以下是两种常见的依赖包形式：

flink-sql-connector-xxx.jar
- 这是一个 FAT JAR，包含了连接器的所有依赖，并将第三方库通过 Shade 打包。
- 适用于 SQL 作业，只需将该 JAR 包放入 lib 目录即可。
flink-connector-xxx.jar
- 仅包含连接器的核心代码，不包含其依赖。
- 适用于 DataStream 作业，需要手动管理依赖并处理冲突。

示例：MySQL CDC 的 Maven 依赖

如果您使用的是 DataStream 作业，可以在 pom.xml 文件中添加以下依赖：

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>${vvr.version}</version>
</dependency>

如果需要排除冲突依赖，可以使用 exclusions 标签：

<dependency>
    <groupId>foo</groupId>
    <artifactId>bar</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
        </exclusion>
    </exclusions>
</dependency>

2. Flink CDC 的 DEMO 示例

以下是一些常见的 Flink CDC 使用场景及其 DEMO 配置：

(1) MySQL 到 Hologres 的实时同步

这是一个典型的 Flink CDC 数据同步案例，展示了如何将 MySQL 数据实时同步到 Hologres。

操作步骤： 1. 准备测试数据： - 下载 tpc_ds.sql、user_db1.sql 等文件并导入 RDS MySQL。 - 在 Hologres 中创建目标数据库 my_user。 2. 配置白名单： - 获取 Flink 工作空间的 VPC 网段信息。 - 将网段信息添加到 MySQL 和 Hologres 的 IP 白名单中。 3. 编写 SQL 作业：

CREATE TABLE mysql_source (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<yourHostname>',
    'port' = '<yourPort>',
    'username' = '<yourUserName>',
    'password' = '<yourPassWord>',
    'database-name' = 'tpc_ds',
    'table-name' = 'orders'
);

CREATE TABLE hologres_sink (
    id INT,
    name STRING
) WITH (
    'connector' = 'hologres',
    'url' = '<yourHologresUrl>',
    'tablename' = 'public.orders',
    'username' = '<yourUserName>',
    'password' = '<yourPassWord>'
);

INSERT INTO hologres_sink SELECT * FROM mysql_source;

(2) PolarDB PostgreSQL 的 CDC 同步

PolarDB PostgreSQL 的 Flink CDC 连接器支持全量快照和增量变更数据的读取。

操作步骤： 1. 前提准备： - 购买 PolarDB PostgreSQL 集群并创建高权限账户。 - 创建源表 shipments 和目标表 shipments_sink。 2. 编写 SQL 作业：

CREATE TEMPORARY TABLE shipments (
    shipment_id INT,
    order_id INT,
    origin STRING,
    destination STRING,
    is_arrived BOOLEAN,
    order_time TIMESTAMP,
    PRIMARY KEY (shipment_id) NOT ENFORCED
) WITH (
    'connector' = 'polardbo-cdc',
    'hostname' = '<yourHostname>',
    'port' = '<yourPort>',
    'username' = '<yourUserName>',
    'password' = '<yourPassWord>',
    'database-name' = 'flink_source',
    'schema-name' = 'public',
    'table-name' = 'shipments',
    'decoding.plugin.name' = 'pgoutput',
    'slot.name' = 'flink'
);

CREATE TEMPORARY TABLE shipments_sink (
    shipment_id INT,
    order_id INT,
    origin STRING,
    destination STRING,
    is_arrived BOOLEAN,
    order_time TIMESTAMP
) WITH (
    'connector' = 'jdbc',
    'url' = '<yourJdbcUrl>',
    'table-name' = 'shipments_sink'
);

INSERT INTO shipments_sink SELECT * FROM shipments;