Flinkcdc2.0 uses flinksql to collect MySQL

1. Dependency management

Put the following dependent packages into FLINK_HOME/lib Next .

flink-sql-connector-mysql-cdc-2.2.0.jar

flink-connector-jdbc_2.11-1.14.3.jar

flink-shaded-hadoop-2-uber-2.8.3-10.0.jar

2.Flink Global configuration

modify flink-conf.yaml file ：

execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

execution.checkpointing.max-concurrent-checkpoints: 1

execution.checkpointing.mode: EXACTLY_ONCE

execution.checkpointing.timeout: 10min

state.backend: filesystem

state.checkpoints.dir: hdfs://mycluster/flinkcdc-checkpoints

3.sql-client Submit job mode

1.Standalone Pattern

start-up sql-client：bin/sql-client.sh embedded

Be careful , If you use standalone mode , You need to start one Flink standalone colony , The method is as follows ：

bin/start-cluster.sh

2.yarn-session Pattern （ How to use this case ）

Start... First Flink yarn-session colony ：bin/yarn-session.sh -s 1 -jm 1024 -tm 1024

And then it starts sql-client：bin/sql-client.sh embedded -s yarn-session

4.checkpoint To configure

Official website address ：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/config/#checkpointing

#sql-client Set up checkpoint Parameters

SET 'execution.checkpointing.interval' = '10s';

SET 'parallelism.default' = '3';

5. establish source table

CREATE TABLE `cars`(

 `id` BIGINT,

 `owerId` BIGINT,

 `carCode` STRING,

 `carColor` STRING,

 `type` BIGINT,

 `remark` STRING,

 PRIMARY KEY(id) NOT ENFORCED

) WITH (

 'connector' = 'mysql-cdc',

 'hostname' = 'hadoop1',

 'port' = '3306',

 'username' = 'hive',

 'password' = 'hive',

 'database-name' = 'sca',

 'table-name' = 'cars',

 'connect.timeout' = '60s'

);

6. establish sink table

CREATE TABLE `cars_copy`(

 `id` BIGINT,

 `owerId` BIGINT,

 `carCode` STRING,

 `carColor` STRING,

 `type` BIGINT,

 `remark` STRING,

 PRIMARY KEY(id) NOT ENFORCED

) WITH (

 'connector' = 'jdbc',

 'url' = 'jdbc:mysql://hadoop1:3306/sca?useUnicode=true&characterEncoding=utf8',

 'username' = 'hive',

 'password' = 'hive',

 'table-name' = 'cars_copy',

 'sink.parallelism' = '2'

);

7.source to sink table

Write the collected data into MySQL

insert into cars_copy SELECT * FROM cars;

The number of data records in the query result table

select count(*) from cars_copy

New test data set （ Check the result table again ）

insert into `cars` (`id`, `owerId`, `carCode`, `carColor`, `type`, `remark`) values('10096006','10244815',' harbor T·7RONE',' Red ','1',NULL);

insert into `cars` (`id`, `owerId`, `carCode`, `carColor`, `type`, `remark`) values('10096007','10244816',' harbor T·7RONE',' yellow ','1',NULL);

remarks ： If manually cacel job, Next time restart job The data in the table will still be collected again .

8.cacel job Save on Save point

bin/flink stop --savepointPath hdfs://mycluster/flinkcdc-savepoints -Dyarn.application.id=application_1658045078748_0001 79ce915e39fc1d18a194b6a464d7c3fd

remarks ： The last parameter is yarn Medium job id, The second parameter is flink Of job id.

9.cacel job Then resume job

# Set up job From the last time savepoint Position start processing

SET 'execution.checkpointing.interval' = '10s';

SET 'parallelism.default' = '3';

SET 'execution.savepoint.path' = 'hdfs://mycluster/flinkcdc-savepoints/savepoint-79ce91-92206bcaaad2';

remarks ： The value of this parameter is savepoint route .

# perform flink sql job

insert into cars_copy SELECT * FROM cars;

# New test data set

insert into `cars` (`id`, `owerId`, `carCode`, `carColor`, `type`, `remark`) values('10096008','10244815',' harbor T·7RONE',' Red ','1',NULL);

insert into `cars` (`id`, `owerId`, `carCode`, `carColor`, `type`, `remark`) values('10096009','10244816',' harbor T·7RONE',' yellow ','1',NULL);

# Re query the number of data records in the result table

select count(*) from cars_copy

Normal condition , What is collected at this time is the new data , Historical data will not be collected .

remarks ：Flink  SQL Way to collect MySQL data , Easy to use , But only single tables are supported .