flinkCDC功(gong)能(neng)是面(mian)向binlog進行同步、對數(shu)(shu)據的增刪(shan)改(gai)進行同步的工具，能(neng)夠實現(xian)對數(shu)(shu)據的動態監(jian)聽。目(mu)前(qian)其(qi)實現(xian)原理主要為監(jian)聽數(shu)(shu)據源的binlog對數(shu)(shu)據的變化有所感知。
<!-- flink connector cdc -->
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>${flink.sql.connector.cdc.version}</version>
</dependency>
，在本實例中，選擇(ze)2.2.1版本的mysqlcdc進行演示。
| Flink® Version | |
|---|---|
| 1.0.0 | 1.11.* |
| 1.1.0 | 1.11.* |
| 1.2.0 | 1.12.* |
| 1.3.0 | 1.12.* |
| 1.4.0 | 1.13.* |
| 2.0.* | 1.13.* |
| 2.1.* | 1.13.* |
| 2.2.* | 1.13., 1.14. |
| 2.3.* | 1.13., 1.14., 1.15.*, 1.16.0 |
package cn.ctyun.demo.api.watermark;
import cn.ctyun.demo.api.utils.TransformUtil;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
public class ViewContentStreamWithoutWaterMark {
public static DataStream<JSONObject> getViewContentDataStream(StreamExecutionEnvironment env){
// 1.創建Flink-MySQL-CDC的Source
MySqlSource<String> viewContentSouce = MySqlSource.<String>builder()
.hostname("49.7.189.190")
.port(3307)
.username("root")
.password("Adm@163.comCdc")
.databaseList("test_cdc_source")
.tableList("test_cdc_source.view_content")
.startupOptions(StartupOptions.initial())
.deserializer(new JsonDebeziumDeserializationSchema())
.serverTimeZone("Asia/Shanghai")
.build();
// 2.使用CDC Source從MySQL讀取數據
DataStreamSource<String> mysqlDataStreamSource = env.fromSource(
viewContentSouce,
WatermarkStrategy.noWatermarks(),
"ViewContentStreamNoWatermark Source"
);
// 3.轉換為指定格式
return mysqlDataStreamSource.map(TransformUtil::formatResult);
}
}
flinksql操作，能夠簡化大量操作，具體如下代碼所示(shi)。在這里(li)我們只(zhi)需要提供簡單的sql語句即可(ke)完(wan)成對mysql數據源的動態讀取(qu)。通過指定連接器(qi)類(lei)型(xing)為'connector' = 'mysql-cdc'
package cn.ctyun.demo.flinksql;
import cn.ctyun.demo.flinksql.udf.HashScalarFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/**
* @classname: ReadFromCdc
* @description: 通過cdc獲取數據變化進行輸入
* @author: Liu Xinyuan
* @create: 2023-04-12 15:09
**/
public class FlinkSqlReadFromCdc {
public static void main(String[] args) throws Exception {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.disableOperatorChaining();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 1. 創建讀取表，使用mysql-cdc進行，注意此時應標記主鍵
String source_ddl = "CREATE TABLE UserSource (" +
" id INT, " +
" name VARCHAR, " +
" phone VARCHAR, " +
" sex INT, " +
" primary key (id) not enforced" +
") WITH (" +
" 'connector' = 'mysql-cdc'," +
" 'hostname' = '*******'," +
" 'port' = '3307'," +
" 'username' = '" + parameterTool.get("user") + "', " +
" 'password' = '" + parameterTool.get("passwd") + "'" +
" 'database-name' = 'test_cdc_source'," +
" 'table-name' = 'test_user_table'," +
" 'debezium.log.mining.continuous.mine'='true',"+
" 'debezium.log.mining.strategy'='online_catalog', " +
" 'debezium.database.tablename.case.insensitive'='false',"+
" 'jdbc.properties.useSSL' = 'false' ," +
" 'scan.startup.mode' = 'initial')";
tableEnv.executeSql(source_ddl);
// 2. 創建寫出表，使用mysql進行
String sink_ddl = "CREATE TABLE UserSink (" +
" id INT, " +
" name VARCHAR, " +
" phone VARCHAR, " +
" sex INT, " +
" primary key (id) not enforced" +
") WITH (" +
" 'connector.type' = 'jdbc', " +
" 'connector.url' = 'jdbc:mysql://******:3306/flink_test_sink?useSSL=false', " +
" 'connector.table' = 'test_user_table', " +
" 'connector.username' = '" + parameterTool.get("sinkUser") + "', " +
" 'connector.password' = '" + parameterTool.get("sinkPasswd") + "'" +
" 'connector.write.flush.max-rows' = '1'" +
")";
tableEnv.executeSql(sink_ddl);
// 3.簡單的數據清洗，將電話號碼進行hash掩碼
tableEnv.createTemporarySystemFunction("MyHASH", HashScalarFunction.class);
Table maskedTable = tableEnv.sqlQuery("SELECT id, name, MyHASH(phone) as phone, sex FROM UserSource");
tableEnv.createTemporaryView("MaskedUserInfo", maskedTable);
// 4.使用insert語句進行數據輸出,在這里進行一定地清洗
String insertSql = "INSERT INTO UserSink SELECT * FROM MaskedUserInfo";
TableResult tableResult = tableEnv.executeSql(insertSql);
tableResult.print();
}
}'scan.startup.mode' = 'initial'此處是cdc的(de)關(guan)鍵所在(zai)，MySQL CDC 消費者可選的(de)啟動模式， 合(he)法的(de)模式為(wei) "initial"，"earliest-offset"，"latest-offset"，"specific-offset" 和 "timestamp"
String source_ddl = "CREATE TABLE UserSource (" +
" id INT, " +
" name VARCHAR, " +
" phone VARCHAR, " +
" sex INT, " +
" primary key (id) not enforced" +
") WITH (" +
" 'connector' = 'mysql-cdc'," +
" 'hostname' = '******'," +
" 'port' = '3307'," +
" 'username' = '" + parameterTool.get("user") + "', " +
" 'password' = '" + parameterTool.get("passwd") + "'" +
" 'database-name' = 'test_cdc_source'," +
" 'table-name' = 'test_user_table'," +
" 'debezium.log.mining.continuous.mine'='true',"+
" 'debezium.log.mining.strategy'='online_catalog', " +
" 'debezium.database.tablename.case.insensitive'='false',"+
" 'jdbc.properties.useSSL' = 'false' ," +
" 'scan.startup.mode' = 'initial')";啟用(yong)后(hou)，整(zheng)個流程(cheng)為對其中的(de)數據增量(liang)(liang)(liang)同(tong)步(bu)，由于我們(men)使(shi)用(yong)的(de)是initial模式，因此我們(men)的(de)數據在任務啟動的(de)時候，首先進行了一(yi)次(ci)全量(liang)(liang)(liang)同(tong)步(bu)，全量(liang)(liang)(liang)地將(jiang)信息同(tong)步(bu)，并(bing)且進行了掩(yan)碼(ma)操作。
后續如(ru)果添(tian)加新的信(xin)息(xi)也(ye)會進行同(tong)步，刪除亦然。
