pyflink 1.13.3版本，接受kafka消息，流处理，然后写入数据库

每次接受到kafka消息后，处理结果写入数据库，但是我就想要最后的结果，比如计数，最后结果为20，我只想把最后一条数据写入数据库，而不是要把流处理的整个过程写入数据库。

展开

收起

游客4iy47oy4ywm2s 2023-08-08 14:30:48 161 1

7 条回答

写回答

取消提交回答

三掌柜666

十分耕耘，一定会有一分收获！

楼主你好，要实现最终结果只写入最后一条数据到数据库，可以使用PyFlink的窗口操作。例如使用滚动窗口，设置窗口大小为1，每次接收到一条kafka消息就会触发一次窗口操作，将这条消息加入窗口内进行计算，并在窗口关闭时将最终结果写入数据库。

代码示例：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import EnvironmentSettings, TableConfig
from pyflink.table.functions import *
from pyflink.table.descriptors import Kafka, Json, Rowtime, FileSystem
from pyflink.table.window import Tumble

env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
t_config = TableConfig()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings, table_config=t_config)

# 设置Kafka连接信息
kafka_props = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'test-group'
}
kafka_source = Kafka()
kafka_source.version('universal').topic('test-topic') \
    .properties(kafka_props).start_from_latest() \
    .value_format(Json().fail_on_missing_field(True))

# 注册Kafka数据源
t_env.connect(kafka_source) \
    .with_format(Json().fail_on_missing_field(True)) \
    .with_schema(
    Schema()
        .field('id', DataTypes.INTEGER())
        .field('name', DataTypes.STRING())
        .field('time', DataTypes.TIMESTAMP(3)).rowtime(Rowtime()
                                                         .timestamps_from_field('time')
                                                         .watermarks_periodic_bounded(60000))) \
    .in_append_mode().register_table_source('kafka_source')

# 注册MySQL数据输出
mysql_props = {
    "url": "jdbc:mysql://localhost:3306/test",
    "driver": "com.mysql.jdbc.Driver",
    "username": "root",
    "password": "123456",
    "table-name": "result_sink"
}
mysql_sink = FileSystem().path('/tmp/pyflink/output') \
    .with_format(Json().derive_schema()) \
    .with_schema(
    Schema()
        .field('id', DataTypes.BIGINT())
        .field('name', DataTypes.STRING())
        .field('count', DataTypes.BIGINT())
) \
    .create_temporary_table('mysql_sink')

# 查询数据并写入MySQL
t_env.from_path('kafka_source') \
    .window(Tumble.over('1.rows').on('time').alias('w')) \
    .group_by('name, w') \
    .select('name, COUNT(id) as count') \
    .insert_into('mysql_sink')

# 执行作业
t_env.execute("kafka_to_mysql")

这个示例程序会将kafka的数据流以1行的滚动窗口进行处理，每当窗口内收到1条消息时，就会触发一次计算，并将当前计算结果作为一条数据插入到MySQL数据库中。当所有窗口都关闭时，只会将最后一条结果写入到数据库中。

2023-08-29 14:57:22

赞同展开评论打赏

vohelon

要在 PyFlink 1.13.3 中接收 Kafka 消息并进行流处理，然后将最终结果写入数据库，可以按照以下步骤进行操作：

安装 PyFlink 和相关依赖：
```
pip install apache-flink==1.13.3
```

编写 PyFlink 程序代码：创建一个新的 Python 脚本，导入必要的模块，并编写流处理的逻辑。

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

# 创建流处理的执行环境
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)  # 设置并行度为 1
t_env = StreamTableEnvironment.create(env, environment_settings=EnvironmentSettings.new_instance().in_streaming_mode())

# 定义 Kafka 连接属性
kafka_properties = {...}  # 替换为实际的 Kafka 连接属性

# 从 Kafka 消息队列读取数据
source_ddl = """
CREATE TABLE source_table (
    ...
) WITH (
    'connector' = 'kafka',
    'topic' = 'your_topic',
    'properties.bootstrap.servers' = 'your_bootstrap_servers',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
)
"""
t_env.execute_sql(source_ddl)

# 执行流处理操作
t_env.execute_sql("""
    INSERT INTO result_table
    SELECT ...
    FROM source_table
    GROUP BY ...
""")

