flink sql cdc怎么使用

本篇內(nèi)容主要講解“flink sql cdc怎么使用”，感興趣的朋友不妨來看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“flink sql cdc怎么使用”吧!

前言

CDC,Change Data Capture,變更數(shù)據(jù)獲取的簡(jiǎn)稱，使用CDC我們可以從數(shù)據(jù)庫(kù)中獲取已提交的更改并將這些更改發(fā)送到下游，供下游使用。這些變更可以包括INSERT,DELETE,UPDATE等.

用戶可以在如下的場(chǎng)景使用cdc：

• 實(shí)時(shí)數(shù)據(jù)同步：比如我們將mysql庫(kù)中的數(shù)據(jù)同步到我們的數(shù)倉(cāng)中。

• 數(shù)據(jù)庫(kù)的實(shí)時(shí)物化視圖。

flink消費(fèi)cdc數(shù)據(jù)

在以前的數(shù)據(jù)同步中，比如我們想實(shí)時(shí)獲取數(shù)據(jù)庫(kù)的數(shù)據(jù)，一般采用的架構(gòu)就是采用第三方工具，比如canal、debezium等，實(shí)時(shí)采集數(shù)據(jù)庫(kù)的變更日志，然后將數(shù)據(jù)發(fā)送到kafka等消息隊(duì)列。然后再通過其他的組件，比如flink、spark等等來消費(fèi)kafka的數(shù)據(jù)，計(jì)算之后發(fā)送到下游系統(tǒng)。整體的架構(gòu)如下所示：

對(duì)于上面的這種架構(gòu)，flink承擔(dān)的角色是計(jì)算層，目前flink提供的format有兩種格式：canal-json和debezium-json，下面我們簡(jiǎn)單的介紹下。

canal format

在國(guó)內(nèi)，用的比較多的是阿里巴巴開源的canal，我們可以使用canal訂閱mysql的binlog日志，canal會(huì)將mysql庫(kù)的變更數(shù)據(jù)組織成它固定的JSON或protobuf 格式發(fā)到kafka，以供下游使用。

canal解析后的json數(shù)據(jù)格式如下：

{
  "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"
}

簡(jiǎn)單講下幾個(gè)核心的字段:

• type : 描述操作的類型，包括‘UPDATE’, 'INSERT', 'DELETE'。

• data : 代表操作的數(shù)據(jù)。如果為'INSERT'，則表示行的內(nèi)容；如果為'UPDATE'，則表示行的更新后的狀態(tài)；如果為'DELETE'，則表示刪除前的狀態(tài)。

• old ：可選字段，如果存在，則表示更新之前的內(nèi)容，如果不是update操作，則為 null。

完整的語(yǔ)義如下;

    private String                    destination;                            // 對(duì)應(yīng)canal的實(shí)例或者M(jìn)Q的topic
    private String                    groupId;                                // 對(duì)應(yīng)mq的group id
    private String                    database;                               // 數(shù)據(jù)庫(kù)或schema
    private String                    table;                                  // 表名
    private List<String>              pkNames;
    private Boolean                   isDdl;
    private String                    type;                                   // 類型: INSERT UPDATE DELETE
    // binlog executeTime
    private Long                      es;                                     // 執(zhí)行耗時(shí)
    // dml build timeStamp
    private Long                      ts;                                     // 同步時(shí)間
    private String                    sql;                                    // 執(zhí)行的sql, dml sql為空
    private List<Map<String, Object>> data;                                   // 數(shù)據(jù)列表
    private List<Map<String, Object>> old;                                    // 舊數(shù)據(jù)列表, 用于update, size和data的size一一對(duì)應(yīng)

在flink sql中，消費(fèi)這個(gè)數(shù)據(jù)的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了。

debezium format

在國(guó)外，比較有名的類似canal的開源工具有debezium，它的功能較canal更加強(qiáng)大一些，不僅僅支持mysql。還支持其他的數(shù)據(jù)庫(kù)的同步，比如 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來消費(fèi)的時(shí)候，sql和上面使用canal類似，只需要把foramt改成debezium-json即可。

