Apache Flink CDC（Change Data Capture）是一種用于從數據庫中捕獲變化（增刪改）的工具，可以將基于數據庫的實時應用與Apache Flink進行集成。在本文中，我們將介紹如何使用Flink CDC從MySQL數據庫讀取數據。

在開始之前，請確保已經安裝了Flink和MySQL數據庫，并設置了相應的權限和配置。

bin/start-cluster.sh # 啟動Flink集群
mysql -u username -p # 進入MySQL控制臺

首先，在MySQL數據庫中創建一個測試表，以便我們可以使用Flink CDC讀取其數據。

CREATE TABLE test (
id INT(11) PRIMARY KEY,
name VARCHAR(255),
age INT(11)
);

接下來，使用Flink CDC讀取MySQL表中的數據。首先，我們需要將Flink CDC依賴項添加到項目中。

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.12</artifactId>
<version>1.12.2</version>
</dependency>

然后，在Flink的代碼中，我們可以使用以下代碼來創建一個Flink JDBC讀取器并從MySQL表中讀取數據。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000);
Properties props = new Properties();
props.setProperty("url", "jdbc:mysql://localhost:3306/test");
props.setProperty("user", "username");
props.setProperty("password", "password");
FlinkJdbcReaderreader = FlinkJdbcReader.builder()
.setDataSourceProvider(new SimpleDataSourceProvider(props))
.setRowMapper(new StringRowMapper())
.setQuery("SELECT * FROM test")
.build();
DataStreamSourcesource = env.createInput(reader);
source.print();

此代碼將從MySQL表中讀取所有行并將其作為字符串打印到控制臺上。

在此基礎上，您可以使用Flink的一些其他功能來轉換、聚合或計算數據。例如，您可以使用Flink的窗口操作來計算每個小時的平均年齡，或使用Flink的SQLAPI來執行SQL查詢。

總之，使用Flink CDC從MySQL讀取數據非常簡單。在Flink和MySQL的幫助下，您可以將基于數據庫的實時應用與Flink進行集成，從而獲得更強大和更靈活的實時數據處理能力。