tencent cloud

CREATE TABLE `rmq_sink_json_table` (
    `id` int,
    `name` STRING
) WITH (
      'connector' = 'rabbitmq', 					 			-- 必须为 'rabbitmq'
    'host' = 'xxxx',    											-- rabbitmq host
    'port' = 'xxxx',			 		      					-- rabbitmq 端口
    'vhost' = '/',				   									-- 虚拟主机
    'username' = 'xxxx',         		 							-- 用户名
    'password' = 'xxxx',     									-- 用户密码
    'exchange' = 'exchange',             		 	-- 交换机名
      'routing-key' = 'Key',										-- 绑定 Key
      'format' = 'json',												-- 定义数据格式（JSON 格式）
    'json.fail-on-missing-field' = 'false',	  -- 如果设置为 false, 则遇到缺失字段不会报错
    'json.ignore-parse-errors' = 'true'       -- 如果设置为 true，则忽略任何解析报错
);

CREATE TABLE `rmq_sink_csv_table` (
    `id` int,
    `name` STRING
) WITH (
      'connector' = 'rabbitmq', 					 			-- 必须为 'rabbitmq'
    'host' = 'xxxx',    											-- rabbitmq host
    'port' = 'xxxx',			 		      					-- rabbitmq 端口
    'vhost' = '/',				   									-- 虚拟主机
    'username' = 'xxxx',         		 							-- 用户名
    'password' = 'xxxx',     									-- 用户密码
    'exchange' = 'exchange',             		 	-- 交换机名
      'routing-key' = 'Key',										-- 绑定 Key
      'format' = 'csv'													-- 定义数据格式（CSV 格式）
);

-- 提示：请将参数替换成所属集群的信息
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;