CanalJson反序列化源碼解析

接下來我們看下flink的源碼中canal-json格式的實(shí)現(xiàn)。canal 格式作為一種flink的格式，而且是source，所以也就是涉及到讀取數(shù)據(jù)的時(shí)候進(jìn)行反序列化，我們接下來就簡(jiǎn)單看看CanalJson的反序列化的實(shí)現(xiàn)。具體的實(shí)現(xiàn)類是CanalJsonDeserializationSchema。

我們看下這個(gè)最核心的反序列化方法：

	@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數(shù)組表示的是要插入的數(shù)據(jù)，則循環(huán)遍歷data，然后添加一個(gè)標(biāo)識(shí)INSERT，構(gòu)造RowData對(duì)象，發(fā)送下游。
				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字段里獲取更新后的數(shù)據(jù)、
				ArrayData data = row.getArray(0);
				// old字段獲取更新之前的數(shù)據(jù)
				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字段非空，則說明進(jìn)行了數(shù)據(jù)的更新，如果old字段是null，則說明更新前后數(shù)據(jù)一樣，這個(gè)時(shí)候把before的數(shù)據(jù)也設(shè)置成after的，也就是發(fā)送給下游的before和after數(shù)據(jù)一樣。
before.setField(f, after.getField(f));
						}
					}
before.setRowKind(RowKind.UPDATE_BEFORE);
after.setRowKind(RowKind.UPDATE_AFTER);
					//把更新前后的數(shù)據(jù)都發(fā)送下游
out.collect(before);
out.collect(after);
				}
			} else if (OP_DELETE.equals(type)) {
				// 如果是刪除操作，data字段里包含將要被刪除的數(shù)據(jù)，把這些數(shù)據(jù)組織起來發(fā)送給下游
				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);
			}
		}
	}

flink cdc connector

背景

對(duì)于上面的架構(gòu)，我們需要部署canal（debezium）+ kafka，然后flink再?gòu)膋afka消費(fèi)數(shù)據(jù)，這種架構(gòu)下我們需要部署多個(gè)組件，并且數(shù)據(jù)也需要落地到kafka，有沒有更好的方案來精簡(jiǎn)下這個(gè)流程呢？我們接下來講講flink提供的cdc connector。

這個(gè)connector并沒有包含在flink的代碼里，具體的地址是在https://github.com/ververica/flink-cdc-connectors里，詳情大家可以看下這里面的內(nèi)容。

這種架構(gòu)下，flink直接消費(fèi)數(shù)據(jù)庫(kù)的增量日志，替代了原來作為數(shù)據(jù)采集層的canal（debezium），然后直接進(jìn)行計(jì)算，經(jīng)過計(jì)算之后，將計(jì)算結(jié)果 發(fā)送到下游。整體架構(gòu)如下：

使用這種架構(gòu)是好處有：

• 減少canal和kafka的維護(hù)成本，鏈路更短，延遲更低

• flink提供了exactly once語(yǔ)義

• 可以從指定position讀取

• 去掉了kafka，減少了消息的存儲(chǔ)成本

mysql-cdc

目前flink支持兩種內(nèi)置的connector，PostgreSQL和mysql，接下來我們以mysql為例簡(jiǎn)單講講。

在使用之前，我們需要引入相應(yīng)的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數(shù)據(jù)庫(kù)的示例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數(shù)據(jù)庫(kù)，我們需要把connector替換成postgres-cdc，DDL中表的schema和數(shù)據(jù)庫(kù)一一對(duì)應(yīng)。

更加詳細(xì)的配置參見：

https://github.com/ververica/flink-cdc-connectors/wiki/MySQL-CDC-Connector

mysql-cdc connector源碼解析

