溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇內容主要講解“flink sql cdc怎么使用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“flink sql cdc怎么使用”吧!
CDC,Change Data Capture,變更數據獲取的簡稱，使用CDC我們可以從數據庫中獲取已提交的更改并將這些更改發送到下游，供下游使用。這些變更可以包括INSERT,DELETE,UPDATE等.
用戶可以在如下的場景使用cdc：
實時數據同步：比如我們將mysql庫中的數據同步到我們的數倉中。
數據庫的實時物化視圖。
在以前的數據同步中，比如我們想實時獲取數據庫的數據，一般采用的架構就是采用第三方工具，比如canal、debezium等，實時采集數據庫的變更日志，然后將數據發送到kafka等消息隊列。然后再通過其他的組件，比如flink、spark等等來消費kafka的數據，計算之后發送到下游系統。整體的架構如下所示：
對于上面的這種架構，flink承擔的角色是計算層，目前flink提供的format有兩種格式：canal-json和debezium-json，下面我們簡單的介紹下。
在國內，用的比較多的是阿里巴巴開源的canal，我們可以使用canal訂閱mysql的binlog日志，canal會將mysql庫的變更數據組織成它固定的JSON或protobuf 格式發到kafka，以供下游使用。
canal解析后的json數據格式如下：
{
"data": [
{
"id": "111",
"name": "scooter",
"description": "Big 2-wheel scooter",
"weight": "5.18"
}
],
"database": "inventory",
"es": 1589373560000,
"id": 9,
"isDdl": false,
"mysqlType": {
"id": "INTEGER",
"name": "VARCHAR(255)",
"description": "VARCHAR(512)",
"weight": "FLOAT"
},
"old": [
{
"weight": "5.15"
}
],
"pkNames": [
"id"
],
"sql": "",
"sqlType": {
"id": 4,
"name": 12,
"description": 12,
"weight": 7
},
"table": "products",
"ts": 1589373560798,
"type": "UPDATE"
}
簡單講下幾個核心的字段:
type : 描述操作的類型，包括‘UPDATE’, 'INSERT', 'DELETE'。
data : 代表操作的數據。如果為'INSERT'，則表示行的內容；如果為'UPDATE'，則表示行的更新后的狀態；如果為'DELETE'，則表示刪除前的狀態。
old ：可選字段，如果存在，則表示更新之前的內容，如果不是update操作，則為 null。
完整的語義如下;
private String destination; // 對應canal的實例或者MQ的topic
private String groupId; // 對應mq的group id
private String database; // 數據庫或schema
private String table; // 表名
private List<String> pkNames;
private Boolean isDdl;
private String type; // 類型: INSERT UPDATE DELETE
// binlog executeTime
private Long es; // 執行耗時
// dml build timeStamp
private Long ts; // 同步時間
private String sql; // 執行的sql, dml sql為空
private List<Map<String, Object>> data; // 數據列表
private List<Map<String, Object>> old; // 舊數據列表, 用于update, size和data的size一一對應
在flink sql中，消費這個數據的sql如下：
CREATE TABLE topic_products (
id BIGINT,
name STRING,
description STRING,
weight DECIMAL(10, 2)
) WITH (
'connector' = 'kafka',
'topic' = 'products_binlog',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'format' = 'canal-json' -- using canal-json as the format
)
其中DDL中的表的字段和類型要和mysql中的字段及類型能匹配的上，接下來我們就可以寫flink sql來查詢我們定義的topic_products了。
在國外，比較有名的類似canal的開源工具有debezium，它的功能較canal更加強大一些，不僅僅支持mysql。還支持其他的數據庫的同步，比如 PostgreSQL、Oracle等，目前debezium支持的序列化格式為 JSON 和 Apache Avro 。
debezium提供的格式如下：
{
"before": {
"id": 111,
"name": "scooter",
"description": "Big 2-wheel scooter",
"weight": 5.18
},
"after": {
"id": 111,
"name": "scooter",
"description": "Big 2-wheel scooter",
"weight": 5.15
},
"source": {...},
"op": "u",
"ts_ms": 1589362330904,
"transaction": null
}
同樣，使用flink sql來消費的時候，sql和上面使用canal類似，只需要把foramt改成debezium-json即可。
接下來我們看下flink的源碼中canal-json格式的實現。canal 格式作為一種flink的格式，而且是source，所以也就是涉及到讀取數據的時候進行反序列化，我們接下來就簡單看看CanalJson的反序列化的實現。具體的實現類是CanalJsonDeserializationSchema。
我們看下這個最核心的反序列化方法：
@Override
public void deserialize(byte[] message, Collector<RowData> out) throws IOException {
try {
//使用json反序列化器將message反序列化成RowData
RowData row = jsonDeserializer.deserialize(message);
//獲取type字段，用于下面的判斷
String type = row.getString(2).toString();
if (OP_INSERT.equals(type)) {
// 如果操作類型是insert，則data數組表示的是要插入的數據，則循環遍歷data，然后添加一個標識INSERT，構造RowData對象，發送下游。
ArrayData data = row.getArray(0);
for (int i = 0; i < data.size(); i++) {
RowData insert = data.getRow(i, fieldCount);
insert.setRowKind(RowKind.INSERT);
out.collect(insert);
}
} else if (OP_UPDATE.equals(type)) {
// 如果是update操作，從data字段里獲取更新后的數據、
ArrayData data = row.getArray(0);
// old字段獲取更新之前的數據
ArrayData old = row.getArray(1);
for (int i = 0; i < data.size(); i++) {
// the underlying JSON deserialization schema always produce GenericRowData.
GenericRowData after = (GenericRowData) data.getRow(i, fieldCount);
GenericRowData before = (GenericRowData) old.getRow(i, fieldCount);
for (int f = 0; f < fieldCount; f++) {
if (before.isNullAt(f)) {
//如果old字段非空，則說明進行了數據的更新，如果old字段是null，則說明更新前后數據一樣，這個時候把before的數據也設置成after的，也就是發送給下游的before和after數據一樣。
before.setField(f, after.getField(f));
}
}
before.setRowKind(RowKind.UPDATE_BEFORE);
after.setRowKind(RowKind.UPDATE_AFTER);
//把更新前后的數據都發送下游
out.collect(before);
out.collect(after);
}
} else if (OP_DELETE.equals(type)) {
// 如果是刪除操作，data字段里包含將要被刪除的數據，把這些數據組織起來發送給下游
ArrayData data = row.getArray(0);
for (int i = 0; i < data.size(); i++) {
RowData insert = data.getRow(i, fieldCount);
insert.setRowKind(RowKind.DELETE);
out.collect(insert);
}
} else {
if (!ignoreParseErrors) {
throw new IOException(format(
"Unknown \"type\" value \"%s\". The Canal JSON message is '%s'", type, new String(message)));
}
}
} catch (Throwable t) {
// a big try catch to protect the processing.
if (!ignoreParseErrors) {
throw new IOException(format(
"Corrupt Canal JSON message '%s'.", new String(message)), t);
}
}
}
對于上面的架構，我們需要部署canal（debezium）+ kafka，然后flink再從kafka消費數據，這種架構下我們需要部署多個組件，并且數據也需要落地到kafka，有沒有更好的方案來精簡下這個流程呢？我們接下來講講flink提供的cdc connector。
這個connector并沒有包含在flink的代碼里，具體的地址是在https://github.com/ververica/flink-cdc-connectors里，詳情大家可以看下這里面的內容。
這種架構下，flink直接消費數據庫的增量日志，替代了原來作為數據采集層的canal（debezium），然后直接進行計算，經過計算之后，將計算結果 發送到下游。整體架構如下：
使用這種架構是好處有：
減少canal和kafka的維護成本，鏈路更短，延遲更低
flink提供了exactly once語義
可以從指定position讀取
去掉了kafka，減少了消息的存儲成本
目前flink支持兩種內置的connector，PostgreSQL和mysql，接下來我們以mysql為例簡單講講。
在使用之前，我們需要引入相應的pom，mysql的pom如下：
<dependency>
<groupId>com.alibaba.ververica</groupId>
<!-- add the dependency matching your database -->
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>1.1.0</version>
</dependency>
如果是sql客戶端使用，需要下載 flink-sql-connector-mysql-cdc-1.1.0.jar 并且放到<FLINK_HOME>/lib/下面
連接mysql數據庫的示例sql如下：
CREATE TABLE mysql_binlog (
id INT NOT NULL,
name STRING,
description STRING,
weight DECIMAL(10,3)
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '3306',
'username' = 'flinkuser',
'password' = 'flinkpw',
'database-name' = 'inventory',
'table-name' = 'products'
)
如果訂閱的是postgres數據庫，我們需要把connector替換成postgres-cdc，DDL中表的schema和數據庫一一對應。
更加詳細的配置參見：
https://github.com/ververica/flink-cdc-connectors/wiki/MySQL-CDC-Connector
接下來我們以mysql-cdc為例，看看源碼層級是怎么實現的。既然作為一個sql的connector，那么就首先會有一個對應的TableFactory，然后在工廠類里面構造相應的source，最后將消費下來的數據轉成flink認識的RowData格式，發送到下游。
我們按照這個思路來看看flink cdc源碼的實現。
在flink-connector-mysql-cdc module中，找到其對應的工廠類：MySQLTableSourceFactory，進入createDynamicTableSource(Context context)方法，在這個方法里，使用從ddl中的屬性里獲取的host、dbname等信息構造了一個MySQLTableSource類。
在MySQLTableSource#getScanRuntimeProvider方法里，我們看到，首先構造了一個用于序列化的對象RowDataDebeziumDeserializeSchema，這個對象主要是用于將Debezium獲取的SourceRecord格式的數據轉化為flink認識的RowData對象。 我們看下RowDataDebeziumDeserializeSchem#deserialize方法，這里的操作主要就是先判斷下進來的數據類型（insert 、update、delete），然后針對不同的類型（short、int等）分別進行轉換，
最后我們看到用于flink用于獲取數據庫變更日志的Source函數是DebeziumSourceFunction，且最終返回的類型是RowData。
也就是說flink底層是采用了Debezium工具從mysql、postgres等數據庫中獲取的變更數據。
@SuppressWarnings("unchecked")
@Override
public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
RowType rowType = (RowType) physicalSchema.toRowDataType().getLogicalType();
TypeInformation<RowData> typeInfo = (TypeInformation<RowData>) scanContext.createTypeInformation(physicalSchema.toRowDataType());
DebeziumDeserializationSchema<RowData> deserializer = new RowDataDebeziumDeserializeSchema(
rowType,
typeInfo,
((rowData, rowKind) -> {}),
serverTimeZone);
MySQLSource.Builder<RowData> builder = MySQLSource.<RowData>builder()
.hostname(hostname)
..........
DebeziumSourceFunction<RowData> sourceFunction = builder.build();
return SourceFunctionProvider.of(sourceFunction, false);
}
我們接下來看看DebeziumSourceFunction類
@PublicEvolving
public class DebeziumSourceFunction<T> extends RichSourceFunction<T> implements
CheckpointedFunction,
ResultTypeQueryable<T> {
.............
}
我們看到DebeziumSourceFunction類繼承了RichSourceFunction，并且實現了CheckpointedFunction接口，也就是說這個類是flink的一個SourceFunction，會從源端（run方法）獲取數據，發送給下游。此外這個類還實現了CheckpointedFunction接口，也就是會通過checkpoint的機制來保證exactly once語義。
接下來我們進入run方法，看看是如何獲取數據庫的變更數據的。
@Override
public void run(SourceContext<T> sourceContext) throws Exception {
...........................
// DO NOT include schema change, e.g. DDL
properties.setProperty("include.schema.changes", "false");
...........................
//將所有的屬性信息打印出來，以便排查。
// dump the properties
String propsString = properties.entrySet().stream()
.map(t -> "\t" + t.getKey().toString() + " = " + t.getValue().toString() + "\n")
.collect(Collectors.joining());
LOG.info("Debezium Properties:\n{}", propsString);
//用于具體的處理數據的邏輯
this.debeziumConsumer = new DebeziumChangeConsumer<>(
sourceContext,
deserializer,
restoredOffsetState == null, // DB snapshot phase if restore state is null
this::reportError);
// create the engine with this configuration ...
this.engine = DebeziumEngine.create(Connect.class)
.using(properties)
.notifying(debeziumConsumer) // 數據發給上面的debeziumConsumer
.using((success, message, error) -> {
if (!success && error != null) {
this.reportError(error);
}
})
.build();
if (!running) {
return;
}
// run the engine asynchronously
executor.execute(engine);
//循環判斷，當程序被打斷，或者有錯誤的時候，打斷engine，并且拋出異常
// on a clean exit, wait for the runner thread
try {
while (running) {
if (executor.awaitTermination(5, TimeUnit.SECONDS)) {
break;
}
if (error != null) {
running = false;
shutdownEngine();
// rethrow the error from Debezium consumer
ExceptionUtils.rethrow(error);
}
}
}
catch (InterruptedException e) {
// may be the result of a wake-up interruption after an exception.
// we ignore this here and only restore the interruption state
Thread.currentThread().interrupt();
}
}
在函數的開始，設置了很多的properties，比如include.schema.changes 設置為false，也就是不包含表的DDL操作，表結構的變更是不捕獲的。我們這里只關注數據的增刪改。
接下來構造了一個DebeziumChangeConsumer對象，這個類實現了DebeziumEngine.ChangeConsumer接口，主要就是將獲取到的一批數據進行一條條的加工處理。
接下來定一個DebeziumEngine對象，這個對象是真正用來干活的，它的底層使用了kafka的connect-api來進行獲取數據，得到的是一個org.apache.kafka.connect.source.SourceRecord對象。通過notifying方法將得到的數據交給上面定義的DebeziumChangeConsumer來來覆蓋缺省實現以進行復雜的操作。
接下來通過一個線程池ExecutorService來異步的啟動這個engine。
最后，做了一個循環判斷，當程序被打斷，或者有錯誤的時候，打斷engine，并且拋出異常。
總結一下，就是在Flink的source函數里，使用Debezium 引擎獲取對應的數據庫變更數據（SourceRecord），經過一系列的反序列化操作，最終轉成了flink中的RowData對象，發送給下游。
當我們從mysql-cdc獲取數據庫的變更數據，或者寫了一個group by的查詢的時候，這種結果數據都是不斷變化的，我們如何將這些變化的數據發到只支持append mode的kafka隊列呢?
于是flink提供了一種changelog format，其實我們非常簡單的理解為，flink對進來的RowData數據進行了一層包裝，然后加了一個數據的操作類型，包括以下幾種 INSERT,DELETE, UPDATE_BEFORE,UPDATE_AFTER。這樣當下游獲取到這個數據的時候，就可以根據數據的類型來判斷下如何對數據進行操作了。
比如我們的原始數據格式是這樣的。
{"day":"2020-06-18","gmv":100}
經過changelog格式的加工之后，成為了下面的格式：
{"data":{"day":"2020-06-18","gmv":100},"op":"+I"}
也就是說changelog format對原生的格式進行了包裝，添加了一個op字段，表示數據的操作類型，目前有以下幾種：
+I：插入操作。
-U ：更新之前的數據內容：
+U ：更新之后的數據內容。
-D ：刪除操作。
使用的時候需要引入相應的pom
<dependency>
<groupId>com.alibaba.ververica</groupId>
<artifactId>flink-format-changelog-json</artifactId>
<version>1.1.0</version>
</dependency>
使用flink sql操作的方式如下：
CREATE TABLE kafka_gmv (
day_str STRING,
gmv DECIMAL(10, 5)
) WITH (
'connector' = 'kafka',
'topic' = 'kafka_gmv',
'scan.startup.mode' = 'earliest-offset',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'changelog-json'
);
我們定義了一個 format 為 changelog-json 的kafka connector，之后我們就可以對其進行寫入和查詢了。
完整的代碼和配置請參考：
https://github.com/ververica/flink-cdc-connectors/wiki/Changelog-JSON-Format
作為一種flink的format ，我們主要看下其序列化和發序列化方法，changelog-json 使用了flink-json包進行json的處理。
反序列化用的是ChangelogJsonDeserializationSchema類，在其構造方法里，我們看到主要是構造了一個json的序列化器jsonDeserializer用于對數據進行處理。
public ChangelogJsonDeserializationSchema(
RowType rowType,
TypeInformation<RowData> resultTypeInfo,
boolean ignoreParseErrors,
TimestampFormat timestampFormatOption) {
this.resultTypeInfo = resultTypeInfo;
this.ignoreParseErrors = ignoreParseErrors;
this.jsonDeserializer = new JsonRowDataDeserializationSchema(
createJsonRowType(fromLogicalToDataType(rowType)),
// the result type is never used, so it's fine to pass in Debezium's result type
resultTypeInfo,
false, // ignoreParseErrors already contains the functionality of failOnMissingField
ignoreParseErrors,
timestampFormatOption);
}
其中createJsonRowType方法指定了changelog的format是一種Row類型的格式，我們看下代碼：
private static RowType createJsonRowType(DataType databaseSchema) {
DataType payload = DataTypes.ROW(
DataTypes.FIELD("data", databaseSchema),
DataTypes.FIELD("op", DataTypes.STRING()));
return (RowType) payload.getLogicalType();
}
在這里，指定了這個row格式有兩個字段，一個是data，表示數據的內容，一個是op，表示操作的類型。
最后看下最核心的ChangelogJsonDeserializationSchema#deserialize(byte[] bytes, Collector<RowData> out>)
@Override
public void deserialize(byte[] bytes, Collector<RowData> out) throws IOException {
try {
GenericRowData row = (GenericRowData) jsonDeserializer.deserialize(bytes);
GenericRowData data = (GenericRowData) row.getField(0);
String op = row.getString(1).toString();
RowKind rowKind = parseRowKind(op);
data.setRowKind(rowKind);
out.collect(data);
} catch (Throwable t) {
// a big try catch to protect the processing.
if (!ignoreParseErrors) {
throw new IOException(format(
"Corrupt Debezium JSON message '%s'.", new String(bytes)), t);
}
}
}
使用jsonDeserializer對數據進行處理，然后對第二個字段op進行判斷，添加對應的RowKind。
序列化的方法我們看下方法：ChangelogJsonSerializationSchema#serialize
@Override
public byte[] serialize(RowData rowData) {
reuse.setField(0, rowData);
reuse.setField(1, stringifyRowKind(rowData.getRowKind()));
return jsonSerializer.serialize(reuse);
}
這塊沒有什么難度，就是將flink的RowData使用jsonSerializer序列化成字節數組。
到此，相信大家對“flink sql cdc怎么使用”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
