大数据实时分析：Flink 连接 Kafka 和 Flink SQL

Flink 连接 Kafka 前的准备

在使用 Apache Flink 连接 Apache Kafka 之前，需要完成以下准备工作。具体步骤如下：

从 Maven 官方库获取相关的 jar

选择合适的 Kafka 连接器版本

• 根据我们使用的 Flink 版本选择合适的 Kafka 连接器版本。官方建议的版本可以在 Flink 的官方文档中找到。

添加 Maven 依赖

• 打开我们的项目的 pom.xml 文件，并添加以下依赖（假设我们使用的是 Flink 1.13 和 Kafka 2.8.0）：

       org.apache.flink
       flink-connector-kafka_2.12
       1.13.0
   

下载 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 运行这些查询的步骤：

./bin/sql-client.sh

这样，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 运行这些查询的步骤：

./bin/sql-client.sh

这样，Flink 就会从 Kafka 的 input_topic 主题中读取数据，每隔五分钟按定义的 SQL 查询进行处理，并将结果写入 output_topic 主题。