大數據實時分析：Flink 連接 Kafka 和 Flink SQL

作者：demo123567 2024-06-03 08:26:35

數據庫其他數據庫

Flink 就會從 Kafka 的 input_topic? 主題中讀取數據，每隔五分鐘按定義的 SQL 查詢進行處理，并將結果寫入 output_topic 主題。

Flink 連接 Kafka 前的準備

在使用 Apache Flink 連接 Apache Kafka 之前，需要完成以下準備工作。具體步驟如下：

從 Maven 官方庫獲取相關的 jar

選擇合適的 Kafka 連接器版本

根據我們使用的 Flink 版本選擇合適的 Kafka 連接器版本。官方建議的版本可以在 Flink 的官方文檔中找到。

添加 Maven 依賴

打開我們的項目的 pom.xml 文件，并添加以下依賴（假設我們使用的是 Flink 1.13 和 Kafka 2.8.0）：

<dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-kafka_2.12</artifactId>
       <version>1.13.0</version>
   </dependency>

下載 jar 文件

在命令行中運行 mvn clean package 下載依賴的 jar 文件。

將 jar 放到 lib 目錄下

找到下載的 jar 文件

運行 Maven 命令后，相關的 jar 文件會被下載到本地的 Maven 倉庫中，通常位。
于 ~/.m2/repository/org/apache/flink/ 下。

復制 jar 文件到 Flink 的 lib 目錄

找到相關的 jar 文件并將其復制到 Flink 的 lib 目錄中。假設 Flink 安裝在 /opt/flink 路徑下，執行以下命令：

cp ~/.m2/repository/org/apache/flink/flink-connector-kafka_2.12/1.13.0/flink-connector-kafka_2.12-1.13.0.jar /opt/flink/lib/

重啟 Flink

停止 Flink 集群

執行以下命令停止 Flink 集群：

/opt/flink/bin/stop-cluster.sh

啟動 Flink 集群

執行以下命令啟動 Flink 集群：

/opt/flink/bin/start-cluster.sh

完成上述步驟后，Flink 將能夠連接并消費 Kafka 的消息。

Flink連接Kafka的例子

在 Apache Flink 中，通過 Flink SQL 從 Kafka 中讀取數據，通常需要以下幾個步驟：

定義 Kafka 數據源表

使用 SQL 語句定義一個 Kafka 表，該表描述了如何從 Kafka 主題中讀取數據以及數據的格式。

執行 SQL 查詢

編寫 SQL 查詢來處理從 Kafka 讀取的數據。下面是一個詳細的示例，演示如何通過 Flink SQL 從 Kafka 中讀取數據：

定義 Kafka 數據源表

首先，我們需要定義一個 Kafka 表。假設我們有一個 Kafka 主題 input_topic，它包含 JSON 格式的數據。我們可以使用 CREATE TABLE 語句來定義這張表。

CREATE TABLE input_table (
  user_id STRING,
  action STRING,
  timestamp TIMESTAMP(3),
  WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'input_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink_consumer_group',
  'format' = 'json'
);

編寫 SQL 查詢

定義好 Kafka 表后，我們可以編寫 SQL 查詢來處理從 Kafka 中讀取的數據。例如，我們可以計算每個用戶的操作次數，并將結果插入到另一個 Kafka 主題。

