如何在pyflink1.15下将kafka内容写入到mysql？

要将 Kafka 中的数据写入 MySQL 中，您需要使用 Flink 的 Table API 和 SQL API，以及相应的 Flink Connectors 来连接 Kafka 和 MySQL。

下面是一个简单的示例代码，演示了如何使用 Flink 将 Kafka 中的数据写入 MySQL 中：

scheme Copy from pyflink.table import * from pyflink.table.descriptors import * from pyflink.table.types import DataTypes

env_settings = EnvironmentSettings
.new_instance()
.in_streaming_mode()
.use_blink_planner()
.build()

t_env = TableEnvironment.create(env_settings)

kafka_props = {'bootstrap.servers': 'kafka-broker:9092', 'group.id': 'group1'}

t_env
.connect( Kafka() .version('universal') .topic('input-topic') .start_from_earliest() .properties(kafka_props))
.with_format( Json() .json_schema('{"type":"object","properties":{"f0":{"type":"integer"},"f1":{"type":"float"}}}') .fail_on_missing_field(True))
.with_schema( Schema() .field('f0', DataTypes.INT()) .field('f1', DataTypes.FLOAT()))
.create_temporary_table('input_table')

t_env
.connect( JDBC() .url('jdbc:mysql://localhost:3306/mydb') .table_name('output_table') .driver_name('com.mysql.jdbc.Driver') .username('user') .password('password'))
.with_schema( Schema() .field('f0', DataTypes.INT()) .field('f1', DataTypes.FLOAT()))
.create_temporary_table('output_table')

t_env
.from_path('input_table')
.insert_into('output_table')

t_env.execute('kafka_to_mysql') 在上面的示例中，我们首先创建了一个 Kafka source，使用 Json() 格式来解析 Kafka 中的数据。然后我们将解析得到的数据插入到一个 MySQL sink 中，使用 JDBC() 连接器来连接 MySQL 数据库。

需要注意的是，您需要根据实际情况来修改代码中的连接参数和表名等信息，以及将 Json() 格式中的 json_schema 参数修改为适合您数据格式的 JSON Schema。