Realtime Compute for Apache Flink:Create a Message Queue for Apache RocketMQ source table

What is Message Queue for Apache RocketMQ?

Message Queue for Apache RocketMQ is a distributed messaging middleware that is developed by Alibaba Cloud based on Apache RocketMQ. Message Queue for Apache RocketMQ features low latency, high concurrency, high availability, and high reliability. Message Queue for Apache RocketMQ provides the asynchronous decoupling and load shifting features for distributed application systems. It also supports features for Internet applications, including massive message accumulation, high throughput, and reliable retry.

Prerequisites

Resources are created in the Message Queue for Apache RocketMQ console. For more information, see Create resources.

Limits

Only Flink that uses Ververica Runtime (VVR) 2.0.0 or later supports Message Queue for Apache RocketMQ connectors.

DDL syntax

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' = '|'
);

Parameters in the WITH clause

Data type mappings

Attribute fields

Sample code

• CSV format
  The following example shows a Message Queue for Apache RocketMQ message in the CSV format.

  1,name,male 
  2,name,female

  Note A Message Queue for Apache RocketMQ message can contain any number of data records. Multiple data records are separated by \n.
  You can use the following DDL statement to declare a Message Queue for Apache RocketMQ source table in a Flink job:

  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' = '|'
  );

• Binary format

  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;