PieCloudDB Flink Connector：让数据流动起来

create table student (id int primary key, name varchar(32), score int, sex char(1)); 

mysql> desc student;
+---------+-------------+------+-----+---------+-------+
| Field   | Type        | Null | Key | Default | Extra |
+---------+-------------+------+-----+---------+-------+
| id      | int         | NO   | PRI | NULL    |       |
| name    | varchar(32) | YES  |     | NULL    |       |
| score   | int         | YES  |     | NULL    |       |
| sex     | char(1)     | YES  |     | NULL    |       |
+---------+-------------+------+-----+---------+-------+
4 rows in set (0.03 sec)

insert into student (id, name, score, sex) values (1, 'student1', 65, '1');
insert into student (id, name, score, sex) values (2, 'student2', 75, '0');
insert into student (id, name, score, sex) values (3, 'student3', 85, '1');
insert into student (id, name, score, sex) values (4, 'student4', 95, '0');

mysql> select * from student;
+----+----------+-------+------+
| id | name     | score | sex  |
+----+----------+-------+------+
| 1  | student1 | 65    | 1    |
| 2  | student2 | 75    | 1    |
| 3  | student3 | NULL  | NULL |
| 4  | student4 | NULL  | NULL |
+----+----------+-------+------+
4 rows in set (0.01 sec)

create table student (id int primary key, name varchar(32), score int, sex char(1)); 

demo=> \d student
                    Table "public.student"
 Column | Type                  | Collation | Nullable | Default
--------+-----------------------+-----------+----------+---------
id      | integer               |           | not null | 
name    | character varying(32) |           |          |
score   | integer               |           |          |
sex     | character(1)          |           |          |

demo=> select * from student;
 id | name | score | sex
----+------+-------+-----
(0 rows)

ubuntu :: work/flink/flink-1.18.0 >> bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host ubuntu.
Starting taskexecutor daemon on host ubuntu.
[INFo] 1 instance(s) of taskexecutor are already running on ubuntu.
Starting taskexecutor daemon on host ubuntu.

Flink SQL> add jar '/home/frankie/work/download/flink-sql-connector-mysql-cdc-2.4.0.jar'; 
[INFO] Execute statement succeed.
Flink SQL> add jar '/home/frankie/work/download/flink-sql-connector-pieclouddb-1.2.0.jar'; 
[INFO] Execute statement succeed.
Flink SQL> SET 'execution.checkpointing.interval' = '3s'; 
[INFO] Execute statement succeed.

Flink SQL> CREATE TABLE source_student_mysql ( 
     id INT, 
     name STRING, 
     score INT, 
     sex CHAR(1), 
     PRIMARY KEY(id) NOT ENFORCED 
     ) WITH ( 
     'connector' = 'mysql-cdc', 
     'hostname' = 'mysql-host', 
     'port' = '3306', 
     'username' = 'root', 
     'password' = '123456', 
     'database-name' = 'testdb', 
     'table-name' = 'student'); 
[INFO] Execute statement succeed. 

Flink SQL> CREATE TABLE sink_student_pdb ( 
     id INT, 
     name STRING, 
     score INT, 
     sex CHAR(1), 
     PRIMARY KEY(id) NOT ENFORCED 
     ) WITH ( 
     'connector' = 'pieclouddb', 
     'hostname' = 'pieclouddb-host', 
     'port' = 'your-pieclouddb-port', 
     'username' = 'your-username', 
     'password' = 'your-password', 
     'pdb_warehouse' = 'your-pdbwarehouse', 
     'database-name' = 'demo', 
     'table-name' = 'student', 
     'load_mode' = 'merge'); 
[INFO] Execute statement succeed.

Flink SQL> INSERT INTO sink_student_pdb SELECT * FROM source_student_mysql; 
[INFO] Submitting SQL update statement to the cluster... 
[INFO] SQL update statement has been successfully submitted to the cluster: 
Job ID: 660b747ef8fb64f95064a461af9924bc