Open Source Data Lakehouse Framework LakeSoul Support Flink CDC Whole Database Synchronization
Published: 2023-05-24 08:50:11

**First, attach the Github website**

**LakeSoul：[https://github.com/lakesoul-io/LakeSoul](https://github.com/lakesoul-io/LakeSoul)**

## 1. Introduction

LakeSoul is a cloud-native Lakehouse framework developed by DMetaSoul team, and supports scalable metadata management, ACID transactions, efficient and flexible upsert operation, schema evolution, and unified streaming & batch processing. LakeSoul Flink CDC Sink supports the entire database synchronization from MySQL to LakeSoul, and can support automatic table creation, automatic schema change, exactly once semantics, etc.

This article will introduce the functional advantages of Flink CDC and the tutorial of using LakeSoul Flink CDC whole database synchronization. In this tutorial, we fully demonstrate synchronizing a MySQL database to LakeSoul, including automatic table creation, DDL changes and other operations.

## 2. Flink CDC

### 2.1 CDC Introduction

CDC stands for Change Data Capture, which monitors and captures incremental change records from the source database, synchronizing them to one or more data destinations (Sinks). The database changes that can be captured include INSERT, UPDATE, DELETE of data or data tables, and these changes are recorded in their entirety and written to the message middleware in the order in which they occur.

LakeSoul provides a set of independent CDC semantic expression specification, set a CDC Op column through the table attribute, which can indicate the operation type of each data, and will automatically merge according to the operation semantic in the subsequent Merge. You can import CDC data into LakeSoul after conversion by Debezium, Canal, Flink, etc.

### 2.2 Why use Flink CDC

Instead of using the Flink CDC, the LakeSoul framework initially transforms the CDC data into LakeSoul via Debezium:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/08v89pl5tel3exja1ath.PNG)

The overall process can be divided into the following:

1. link Mysql and kafka
2. Create Debezium CDC synchronization task
3. Use Spark Streaming, consume Kafka data and synchronize updates to LakeSoul

It can be seen that the processing link for importing CDC data into LakeSoul after Debezium conversion is relatively long, and there are also many components required. Debezium achieves CDC functionality through Kafka Streams, while LakeSoul currently uses the Flink CDC module, which can skip the transition between Debezium and Kafka and use Flink cdc connectors to directly subscribe to changes in upstream data sources.

The flink-cdc-connectors component developed by the Flink community is a source component that can read full and incremental change data directly from MySQL, PostgreSQL, and other databases. flink-cdc-connectors can capture all changes that occur in one or more tables. The schema usually has a pre-record and a post-record. flink-cdc-connectors can be used directly in Flink in unconstrained mode without using middleware such as kafka to transmit data.

Flink CDC can get the binlog directly from the database for downstream business calculation and analysis. In terms of internal implementation, flink-cdc-connectors component has a set of Debezium and Kafka components built-in, but this detail is shielded from users.

You can see that the CDC data is transformed by Flink and imported into LakeSoul as shown in the figure below, and the data is no longer synchronized by kafka, simplifying the overall architecture:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/airfue2tw9uf0sh14wju.PNG)

After the LakeSoul Flink job is started, the initialization phase starts by reading all tables in the configured MySQL DB (excluding tables that do not need to be synchronized). For each table, it first determines if it exists in LakeSoul, and if not, it automatically creates a LakeSoul table with the same Schema as the MySQL counterpart. After initialization, the CDC Streams of all tables are read and written to each LakeSoul table in an Upsert manner. If a DDL Schema change occurs to a MySQL table during synchronization, the change is also applied to the corresponding LakeSoul table. The following is a complete tutorial on how to use LakeSoul Flink CDC whole database synchronization.

## 3. LakeSoul Flink CDC Whole Database Synchronization

### 3.1  Prepare the environment

#### 3.1.1 Start a local MySQL database

It is recommended to use the MySQL Docker image to quickly start a MySQL database instance:

```bash
cd docker/lakesoul-docker-compose-env/

docker run --name lakesoul-test-mysql -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=test_cdc -p 3306:3306 -d mysql:8
```

#### 3.1.2 Configuring LakeSoul Meta DB and Spark Environment
##### 1. Start A Local PostgreSQL DB

The quickest way to start a pg DB is via docker container:

```bash
docker run -d --name lakesoul-test-pg -p5432:5432 -e POSTGRES_USER=lakesoul_test -e POSTGRES_PASSWORD=lakesoul_test -e POSTGRES_DB=lakesoul_test -d swr.cn-north-4.myhuaweicloud.com/dmetasoul-repo/postgres:14.5
```

Perform PG database initialization, enter the docker container, copy `meta_init.sql` to the container, and you can check the container id with the `docker ps` command:

```bash
docker exec -it container id /bin/bash
docker cp script/meta_init.sql container id:script/meta_init.sql
```

Execute in the Docker container:

```bash
PGPASSWORD=lakesoul_test psql -h localhost -p 5432 -U lakesoul_test -f script/meta_init.sql
```