# 将最终结果写入数据库
final_result = t_env.from_path("result_table")
final_result.execute_insert("your_database_sink")

# 执行流处理任务
env.execute("Kafka to Database")

编写 SQL 查询：根据您的需求，在上述代码中的 SELECT 和 GROUP BY 子句中定义要计算的指标和条件。
配置数据库连接器（Sink）：在 PyFlink 中，您需要配置适当的数据库连接器来将最终结果写入数据库。具体设置取决于您所使用的数据库类型和驱动程序。请参考 PyFlink 文档中关于数据库连接器设置的部分，以找到适合您的数据库和驱动程序的示例。
运行 PyFlink 程序：在命令行中执行以下命令运行您编写的 PyFlink 程序。
```
python your_pyflink_program.py
```

这样，PyFlink 将会从 Kafka 消息队列读取数据，并进行流处理操作。最终结果会被写入指定的数据库中。如果您只想将最后的结果写入数据库，可以根据您的计算逻辑，在流处理的最后使用 sink 操作将结果写入数据库，而不是在每次接收到消息时都写入数据库。

2023-08-09 08:55:42

赞同展开评论打赏

算精通

北京阿里云ACE会长
一种可能的解决方案是使用Flink的聚合函数（Aggregation Function）来计算最终的结果，并将该结果写入数据库。以下是一个示例代码：

python
Copy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
from pyflink.table.udf import AggregateFunction

class CountAggregate(AggregateFunction):
def create_accumulator(self):
return 0
```
def add(self, value, accumulator):
    return accumulator + 1

def get_result(self, accumulator):
    return accumulator

def get_accumulator_type(self):
    return DataTypes.BIGINT()

def get_result_type(self):
    return DataTypes.BIGINT()
```
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

t_env = StreamTableEnvironment.create(
env,
environment_settings=EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
)

kafka_props = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'my_group',
'auto.offset.reset': 'latest'
}

source_ddl = f"""
CREATE TABLE kafka_source (
data VARCHAR
) WITH (
'connector' = 'kafka',
'topic' = 'my_topic',
'properties.bootstrap.servers' = '{kafka_props["bootstrap.servers"]}',
'properties.group.id' = '{kafka_props["group.id"]}',
'scan.startup.mode' = '{kafka_props["auto.offset.reset"]}',
'format' = 'json',
'json.fail-on-missing-field' = 'false'
)
"""

sink_ddl = """
CREATE TABLE result_sink (
result BIGINT
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/my_database',
'table-name' = 'result_table',
'username' = 'my_username',
'password' = 'my_password',
'driver' = 'com.mysql.jdbc.Driver'
)
"""

t_env.execute_sql(source_ddl)
t_env.execute_sql(sink_ddl)

t_env.register_function("count_agg", CountAggregate())

t_env.from_path("kafka_source") \
.group_by() \
.select("count_agg(data) as result") \
.execute_insert("result_sink")

env.execute("Write to database")
在上述示例中，我们首先定义了一个名为CountAggregate的自定义聚合函数，用于计算消息的数量。然后，我们创建了一个名为kafka_source的Kafka数据源表和一个名为result_sink的目标表。接着，我们使用count_agg函数对kafka_source表进行聚合，并将结果写入result_sink表。

请根据您的实际情况修改代码中的Kafka和数据库连接属性，以及表名和字段名等相关信息。
2023-08-08 20:53:20

赞同展开评论打赏
Star时光
如果你只想将流处理的最终结果写入数据库，而不将整个过程中的所有数据都写入数据库，可以考虑使用状态或缓冲区来累积结果，并在流处理结束时将最终结果写入数据库。

以下是一种可能的实现方式：
1. 在流处理过程中使用状态：在PyFlink中，你可以使用ValueState或ListState等状态变量来保存并更新计数结果。
2. 设置触发器：使用PyFlink提供的触发器机制，当满足特定条件时（例如，当接收到一定数量的消息时），触发操作将结果写入数据库。你可以自定义触发器逻辑，以满足你的需求。
3. 缓冲区和异步写入：使用一个缓冲区（例如列表）来存储要写入数据库的数据。每次处理到一条消息时，将其加入缓冲区。同时，使用异步写入的方式进行数据库写入，以提高性能和效率。
4. 在流处理结束时写入数据库：当流处理任务结束时，可以通过添加一个关闭钩子函数来确保最后一个结果被写入数据库。这样，当任务停止时，你可以在关闭钩子函数中将最终的计数结果从缓冲区中取出并写入数据库。
请注意，上述解决方案仅为一种实现思路，并且基于PyFlink的常用功能。根据具体情况，你可能需要根据PyFlink的API和文档进行适当调整和扩展。
2023-08-08 18:39:45

