Oceanus practice consumption CMQ subject model data source

Oceanus Flink CMQ connector Support the data source table and destination table of queue model , The topic model data source table and destination table are not supported at the moment .CMQ Topic subscription can synchronize topic model data to queue model in real time , With this mechanism , We can do it in Oceanus Realization CMQ Reading of topic model data source table .

1. Environment building

1.1 establish Oceanus colony

stay Oceanus Console 【 Cluster management 】->【 New cluster 】 Page create cluster , Choose the region 、 Availability zone 、VPC、 journal 、 Storage , Set the initial password, etc .

If not used before VPC, journal , Store these components , You need to create it first .

The created cluster is as follows ：

Oceanus colony

1.2 newly build CMQ The theme

stay CMQ Console 【 Theme subscription 】-> 【 newly build 】 The theme , Enter the topic name , Others can keep the default value . The new theme is as follows ：

CMQ The theme

1.3 newly build CMQ queue

stay CMQ Console 【 queue 】-> 【 newly build 】 The theme , Enter the queue name 、 Message lifecycle 、 Maximum number of stacked messages , Others can keep the default value . Let's create two new queues here , One of them is used to subscribe to CMQ Subject model data , The other is used as Oceanus The purpose table of the operation . The new theme is as follows ：

CMQ queue

1.4 newly build CMQ Theme subscription

stay CMQ Topic list page , Click... In the subject action column 【 subscribe 】 link , Get into 【 subscriber 】 list , New subscription , Enter the subscription name , Terminal type selection Queue Queue service , Subscription address selection cs2 Queues , Others remain at default . The new subscribers are as follows ：

CMQ Theme subscription

2. Job creation

2.1 establish SQL Homework

stay Oceanus Console 【 Job management 】->【 New job 】-> SQL Homework , Select the cluster creation job just created . Then at the end of the assignment 【 Development and debugging 】->【 Operation parameters 】 Add necessary connector cmq-1.1.1.

SQL Homework

2.2 Create data source table and destination table

At the end of the job 【 Development and debugging 】->【 Insert template 】 choice CMQ Read & Write template , And add . Modify the parameters queue、secret-id、secret-key.

Be careful ： It is strongly recommended to use a with minimal permissions secret-id and secret-key, And keep it confidential , Prevent safety risks caused by leakage .

CMQ Read & write in

CREATE TABLE `CMQSourceTable` (
    `id` bigint,
    `request_method` varchar(80),
    `response` varchar(80),
    PRIMARY KEY (`id`) NOT ENFORCED -- If you want to do data De duplication , You need to specify PK, Distinguish different data according to this primary key 
) WITH (
	'connector' = 'cmq', 											-- It has to be for  'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',	--cmq Of the region nameServer
    'queue' = 'cs2',											--cmq Name of the queue 
    'secret-id' = 'Your SecretId',			                                -- account number secretId
    'secret-key' = 'Your SecretKey',                                          -- account number secretKey
    'sign-method' = 'HmacSHA1',                                     -- The way of signing 
    'format' = 'csv',                                               -- Define the data format （JSON  Format ）
    'batch-size' = '16',                                            -- Number of mass consumption messages / Number of messages sent in batch 
    'request-timeout' = '5000ms',                                   -- The timeout for the request 
    'polling-wait-timeout'= '10s',                                  --source Parameters ;  Wait time when data cannot be obtained 
    'key-alive-timeout'= '5min',                                    --source Parameters ; contain primary key The news of ,CMQ Effective time for weight removal 
    'retry-times' = '3',                                            --sink Parameters ; Number of retries to send message 
    'max-block-timeout' = '0s'                                      --sink Parameters ; Maximum waiting time for batch sending data 
);

CREATE TABLE `CMQSinkTable` (
    `id` bigint,
    `request_method` varchar(80),
    `response` varchar(80),
    PRIMARY KEY (`id`) NOT ENFORCED -- If you want to do data De duplication , You need to specify PK, Distinguish different data according to this primary key 
) WITH (
	'connector' = 'cmq', 											-- It has to be for  'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',	--cmq Of the region nameServer
    'queue' = 'sink_queue',											--cmq Name of the queue 
    'secret-id' = 'Your SecretId',			                                -- account number secretId
    'secret-key' = 'Your SecretKey',                                          -- account number secretKey
    'sign-method' = 'HmacSHA1',                                     -- The way of signing 
    'format' = 'csv',                                               -- Define the data format （JSON  Format ）
    'batch-size' = '16',                                            -- Number of mass consumption messages / Number of messages sent in batch 
    'request-timeout' = '5000ms',                                   -- The timeout for the request 
    'polling-wait-timeout'= '10s',                                  --source Parameters ;  Wait time when data cannot be obtained 
    'key-alive-timeout'= '5min',                                    --source Parameters ; contain primary key The news of ,CMQ Effective time for weight removal 
    'retry-times' = '3',                                            --sink Parameters ; Number of retries to send message 
    'max-block-timeout' = '0s'                                      --sink Parameters ; Maximum waiting time for batch sending data 
);

insert into CMQSinkTable select * from CMQSourceTable;

2.3 Operator operation

Only the simplest data insertion is done here .

insert into CMQSinkTable select *from CMQSourceTable;

3. Validation summary

stay CMQ The console is called test Send a message in the subject of , Can be found in sink_queue Message received in queue .

Send a subject message

Receive queue message