消息队列RocketMQ版源表

什么是消息队列RocketMQ版

消息队列RocketMQ版是阿里云基于Apache RocketMQ构建的低延迟、高并发、高可用和高可靠的分布式消息中间件。消息队列RocketMQ版既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐和可靠重试等特性。

DDL定义

create table mq_source(
  x varchar,
  y varchar,
  z varchar
) with (
  'connector' = 'mq',
  'topic' = '<yourTopicName>',
  'endpoint' = '<yourEndpoint>',
  'pullIntervalMs' = '1000',
  'accessId' = '<yourAccessId>',
  'accessKey' = '<yourAccessSecret>',
  'startMessageOffset' = '1000',
  'consumerGroup' = '<yourConsumerGroup>',
  'fieldDelimiter' = '|'
);

WITH参数

类型映射

属性字段

代码示例

• CSV格式
  假设您的1条CSV格式消息记录如下。

  1,name,male 
  2,name,female

  说明 1条RocketMQ消息可以包括0条到多条数据记录，记录之间使用\n分隔。
  Flink作业中，声明RocketMQ数据源表的DDL如下。

  create table mq_source(
     id varchar,
     name varchar,
     gender varchar,
     topic varchar metadata virtual
  ) with (
     'connector' = 'mq',
     'topic' = 'mq-test',
     'endpoint' = '<yourEndpoint>',
     'pullIntervalMs' = '1000',
     'accessId' = '<yourAccessId>',
     'accessKey' = '<yourAccessSecret>',
     'startMessageOffset' = '1000',
     'consumerGroup' = 'mq-group',
     'fieldDelimiter' = '|'
  );

• 二进制格式

  create temporary table source_table (
    mess varbinary
  ) with (
    'connector' = 'mq',
    'endpoint' = '<yourEndpoint>',
    'pullIntervalMs' = '500',
    'accessId' = '<yourAccessId>',
    'accessKey' = '<yourAccessSecret>',
    'topic' = 'mq-test',
    'consumerGroup' = 'mq-group'
  );
  
  create temporary table out_table (
    commodity varchar
  ) with (
    'connector' = 'print'
  );
  
  insert into out_table
  select 
    cast(mess as varchar)
  from source_table;