##### 2. Install an Apache Spark environment

You could download spark distribution from https://spark.apache.org/downloads.html, and please choose spark version 3.3.0 or above. Note that the official package from Apache Spark does not include hadoop-cloud component. We provide a Spark package with Hadoop cloud dependencies, download it from https://dmetasoul-bucket.obs.cn-southwest-2.myhuaweicloud.com/releases/spark/spark-3.3.2-bin-hadoop-3.3.5.tgz.

After unpacking spark package, you could find LakeSoul distribution jar from https://github.com/meta-soul/LakeSoul/releases. Download the jar file put it into `jars` directory of your spark environment.

```bash
wget https://dmetasoul-bucket.obs.cn-southwest-2.myhuaweicloud.com/releases/spark/spark-3.3.2-bin-hadoop-3.3.5.tgz
tar xf spark-3.3.2-bin-hadoop-3.3.5.tgz
export SPARK_HOME=${PWD}/spark-3.3.2-bin-dmetasoul
wget https://github.com/meta-soul/LakeSoul/releases/download/v2.2.0/lakesoul-spark-2.2.0-spark-3.3.jar -P $SPARK_HOME/jars
```

>For production deployment on Hadoop, it's recommended to use spark release without bundled hadoop:https://dlcdn.apache.org/spark/spark-3.3.2/spark-3.3.2-bin-without-hadoop.tgz
>
>Refer to https://spark.apache.org/docs/latest/hadoop-provided.html on how to setup hadoop classpath.
>
>Since 2.1.0, LakeSoul package all its dependencies into one single jar via maven shade plugin. Before that all jars were packaged into one tar.gz file.

##### 3. Adding PG database configuration for LakeSoul

By default, the pg database is connected to the local database, with the following configuration information:

```txt
lakesoul.pg.driver=com.lakesoul.shaded.org.postgresql.Driver
lakesoul.pg.url=jdbc:postgresql://127.0.0.1:5432/lakesoul_test?stringtype=unspecified
lakesoul.pg.username=lakesoul_test
lakesoul.pg.password=lakesoul_test 
```

Customizing PG database configuration information requires adding an environment variable `lakesoul_home` and importing configuration file information . If the PG database configuration information file path name is : /opt/soft/pg. property, then this environment variable needs to be added before the program starts:

```bash
export lakesoul_home=/opt/soft/pg.property
```

Users can customize the database configuration information here, so that the customized PG DB configuration information will take effect in Spark jobs.

##### 4. Starting the Spark environment

Start a `spark-sql` SQL interactive query command line environment:

```bash
$SPARK_HOME/bin/spark-sql --conf spark.sql.extensions=com.dmetasoul.lakesoul.sql.LakeSoulSparkSessionExtension --conf spark.sql.catalog.lakesoul=org.apache.spark.sql.lakesoul.catalog.LakeSoulCatalog --conf spark.sql.defaultCatalog=lakesoul --conf spark.sql.warehouse.dir=/tmp/lakesoul --conf spark.dmetasoul.lakesoul.snapshot.cache.expire.seconds=10
```

>This command starts a Spark local job, adding two options:
>
>1. spark.sql.warehouse.dir=/tmp/lakesoul This parameter is set because the default table storage location in Spark SQL needs to be set to the same directory as the Flink job output directory.
>2. spark.dmetasoul.lakesoul.snapshot.cache.expire.seconds=10 This parameter is set because LakeSoul caches metadata information in Spark, setting a smaller cache expiration time to facilitate querying the latest data.

After starting the Spark SQL command line, you can execute:

```sql
SHOW DATABASES;
SHOW TABLES IN default;
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xmxllnrhcp4du76ag2l2.png)

You can see that there is currently only one `default` database in LakeSoul, and there are no tables in it.

#### 3.1.3 Create a table in MySQL in advance and write data

1. Install mycli

```bash
pip install mycli
```

2. Start mycli and connect to the MySQL database

```bash
 mycli mysql://root@localhost:3306/test_cdc -p root
```

3. Create table and write data

```sql
 CREATE TABLE mysql_test_1 (id INT PRIMARY KEY, name VARCHAR(255), type SMALLINT);
 INSERT INTO mysql_test_1 VALUES (1, 'Bob', 10);
 SELECT * FROM mysql_test_1;
