Flink CDC 纯写sql的方式可以实现1对多关联，把mysql数据写入es么？

使用Flink CDC纯SQL方式实现MySQL到Elasticsearch的1对多关联写入

要实现将MySQL中的数据通过Flink CDC以纯SQL的方式写入Elasticsearch，并生成类似以下结构的嵌套JSON文档：

{
    "id": 1,
    "name": "张三",
    "scoreList": [
        {
            "id": 1,
            "name": "语文",
            "score": 90
        },
        {
            "id": 2,
            "name": "数学",
            "score": 80
        }
    ]
}

以下是详细的实现步骤和配置说明。

1. 数据源表设计

假设MySQL中有两张表： - 用户表（user）：存储用户基本信息。

CREATE TABLE user (
    id INT PRIMARY KEY,
    name STRING
);

• 成绩表（score）：存储用户的各科成绩，与用户表通过user_id字段关联。

  CREATE TABLE score (
    id INT PRIMARY KEY,
    user_id INT,
    subject STRING,
    score DOUBLE
  );
  

2. 创建Flink SQL作业

2.1 配置MySQL CDC源表

使用Flink CDC连接器读取MySQL中的user表和score表。需要为每个表指定server-id以避免冲突。

-- 用户表CDC源
CREATE TABLE mysql_user (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<MySQL主机名>',
    'port' = '<MySQL端口>',
    'username' = '<用户名>',
    'password' = '<密码>',
    'database-name' = '<数据库名>',
    'table-name' = 'user',
    'server-id' = '5400-5401'
);

-- 成绩表CDC源
CREATE TABLE mysql_score (
    id INT,
    user_id INT,
    subject STRING,
    score DOUBLE,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<MySQL主机名>',
    'port' = '<MySQL端口>',
    'username' = '<用户名>',
    'password' = '<密码>',
    'database-name' = '<数据库名>',
    'table-name' = 'score',
    'server-id' = '5402-5403'
);

2.2 创建Elasticsearch结果表

定义Elasticsearch的目标表结构，支持嵌套JSON格式。使用ARRAY类型表示scoreList字段。

CREATE TABLE es_sink (
    id INT,
    name STRING,
    scoreList ARRAY<ROW<id INT, name STRING, score DOUBLE>>
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = '<Elasticsearch地址>',
    'index' = '<索引名称>',
    'document-type' = '_doc',
    'username' = '<用户名>',
    'password' = '<密码>',
    'format' = 'json'
);

2.3 数据转换逻辑

通过SQL实现1对多关联，并将score表的数据聚合为数组，嵌套到scoreList字段中。

INSERT INTO es_sink
SELECT
    u.id AS id,
    u.name AS name,
    ARRAY_AGG(
        ROW(s.id, s.subject, s.score)
    ) AS scoreList
FROM
    mysql_user AS u
LEFT JOIN
    mysql_score AS s
ON
    u.id = s.user_id
GROUP BY
    u.id, u.name;

3. 关键点解析

3.1 MySQL CDC配置

• Server ID：确保每个CDC源表的server-id范围不重叠，避免Binlog消费冲突。
• 增量快照框架：如果开启增量快照框架且并行度大于1，需为server-id指定范围，范围大小应不小于并行度。

3.2 Elasticsearch嵌套结构

• ARRAY类型：Flink SQL支持ARRAY类型，用于表示嵌套的JSON数组。
• ROW类型：ROW类型用于定义数组中每个元素的结构。

3.3 数据聚合

• ARRAY_AGG函数：用于将多行数据聚合为一个数组，适合处理1对多关系。
• GROUP BY：按用户ID和姓名分组，确保每个用户只生成一条记录。

4. 注意事项

• 版本要求：确保Flink版本支持MySQL CDC和Elasticsearch连接器，推荐使用VVR 6.x及以上版本。
• 网络连通性：确保Flink集群能够访问MySQL和Elasticsearch实例。
• Schema变更：如果MySQL表结构发生变更，需重新部署Flink作业以同步最新Schema。
• 性能优化：对于大规模数据，建议调整sink.parallelism和batchSize参数以提升写入性能。

5. 总结

通过上述步骤，您可以使用Flink CDC纯SQL方式实现MySQL到Elasticsearch的1对多关联写入，并生成嵌套JSON文档。关键在于正确配置MySQL CDC源表、定义Elasticsearch目标表结构，以及利用ARRAY_AGG函数完成数据聚合。