tencent cloud

CREATE TABLE `mysql_cdc` (
  `user_id` bigint,
  `item_id` bigint,
  `behavior` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',      -- Here, it should be 'mysql-cdc'.
  'hostname' = '9.134.34.15',   -- IP of the database server.
  'port' = '3306',                --  Integer port number of the database server.
  'username' = 'root',        -- Name of the database to use when connecting to the MySQL database server (SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, and RELOAD permissions are required).
  'password' = 'xxx',    -- Password to use when connecting to the MySQL database server.
  'database-name' = 'test',   -- Database name of the MySQL server to monitor.
  'table-name' = 'user_behavior'      -- Table name of the MySQL database to monitor.
);

CREATE TABLE `pk_starrocks`(
  `user_id` bigint,
  `item_id` bigint,
  `behavior` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'starrocks',
  'jdbc-url' = 'jdbc:mysql://172.28.28.98:9030',
  'load-url' = '172.28.28.98:8030',
  'database-name' = 'oceanus',
  'table-name' = 'pk_user_behavior',
  'username' = 'root',
  'password' = 'xxx',
  'sink.buffer-flush.interval-ms' = '15000',
  'sink.properties.format' = 'json',
  'sink.properties.strip_outer_array' = 'true', -- Here, it should be "true".
  -- 'sink.parallelism' = '1',
  'sink.max-retries' = '3',
  'sink.semantic' = 'exactly-once'
);

INSERT INTO `pk_starrocks` SELECT * FROM `mysql_cdc`;

CREATE TABLE `starrocks` (
  `user_id` bigint,
  `item_id` bigint,
  `behavior` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'starrocks' ,
  'jdbc-url' = 'jdbc:mysql://172.28.28.98:9030', -- query_port, FE mysql server port
  'scan-url' = '172.28.28.98:8030', -- http_port
  'database-name' = 'oceanus',
  'table-name' = 'pk_user_behavior',
  'username' = 'root',
  'password' = 'xxx'
);

CREATE TABLE `print_sink` (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'logger'
);

INSERT INTO `print_sink`
SELECT * FROM starrocks;

CREATE TABLE `starrocks` (
  `user_id` bigint,
  `item_id` bigint,
  `behavior` STRING,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'starrocks' ,
  'jdbc-url' = 'jdbc:mysql://172.28.28.98:9030', -- query_port, FE mysql server port
  'scan-url' = '172.28.28.98:8030', -- http_port
  'database-name' = 'oceanus',
  'table-name' = 'pk_user_behavior',
  'username' = 'root',
  'password' = 'xxx'
);

CREATE TABLE `datagen` (
  `user_id` BIGINT,
  `ts` as PROCTIME(),
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '20'
);

CREATE TABLE `print_sink` (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP,
  PRIMARY KEY (`user_id`) NOT ENFORCED
) WITH (
  'connector' = 'logger'
);

INSERT INTO `print_sink`
SELECT a.user_id,b.item_id,b.behavior,a.ts
FROM `datagen` a LEFT JOIN `starrocks` FOR SYSTEM_TIME AS OF a.ts as b
ON a.user_id = b.user_id;