Flink CDC怎么配置登录？

Flink CDC 的登录配置主要涉及以下几个步骤，具体操作如下：

1. 配置 PolarDB 或 MySQL 数据库

在使用 Flink CDC 连接数据库之前，需要确保数据库的相关参数已正确配置： - PolarDB 配置： - 将 wal_level 参数设置为 logical。 - 执行以下 SQL 命令以启用表的全量复制： sql ALTER TABLE schema.table REPLICA IDENTITY FULL; - 确保 max_wal_senders 和 max_replication_slots 参数值大于当前已使用的数量与 Flink 作业所需的 slot 数量。 - 使用高权限账号，该账号需具备 LOGIN 和 REPLICATION 权限，并对订阅表具有 SELECT 权限。 - 使用 PolarDB 集群的主地址进行连接。

• MySQL 配置：
  • 确保 MySQL 版本为 5.7 及以上。
  • 创建测试数据库和账号，并授权读写权限。例如：

  CREATE DATABASE test_cdc;
  GRANT ALL PRIVILEGES ON test_cdc.* TO 'test_user'@'%' IDENTIFIED BY 'password';
  FLUSH PRIVILEGES;
  

2. 配置 Flink CDC 连接器

Flink CDC 连接器的配置需要在 Flink SQL 中完成，以下是关键步骤：

（1）上传自定义连接器 JAR 包

• 登录实时计算控制台。
• 在左侧导航栏中，单击 连接器。
• 在 自定义连接器 页面，上传 Flink CDC 连接器的 JAR 文件（如 PolarDBO Flink CDC JAR 包）。
• 上传完成后，系统会解析连接器内容。如果解析成功，继续下一步。

（2）创建临时表并配置连接参数

在 Flink 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'
);

说明： - 替换 <yourHostname>、<yourPort>、<yourUserName> 和 <yourPassWord> 为实际的数据库连接信息。 - slot.name 是用于逻辑复制的槽名称，需确保唯一性。

3. 启动 Flink 任务

完成上述配置后，可以通过以下步骤启动 Flink 任务：

（1）下载并上传 Flink CDC Connector

• 下载 Flink CDC Connector 和目标存储系统的 Connector（如 StarRocks Connector）。
• 将下载的文件上传到 DataFlow 集群的 /opt/apps/FLINK/flink-current/lib 目录下。

（2）启动 Flink 集群

执行以下命令启动 Flink 集群：

/opt/apps/FLINK/flink-current/bin/start-cluster.sh

（3）编写并执行 Flink SQL 作业

• 编写 Flink SQL 作业并保存为 .sql 文件。例如：

  CREATE DATABASE IF NOT EXISTS `default_catalog`.`flink_cdc`;
  CREATE TABLE IF NOT EXISTS `default_catalog`.`flink_cdc`.`customers_src` (
   `customer_id` INT NOT NULL,
   `customer_age` FLOAT NOT NULL,
   `customer_name` STRING NOT NULL,
   PRIMARY KEY(`customer_id`) NOT ENFORCED
  ) WITH (
   'connector' = 'mysql-cdc',
   'hostname' = 'rm-2ze8398257383****.mysql.rds.aliyuncs.com',
   'port' = '3306',
   'username' = 'emr_test',
   'password' = 'Yz12****',
   'database-name' = 'flink_cdc'
  );
  

• 执行以下命令运行 SQL 作业：

  /opt/apps/FLINK/flink-current/bin/sql-client.sh -f demo.sql
  

4. 验证数据同步

• 登录目标存储系统（如 StarRocks 或 Hologres），检查数据是否成功同步。
• 插入、更新或删除源数据库中的数据，观察目标系统中的变化。

重要提醒

• 网络要求：确保 DataFlow 集群、数据库实例和目标存储系统在同一个 VPC 下，并开启公网访问（如需外部访问）。
• 版本兼容性：Flink CDC 连接器需与 Flink 引擎版本兼容，建议使用 vvr-4.0.12-flink-1.13 及以上版本。
• 安全性：避免在代码中直接暴露敏感信息（如用户名和密码），建议使用变量管理功能。

通过以上步骤，您可以成功配置并登录 Flink CDC，实现数据的实时同步和处理。