接下來我們以mysql-cdc為例，看看源碼層級(jí)是怎么實(shí)現(xiàn)的。既然作為一個(gè)sql的connector，那么就首先會(huì)有一個(gè)對(duì)應(yīng)的TableFactory，然后在工廠類里面構(gòu)造相應(yīng)的source，最后將消費(fèi)下來的數(shù)據(jù)轉(zhuǎn)成flink認(rèn)識(shí)的RowData格式，發(fā)送到下游。

我們按照這個(gè)思路來看看flink cdc源碼的實(shí)現(xiàn)。

在flink-connector-mysql-cdc module中，找到其對(duì)應(yīng)的工廠類：MySQLTableSourceFactory，進(jìn)入createDynamicTableSource(Context context)方法，在這個(gè)方法里，使用從ddl中的屬性里獲取的host、dbname等信息構(gòu)造了一個(gè)MySQLTableSource類。

MySQLTableSource

在MySQLTableSource#getScanRuntimeProvider方法里，我們看到，首先構(gòu)造了一個(gè)用于序列化的對(duì)象RowDataDebeziumDeserializeSchema，這個(gè)對(duì)象主要是用于將Debezium獲取的SourceRecord格式的數(shù)據(jù)轉(zhuǎn)化為flink認(rèn)識(shí)的RowData對(duì)象。 我們看下RowDataDebeziumDeserializeSchem#deserialize方法，這里的操作主要就是先判斷下進(jìn)來的數(shù)據(jù)類型（insert 、update、delete），然后針對(duì)不同的類型（short、int等）分別進(jìn)行轉(zhuǎn)換，

最后我們看到用于flink用于獲取數(shù)據(jù)庫(kù)變更日志的Source函數(shù)是DebeziumSourceFunction，且最終返回的類型是RowData。

也就是說flink底層是采用了Debezium工具從mysql、postgres等數(shù)據(jù)庫(kù)中獲取的變更數(shù)據(jù)。

	@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

我們接下來看看DebeziumSourceFunction類

@PublicEvolving
public class DebeziumSourceFunction<T> extends RichSourceFunction<T> implements
CheckpointedFunction,
ResultTypeQueryable<T> {
.............
		}

我們看到DebeziumSourceFunction類繼承了RichSourceFunction，并且實(shí)現(xiàn)了CheckpointedFunction接口，也就是說這個(gè)類是flink的一個(gè)SourceFunction，會(huì)從源端（run方法）獲取數(shù)據(jù)，發(fā)送給下游。此外這個(gè)類還實(shí)現(xiàn)了CheckpointedFunction接口，也就是會(huì)通過checkpoint的機(jī)制來保證exactly once語(yǔ)義。

接下來我們進(jìn)入run方法，看看是如何獲取數(shù)據(jù)庫(kù)的變更數(shù)據(jù)的。

	@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);

//用于具體的處理數(shù)據(jù)的邏輯
		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)  // 數(shù)據(jù)發(fā)給上面的debeziumConsumer
			.using((success, message, error) -> {
if (!success && error != null) {
					this.reportError(error);
				}
			})
			.build();

if (!running) {
return;
		}

// run the engine asynchronously
		executor.execute(engine);

        //循環(huán)判斷，當(dāng)程序被打斷，或者有錯(cuò)誤的時(shí)候，打斷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();
		}
	}

在函數(shù)的開始，設(shè)置了很多的properties，比如include.schema.changes 設(shè)置為false，也就是不包含表的DDL操作，表結(jié)構(gòu)的變更是不捕獲的。我們這里只關(guān)注數(shù)據(jù)的增刪改。

接下來構(gòu)造了一個(gè)DebeziumChangeConsumer對(duì)象，這個(gè)類實(shí)現(xiàn)了DebeziumEngine.ChangeConsumer接口，主要就是將獲取到的一批數(shù)據(jù)進(jìn)行一條條的加工處理。

