pyflink kafka to mysql

在 PyFlink 中使用 Kafka 和 MySQL 需要安装相应的 Python 库。可以使用 `pip` 命令安装 `kafka-python` 和 `mysql-connector-python` 库：

pip install kafka-python
pip install mysql-connector-python

然后，可以按照以下步骤将 Kafka 中的数据写入 MySQL 数据库：

1. 导入必要的库：

from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkJdbcSink
import mysql.connector

2. 创建 Flink 的 StreamExecutionEnvironment：

env = StreamExecutionEnvironment.get_execution_environment()

3. 创建 Kafka 消费者：

properties = {'bootstrap.servers': 'localhost:9092', 'group.id': 'test-group'}
consumer = FlinkKafkaConsumer('test-topic', SimpleStringSchema(), properties=properties)

4. 从 Kafka 中读取数据：

stream = env.add_source(consumer)

5. 将数据写入 MySQL 数据库：

jdbc_sink = FlinkJdbcSink(
    "INSERT INTO my_table (id, name) VALUES (?, ?)",
    connection_options={
        "url": "jdbc:mysql://localhost:3306/my_database",
        "driver": "com.mysql.jdbc.Driver",
        "username": "my_username",
        "password": "my_password"
    },
    statement_execute_parallelism=2,
    flush_on_checkpoint=True)
stream.add_sink(jdbc_sink)

其中，`INSERT INTO my_table (id, name) VALUES (?, ?)` 是 MySQL 的 SQL 语句，`id` 和 `name` 分别是表中的两个字段。`connection_options` 是连接 MySQL 数据库的参数，需要设置 MySQL 的 URL、驱动程序、用户名和密码。

完整的代码示例：

from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkJdbcSink
import mysql.connector

env = StreamExecutionEnvironment.get_execution_environment()

properties = {'bootstrap.servers': 'localhost:9092', 'group.id': 'test-group'}
consumer = FlinkKafkaConsumer('test-topic', SimpleStringSchema(), properties=properties)

stream = env.add_source(consumer)

jdbc_sink = FlinkJdbcSink(
    "INSERT INTO my_table (id, name) VALUES (?, ?)",
    connection_options={
        "url": "jdbc:mysql://localhost:3306/my_database",
        "driver": "com.mysql.jdbc.Driver",
        "username": "my_username",
        "password": "my_password"
    },
    statement_execute_parallelism=2,
    flush_on_checkpoint=True)
stream.add_sink(jdbc_sink)

env.execute("Kafka to MySQL")