[AIGC] flink sql 消费kafka消息，然后写到mysql中的demo_flink sql消费kafka

这是一个使用 Flink SQL 从 Kafka 中消费数据并写入 MySQL 的示例。在这个示例中，我们将假设有一个 Kafka 主题 “input_topic”，它产生格式为 (user_id: int, item_id: int, behavior: string, timestamp: long) 的数据，我们需要把这些数据写入名为"output_table"的 MySQL 表中。

首先，我们需要创建表达 Kafka 和 MySQL 的源和目标表条目：

CREATE TABLE kafka_source (
    user_id INT,
    item_id INT,
    behavior STRING,
    timestamp AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` / 1000, 'yyyy-MM-dd HH:mm:ss')),
    WATERMARK FOR timestamp as timestamp - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'input_topic',
    'scan.startup.mode' = 'earliest-offset',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

CREATE TABLE mysql_sink (
    user_id INT,
    item_id INT,
    behavior STRING,
    timestamp TIMESTAMP(3)
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/test',
    'table-name' = 'output_table',
    'username' = 'test',
    'password' = '123456'
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

进行消费 Kafka 和写入 MySQL 的操作：

INSERT INTO mysql_sink SELECT * FROM kafka_source;
1

首先，运行上述 SQL 创建对应 Kafka 和 MySQL 的表。其次，运行 INSERT INTO 语句进行从 Kafka 消费数据并将结果写入 MySQL 表的操作。这是一个基本的操作，你可以根据自己的需求进行相应的修改。

注意，需要根据实际的 Kafka 和 MySQL 的配置，如地址、用户名和密码，来修改上述 SQL 语句中的对应部分。

另外，Flink SQL 对 SQL 语句的语法进行了些许改变以适应流处理的特性，如在 kafka_source 表中的 WATERMARK 和 timestamp 的定义。

这个例子中，你需要确保已经引入了 flink-connector-kafka_2.11、flink-connector-jdbc_2.11 和 mysql-connector-java 等相关的 jar 包依赖。