```

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/su0cajdhlnhoc57qul1c.png)

### 3.2 Start the sync job

#### 3.2.1 Start a local Flink Cluster

You can download [Flink 1.14.5](https://archive.apache.org/dist/flink/flink-1.14.5/flink-1.14.5-bin-scala_2.12.tgz) from the Flink download page.

Unzip the downloaded Flink installation package:

```bash
tar xf flink-1.14.5-bin-scala_2.12.tgz
export FLINK_HOME=${PWD}/flink-1.14.5
```

Then start a local Flink Cluster:

```bash
$FLINK_HOME/bin/start-cluster.sh
```

You can open http://localhost:8081 to see if the Flink local cluster has started normally:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/f1t7pe0bmud2aimwhs47.png)

#### 3.2.2 Submit LakeSoul Flink CDC Sink job

Submit a LakeSoul Flink CDC Sink job to the Flink cluster started above:

```bash
./bin/flink run -ys 1 -yjm 1G -ytm 2G \
   -c org.apache.flink.lakesoul.entry.MysqlCdc \
   lakesoul-flink-2.2.0-flink-1.14.jar \
   --source_db.host localhost \
   --source_db.port 3306 \
   --source_db.db_name test_cdc \
   --source_db.user root \
   --source_db.password root \
   --source.parallelism 1 \
   --sink.parallelism 1 \
   --warehouse_path file:/tmp/lakesoul \
   --flink.checkpoint file:/tmp/flink/chk \
   --flink.savepoint file:/tmp/flink/svp \
   --job.checkpoint_interval 10000 \
   --server_time_zone UTC
```

The jar package of lakesoul-flink can be downloaded from the [Github Release](https://github.com/meta-soul/LakeSoul/releases/) page.

Refer to [LakeSoul Flink CDC Synchronization of Entire MySQL Database](https://www.dmetasoul.com/en/docs/lakesoul/Usage Doc/flink-cdc-sync/) for detailed usage of the Flink job.

On the http://localhost:8081 Flink job page, click Running Job to check whether the LakeSoul job is already in the `Running` state.


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/cezj4a311co7qw23aqgv.png)

You can click to enter the job page, and you should see that one data record has been synchronized:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/io4kexh7xpy6zoltqvaf.png)
#### 3.2.3 Use Spark SQL to read the synchronized data in the LakeSoul table

Execute in Spark SQL Shell:

```sql
SHOW DATABASES;
SHOW TABLES IN test_cdc;
DESC test_cdc.mysql_test_1;
SELECT * FROM test_cdc.mysql_test_1;
```

You can see the running result of each statement, that is, a `test_cdc` database is automatically created in LakeSoul, and a `mysql_test_1` table is automatically created. The fields and primary keys of the table are the same as those of MySQL (one more rowKinds column)

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/nzrehaix2oj0zpymkj0d.png)
#### 3.2.4 Observe the synchronization situation after executing Update in MySQL

Perform the update in mycli:

```sql
UPDATE mysql_test_1 SET name='Peter' WHERE id=1;
```

Then read again in LakeSoul:

```sql
SELECT * FROM test_cdc.mysql_test_1;
```

You can see that the latest data has been read:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ymnag12uk463gvpfi81j.png)

#### 3.2.5 Observe the synchronization after executing DDL in MySQL, and read new and old data

Modify the structure of the table in mycli:

```sql
ALTER TABLE mysql_test_1 ADD COLUMN new_col FLOAT;
```

That is to add a new column at the end, the default is null. Verify the execution result in mycli:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/rfnr3y3i2onnxr8a7w0a.png)

At this point, the table structure has been synchronized in LakeSoul, and we can view the table structure in spark-sql:

```sql
DESC test_cdc.mysql_test_1;
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/icxeibn0wim0wj7v7kjs.png)

At this time, read data from LakeSoul, and the new column is also null:

```sql
SELECT * FROM test_cdc.mysql_test_1;
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hn5lkn2b2lxc70gkrs2g.png)

Insert a new piece of data into MySQL:

```sql
INSERT INTO mysql_test_1 VALUES (2, 'Alice', 20, 9.9);
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5qk7z9wptqa81i6tdnhe.png)

Read again from LakeSoul:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/nytq4dkt6dtyywh2v4xu.png)

Delete a piece of data from MySQL:

```sql
delete from mysql_test_1 where id=1;
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/og0u93rdqnm3twin6vop.png)

Read from LakeSoul:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/sw788zzpn652mljkgz5r.png)
**You can see that LakeSoul reads the synchronized result every time, which is exactly the same as in MySQL.**

#### 3.2.6 Observe the synchronization after creating a new table in MySQL

Create a new table in MySQL with a different schema from the previous table:

```sql
CREATE TABLE mysql_test_2 (name VARCHAR(100) PRIMARY KEY, phone_no VARCHAR(20));
```


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ewr3hubrse2bog7pkbo5.png)

Create a new table in MySQL with a different schema from the previous table:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/n0y8qswu02r0gocui2l5.png)

Insert a piece of data into a new MySQL table:

```sql
INSERT INTO mysql_test_2 VALUES ('Bob', '10010');
```

LakeSoul also successfully synchronized and read the data of the new table:


![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1ei2u8mk4tdsdxpu4jwx.png)

## 4. Conclusion

LakeSoul Flink CDC supports synchronizing whole databases from MySQL data sources to LakeSoul; Support automatic synchronization of schema changes (DDL) to LakeSoul; Support automatic perception of new tables in the upstream database during operation, and automatic table creation in LakeSoul; Support Exactly Once semantics, even if a Flink job fails, it can ensure that the data is not lost or heavy.