tencent cloud

CREATE TABLE `rmq_sink_json_table` (
    `id` int,
    `name` STRING
) WITH (
      'connector' = 'rabbitmq', 					 			-- Here, it should be 'rabbitmq'.
    'host' = 'xxxx',    											-- The RMQ host.
    'port' = 'xxxx',			 		      					-- The RMQ port.
    'vhost' = '/',				   									-- The virtual host.
    'username' = 'xxxx',         		 							-- The username.
    'password' = 'xxxx',     									-- The password.
    'exchange' = 'exchange',             		 	-- The exchange name.
      'routing-key' = 'Key',										-- The key bound.
      'format' = 'json',                 -- The data format (JSON).
    'json.fail-on-missing-field' = 'false'   -- If this is 'false', no errors will occur even when parameters are missing.
    'json.ignore-parse-errors' = 'true'    -- If this is 'true', all parse errors will be ignored.
);

CREATE TABLE `rmq_sink_csv_table` (
    `id` int,
    `name` STRING
) WITH (
      'connector' = 'rabbitmq', 					 			-- Here, it should be 'rabbitmq'.
    'host' = 'xxxx',    											-- The RMQ host.
    'port' = 'xxxx',			 		      					-- The RMQ port.
    'vhost' = '/',				   									-- The virtual host.
    'username' = 'xxxx',         		 							-- The username.
    'password' = 'xxxx',     									-- The password.
    'exchange' = 'exchange',             		 	-- The exchange name.
      'routing-key' = 'Key',										-- The key bound.
      'format' = 'csv',               -- The data format (CSV).
);

-- Please replace the parameter values below with the corresponding values for the cluster used.
CREATE TABLE `rabbitmq_source_json_table` (`id` INT, `name` STRING) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://host:port/database?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
  'table-name' = 'source_table_name',
  'username' = 'username',
  'password' = 'password'
);

CREATE TABLE `rabbitmq_sink_json_table` (`id` INT, `name` STRING) WITH (
  'connector' = 'rabbitmq',
  'host' = 'host',
  'port' = 'port',
  'vhost' = 'vhost',
  'username' = 'username',
  'password' = 'password',
  'queue' = 'queue-name',
  'exchange'='exchange',
  'routing-key'='key',
  'format' = 'json'
);
insert into rabbitmq_sink_json_table select * from rabbitmq_source_json_table;