tencent cloud

CREATE TABLE `DataInput` (
      `id` INT,
      `name` VARCHAR,
       `age` INT,
) WITH (
    'connector' = 'tdsql-subscribe',   -- Make sure you specify the corresponding connector.
    'tdsql.database.name' = 'test_case_2022_06_0*', -- Filter subscription messages to consume subscription data of databases whose name matches the regex `test_case_2022_06_0*`.
    'tdsql.table.name' = 'test_0*', -- Filter subscription messages to consume subscription data of tables whose name matches the regex `test_0*`.
    'topic' = 'topic-subs-5xop97nffk-tdsqlshard-xxx',  -- Replace it with the topic consumed by the subscription task.
    'scan.startup.mode' = 'earliest-offset', -- Valid values: `latest-offset`, `earliest-offset`, `specific-offsets, and `group-offsets`.
    'properties.bootstrap.servers' = 'guangzhou-kafka-2.cdb-dts.tencentcs.com.cn:3212',  -- Replace it with the Kafka connection address of your subscription task.
    'properties.group.id' = 'consumer-grp-subs-xxx-kk', 
    'format' = 'protobuf', -- Only protobuf is allowed.
    'properties.security.protocol'='SASL_PLAINTEXT', -- Authentication protocol.
    'properties.sasl.mechanism'='SCRAM-SHA-512', -- Authentication method.
    'properties.sasl.jaas.config'='org.apache.kafka.common.security.scram.ScramLoginModule required username="account-subs-xxx-username" password="psw";' -- Username and password.
);
CREATE TABLE `jdbc_upsert_sink_table` (
    id INT PRIMARY KEY NOT ENFORCED,  
    name STRING,
    age INT
) WITH (
    -- Specify the parameters for database connection.
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://172.28.28.138:3306/testdb', -- Replace it with your MySQL database connection URL.
    'table-name' = 'sink', --The table into which the data will be written.
    'username' = 'user',      -- The username (with the INSERT permission required) for database access.
    'password' = 'psw'  -- The password for database access.
);
INSERT INTO jdbc_upsert_sink_table SELECT * FROM DataInput;