CREATE TABLE output_table (
  user_id STRING,
  action_count BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'output_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);
INSERT INTO output_table
SELECT user_id, COUNT(action) AS action_count
FROM input_table
GROUP BY user_id, TUMBLE(timestamp, INTERVAL '10' MINUTE);

詳細解釋

input_table

user_id 和 action 是讀取自 Kafka 消息的字段。

timestamp 是事件時間戳，用于時間語義。

WATERMARK 用于處理遲到的數據，定義了一個 watermark 策略，表示事件時間戳延遲 5 秒。

WITH 子句定義了 Kafka 連接器的配置，包括 Kafka 主題名、服務器地址、消費者組 ID 和消息格式。

output_table
定義了一個輸出表，將結果寫回 Kafka 的 output_topic 主題。
配置與 input_table 類似，定義了 Kafka 連接器的屬性。
SQL 查詢
使用 INSERT INTO ... SELECT ... 語句從 input_table 讀取數據，并將處理結果寫入 output_table。
使用 TUMBLE 函數定義了一個 10 分鐘的滾動窗口，按 user_id 進行分組并計算每個用戶的操作次數。

運行 SQL 查詢

上述 SQL 查詢可以通過 Flink SQL CLI、Flink SQL 程序或 Flink SQL 任務提交工具來運行。以下是通過 Flink SQL CLI 運行這些查詢的步驟：

啟動 Flink 集群。
進入 Flink SQL CLI：

./bin/sql-client.sh

在 SQL CLI 中執行上述 CREATE TABLE 和 INSERT INTO 語句。

這樣，Flink 就會開始從 Kafka 的 input_topic 主題中讀取數據，按定義的 SQL 查詢進行處理，并將結果寫入 output_topic 主題。

Flink連接Kafka-帶有時間屬性

在 Apache Flink SQL 中，可以使用窗口函數來從 Kafka 中每隔五分鐘取一次數據并進行分析。下面是一個詳細的示例，展示了如何定義一個 Kafka 數據源表，并使用滾動窗口（Tumbling Window）來每五分鐘進行一次數據聚合分析。

定義 Kafka 數據源表

首先，需要定義一個 Kafka 表，該表描述了如何從 Kafka 主題中讀取數據以及數據的格式。

CREATE TABLE input_table (
  user_id STRING,
  action STRING,
  timestamp TIMESTAMP(3),
  WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'input_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'flink_consumer_group',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

定義結果表

接下來，需要定義一個輸出表，用于存儲分析結果。這里假設我們將結果寫回到另一個 Kafka 主題。

CREATE TABLE output_table (
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  user_id STRING,
  action_count BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'output_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

編寫 SQL 查詢

然后，編寫 SQL 查詢來從 Kafka 表中每隔五分鐘取一次數據并進行聚合分析。使用 TUMBLE 窗口函數來定義一個滾動窗口。

INSERT INTO output_table
SELECT
  TUMBLE_START(timestamp, INTERVAL '5' MINUTE) AS window_start,
  TUMBLE_END(timestamp, INTERVAL '5' MINUTE) AS window_end,
  user_id,
  COUNT(action) AS action_count
FROM input_table
GROUP BY
  TUMBLE(timestamp, INTERVAL '5' MINUTE),
  user_id;

詳細解釋

input_table

user_id 和 action 是從 Kafka 消息中讀取的字段。

timestamp 是事件時間戳，用于定義時間窗口。

WATERMARK 定義了一個 watermark 策略，允許事件時間戳延遲 5 秒。

WITH 子句定義了 Kafka 連接器的配置，包括 Kafka 主題名、服務器地址、消費者組 ID、啟動模式和消息格式。

output_table
定義了一個輸出表，將結果寫回 Kafka 的 output_topic 主題。
配置與 input_table 類似，定義了 Kafka 連接器的屬性。
SQL 查詢
使用 INSERT INTO ... SELECT ... 語句從 input_table 讀取數據，并將處理結果寫入 output_table。
TUMBLE 函數定義了一個 5 分鐘的滾動窗口。
TUMBLE_START 和 TUMBLE_END 函數分別返回窗口的開始時間和結束時間。
按 user_id 進行分組，并計算每個用戶在每個 5 分鐘窗口內的操作次數。

運行 SQL 查詢

這些 SQL 查詢可以通過 Flink SQL CLI、Flink SQL 程序或 Flink SQL 任務提交工具來運行。以下是通過 Flink SQL CLI 運行這些查詢的步驟：

啟動 Flink 集群。
進入 Flink SQL CLI：

./bin/sql-client.sh

在 SQL CLI 中執行上述 CREATE TABLE 和 INSERT INTO 語句。

這樣，Flink 就會從 Kafka 的 input_topic 主題中讀取數據，每隔五分鐘按定義的 SQL 查詢進行處理，并將結果寫入 output_topic 主題。

責任編輯：武曉燕來源：海燕技術棧