接下來定一個(gè)DebeziumEngine對(duì)象，這個(gè)對(duì)象是真正用來干活的，它的底層使用了kafka的connect-api來進(jìn)行獲取數(shù)據(jù)，得到的是一個(gè)org.apache.kafka.connect.source.SourceRecord對(duì)象。通過notifying方法將得到的數(shù)據(jù)交給上面定義的DebeziumChangeConsumer來來覆蓋缺省實(shí)現(xiàn)以進(jìn)行復(fù)雜的操作。

接下來通過一個(gè)線程池ExecutorService來異步的啟動(dòng)這個(gè)engine。

最后，做了一個(gè)循環(huán)判斷，當(dāng)程序被打斷，或者有錯(cuò)誤的時(shí)候，打斷engine，并且拋出異常。

總結(jié)一下，就是在Flink的source函數(shù)里，使用Debezium 引擎獲取對(duì)應(yīng)的數(shù)據(jù)庫(kù)變更數(shù)據(jù)（SourceRecord），經(jīng)過一系列的反序列化操作，最終轉(zhuǎn)成了flink中的RowData對(duì)象，發(fā)送給下游。

changelog format

使用場(chǎng)景

當(dāng)我們從mysql-cdc獲取數(shù)據(jù)庫(kù)的變更數(shù)據(jù)，或者寫了一個(gè)group by的查詢的時(shí)候，這種結(jié)果數(shù)據(jù)都是不斷變化的，我們?nèi)绾螌⑦@些變化的數(shù)據(jù)發(fā)到只支持append mode的kafka隊(duì)列呢?

于是flink提供了一種changelog format，其實(shí)我們非常簡(jiǎn)單的理解為，flink對(duì)進(jìn)來的RowData數(shù)據(jù)進(jìn)行了一層包裝，然后加了一個(gè)數(shù)據(jù)的操作類型，包括以下幾種 INSERT,DELETE, UPDATE_BEFORE,UPDATE_AFTER。這樣當(dāng)下游獲取到這個(gè)數(shù)據(jù)的時(shí)候，就可以根據(jù)數(shù)據(jù)的類型來判斷下如何對(duì)數(shù)據(jù)進(jìn)行操作了。

比如我們的原始數(shù)據(jù)格式是這樣的。

{"day":"2020-06-18","gmv":100}

經(jīng)過changelog格式的加工之后，成為了下面的格式：

{"data":{"day":"2020-06-18","gmv":100},"op":"+I"}

也就是說changelog format對(duì)原生的格式進(jìn)行了包裝，添加了一個(gè)op字段，表示數(shù)據(jù)的操作類型，目前有以下幾種：

• +I：插入操作。

• -U ：更新之前的數(shù)據(jù)內(nèi)容：

• +U ：更新之后的數(shù)據(jù)內(nèi)容。

• -D ：刪除操作。

示例

使用的時(shí)候需要引入相應(yīng)的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'
);

我們定義了一個(gè) format 為 changelog-json 的kafka connector，之后我們就可以對(duì)其進(jìn)行寫入和查詢了。

完整的代碼和配置請(qǐng)參考：
https://github.com/ververica/flink-cdc-connectors/wiki/Changelog-JSON-Format

源碼淺析

作為一種flink的format ，我們主要看下其序列化和發(fā)序列化方法，changelog-json 使用了flink-json包進(jìn)行json的處理。

反序列化

反序列化用的是ChangelogJsonDeserializationSchema類，在其構(gòu)造方法里，我們看到主要是構(gòu)造了一個(gè)json的序列化器jsonDeserializer用于對(duì)數(shù)據(jù)進(jìn)行處理。

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();
	}

在這里，指定了這個(gè)row格式有兩個(gè)字段，一個(gè)是data，表示數(shù)據(jù)的內(nèi)容，一個(gè)是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對(duì)數(shù)據(jù)進(jìn)行處理，然后對(duì)第二個(gè)字段op進(jìn)行判斷，添加對(duì)應(yīng)的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序列化成字節(jié)數(shù)組。

到此，相信大家對(duì)“flink sql cdc怎么使用”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！