CREATE TABLE `DataInput` (
      `id` INT,
      `name` VARCHAR,
       `age` INT
) WITH (
    'connector' = 'tdsql-subscribe',   -- 注意选择对应的内置  Connector
    'tdsql.database.name' = 'test_case_2022_06_0*', -- 对订阅消息进行过滤，消费数据库名满足 test_case_2022_06_0* 正则的订阅数据
    'tdsql.table.name' = 'test_0*', -- 对订阅消息进行过滤，消费数据表满足 test_0* 正则的订阅数据
    'topic' = 'topic-subs-5xop97nffk-tdsqlshard-xxx',  -- 替换为订阅任务消费的 Topic
    'scan.startup.mode' = 'latest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets 的任何一种
    'properties.bootstrap.servers' = 'guangzhou-kafka-2.cdb-dts.tencentcs.com.cn:3212',  -- 替换为您的订阅任务 Kafka 连接地址
    'properties.group.id' = 'consumer-grp-subs-xxx-kk', 
    'format' = 'protobuf', -- 只能是protobuf格式
    'properties.security.protocol'='SASL_PLAINTEXT', -- 认证协议
    'properties.sasl.mechanism'='SCRAM-SHA-512', -- 认证方式
    'properties.sasl.jaas.config'='org.apache.kafka.common.security.scram.ScramLoginModule required username="account-subs-xxx-username" password="psw";' --用户名和密码
);
CREATE TABLE `jdbc_upsert_sink_table` (
    id INT PRIMARY KEY NOT ENFORCED,  
    name STRING,
    age INT
) WITH (
    -- 指定数据库连接参数
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://172.28.28.138:3306/testdb', -- 请替换为您的实际 MySQL 连接参数
    'table-name' = 'sink', -- 需要写入的数据表
    'username' = 'user',      -- 数据库访问的用户名（需要提供 INSERT 权限）
    'password' = 'psw'  -- 数据库访问的密码
);
INSERT INTO jdbc_upsert_sink_table SELECT * FROM DataInput;