Flink CDC中比如这个sql, 这些变量怎么传过去？如果使用自定义row？

在Flink CDC中，您提到的SQL语句中包含多个变量（如selectId、dt、site、account），这些变量需要通过Flink的变量管理机制进行传递和解析。以下是详细的解决方案，结合了知识库中的相关内容。

1. 使用作业变量传递参数

Flink支持通过作业变量的方式动态传递参数，适用于SQL作业开发场景。以下是具体步骤：

步骤 1：定义变量

在SQL语句中使用${变量名}格式定义变量。例如：

INSERT INTO my_catalog.pi_test.pi_test_paimon_amazon_online_sales 
SELECT ("${selectId}" IS_DELETE) AS selectId, 1 
FROM my_catalog.pi_test.pi_test_paimon_amazon_online_sales 
WHERE dt LIKE '${dt}' 
  AND site = '${site}' 
  AND account = '${account}';

步骤 2：配置变量值

在Flink开发控制台中，按照以下方式配置变量值： 1. 在数据开发 > ETL页面，打开目标作业草稿。 2. 在SQL编辑器中输入上述SQL语句。 3. Flink会自动识别${变量名}格式的变量，并在右侧显示作业变量面板。 4. 在作业变量面板中填写变量的实际值，例如： - selectId: 123 - dt: '2023-10-01' - site: 'US' - account: 'example_account'

步骤 3：启动作业

• 单击目标作业操作列下的启动按钮。
• 在启动面板中可以再次修改变量值（仅对本次启动生效）。

2. 使用项目变量传递敏感信息

如果变量涉及敏感信息（如密码或AccessKey），建议使用项目变量进行管理。项目变量的使用格式为${secret_values.变量名}。

步骤 1：创建项目变量

1. 在Flink开发控制台中，进入运维中心 > 作业运维页面。
2. 单击项目变量，创建新的变量，例如：
   • 变量名：mysqlusername
   • 值：your_username
   • 变量名：mysqlpassword
   • 值：your_password

步骤 2：在SQL中引用项目变量

在SQL语句中使用${secret_values.变量名}格式引用项目变量。例如：

INSERT INTO my_catalog.pi_test.pi_test_paimon_amazon_online_sales 
SELECT ("${selectId}" IS_DELETE) AS selectId, 1 
FROM my_catalog.pi_test.pi_test_paimon_amazon_online_sales 
WHERE dt LIKE '${dt}' 
  AND site = '${site}' 
  AND account = '${account}'
  AND username = '${secret_values.mysqlusername}'
  AND password = '${secret_values.mysqlpassword}';

3. 自定义Row处理

如果您需要通过自定义Row对象传递参数，可以通过Flink的Table API或DataStream API实现。以下是基于Table API的示例：

步骤 1：注册表并插入数据

1. 创建一个临时表，用于接收自定义Row数据：

   CREATE TEMPORARY TABLE custom_input (
      selectId STRING,
      dt STRING,
      site STRING,
      account STRING
   ) WITH (
      'connector' = 'datagen',
      'rows-per-second' = '1'
   );
   

2. 插入自定义Row数据：

   INSERT INTO custom_input VALUES ('123', '2023-10-01', 'US', 'example_account');
   

步骤 2：关联自定义Row与目标表

将自定义Row数据与目标表进行关联查询：

INSERT INTO my_catalog.pi_test.pi_test_paimon_amazon_online_sales 
SELECT (c.selectId IS_DELETE) AS selectId, 1 
FROM my_catalog.pi_test.pi_test_paimon_amazon_online_sales t
JOIN custom_input c
ON t.dt LIKE c.dt 
   AND t.site = c.site 
   AND t.account = c.account;

4. 注意事项

• 变量唯一性：确保变量名在单个作业内唯一，避免冲突。
• 特殊字符处理：如果变量值包含特殊字符（如'或\），需要进行转义处理。
• SQL语法限制：在持久表（CREATE TABLE）中不支持使用变量，仅支持在临时表（CREATE TEMPORARY TABLE）中使用。
• 并发读取：如果涉及MySQL CDC源表的多并发读取，需配置server-id范围，且范围中的server-id个数需大于等于并发度。

通过以上方法，您可以灵活地将变量传递到Flink CDC SQL中，并根据需求选择适合的变量管理方式。