赞同展开评论打赏

小周sir

面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。

对于你的需求，在PyFlink 1.13.3版本中，你可以使用一个状态变量来保存最后的结果，并在流处理任务结束时将其写入数据库。

首先，你需要定义一个状态变量来保存最后的结果。这可以是一个简单的计数器或其他你想要的类型。例如，我们假设你要计算收到的消息数量：

from pyflink.common import WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.table import StreamTableEnvironment
from pyflink.table.udf import AggregateFunction

class MessageCounter(AggregateFunction):
    def create_accumulator(self):
        return 0

    def add(self, accumulator, value):
        return accumulator + 1

    def get_result(self, accumulator):
        return accumulator

    def merge(self, a, b):
        return a + b

# 创建StreamExecutionEnvironment和StreamTableEnvironment
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# 定义Kafka消费者
kafka_props = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my_consumer_group'
}
consumer = FlinkKafkaConsumer('my_topic', 'my_deserialization_schema', properties=kafka_props)

# 从Kafka读取数据并注册为表
input_table = t_env.from_data_stream(env.add_source(consumer))
t_env.create_temporary_view("my_table", input_table)

# 执行流处理逻辑
result_table = t_env.sql_query("SELECT COUNT(*) AS count FROM my_table")
t_env.register_function("message_counter", MessageCounter())
final_result_table = result_table.group_by().select("message_counter(count) AS final_count")

# 将最后的结果写入数据库
final_result_table.insert_into("my_database.my_table")

# 执行任务
env.execute()

在上述示例中，我们使用MessageCounter自定义了一个聚合函数，用于计算接收到的消息数量。然后，我们根据该聚合函数进行流处理，并将最后的计数结果写入数据库。

请注意，这只是一个基本示例，你需要根据你的实际需求和数据库类型来调整代码。另外，确保你已正确配置Kafka连接、序列化/反序列化和数据库连接等相关参数。

2023-08-08 17:53:03

赞同展开评论打赏

魏红斌

天下风云出我辈，一入江湖岁月催，皇图霸业谈笑中，不胜人生一场醉。

如果您只想把最后一条数据写入数据库，可以使用Kafka消费者的偏移量控制来实现。具体来说，您可以设置消费者的偏移量为最后一条消息的偏移量，然后在处理完最后一条消息后，将处理结果写入数据库。

2023-08-08 17:44:53

赞同展开评论打赏
凌云Cloud

发表文章、提出问题、分享经验、结交志同道合的朋友
kafka提供聚合操作，你的业务场景其实就是最终聚合，需要结合窗口操作完成。
具体实现逻辑如下：

1、需要使用窗口操作来对流进行分组并聚合。可以选择使用滚动窗口或会话窗口，窗口可以对一定时间范围内的数据进行聚合。

2、聚合操作，在窗口内，进行你所需要的聚合操作，如计数、求和等。这将会得到窗口内的聚合结果。

3、最终结果提取，一旦窗口结束，你可以在窗口结束时间点，将窗口内的聚合结果提取出来。这就是你关心的最终结果。

4、写入数据库，最终结果可以被写入数据库，这将是你的最终聚合结果。

示例代码如下：
```
stream = env.add_source(...)  # 从 Kafka 接收数据的流

# 使用窗口操作，例如滚动窗口，对一定时间范围内的数据进行聚合
windowed_stream = stream.key_by(...) \
                      .window(...) \
                      .reduce(...)  # 这里使用 reduce 操作进行简单的聚合

# 最终结果提取，一旦窗口结束，提取窗口内的最终聚合结果
final_result_stream = windowed_stream.apply(lambda key, window, data: data[-1])

# 将最终结果写入数据库
final_result_stream.add_sink(...)  # 将数据写入数据库的 Sink
```
2023-08-08 14:56:43

赞同展开评论打赏