flink1.14sql根据分区读取kafka

在 Flink 1.14 版本中，使用 SQL API 从 Kafka 中读取数据并根据分区进行读取的步骤如下：

1. 首先，需要在 Flink 作业中添加对 Kafka 的依赖，例如：

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

2. 接着，在 Flink SQL 中创建一个 Kafka 表，并指定需要读取的 Kafka 主题、Kafka 服务器地址、分区等信息，例如：

CREATE TABLE kafka_table (
    id BIGINT,
    name STRING,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'my_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'my_group',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.timestamp-format.standard' = 'ISO-8601'
);

在上面的代码中，我们使用了 Flink SQL 的 `CREATE TABLE` 语句来创建一个名为 `kafka_table` 的表，并指定了需要从 Kafka 中读取的数据格式为 JSON 格式。

3. 接下来，可以使用 Flink SQL 中的 `SELECT` 语句来查询 Kafka 表中的数据，例如：

SELECT id, name, event_time
FROM kafka_table
WHERE _partition = 0;

在上面的代码中，我们使用了 `WHERE` 子句来筛选需要读取的分区，这里筛选了分区编号为 0 的数据。

4. 最后，可以使用 Flink SQL 的 `INSERT INTO` 语句将查询结果写入到其他的表或者流中，例如：

INSERT INTO result_table
SELECT id, name, event_time
FROM kafka_table
WHERE _partition = 0;

在上面的代码中，我们使用了 `INSERT INTO` 语句将查询结果写入到名为 `result_table` 的表中。