Python实战：Python在实时数据流处理中的Flink与Kafka集成

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common import WatermarkStrategy, Row
from pyflink.common.typeinfo import Types
from pyflink.connector.kafka import KafkaSource, KafkaSink
 
def print_result(row):
    print(row)
 
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
 
# 设置水印策略
env.set_stream_time_characteristic(WatermarkStrategy.for_monotonous_timestamps())
 
# 创建Kafka源
kafka_source = KafkaSource() \
    .for_bootstrap_servers(["localhost:9092"]) \
    .for_topic("your_input_topic") \
    .build()
 
# 创建数据流
stream = env.add_source(kafka_source, type_info=Types.ROW([Types.INT(), Types.STRING()]))
 
# 注册结果打印函数
stream.map(lambda r: Row(r[0], r[1]), output_type=Types.ROW([Types.INT(), Types.STRING()])).add_sink(print_result)
 
# 执行作业
env.execute("Kafka Flink Integration Job")

这段代码演示了如何使用PyFlink库从Kafka主题读取数据，并在Flink数据流中处理这些数据，最后打印结果。代码中包含了设置并行度、设置水印策略、创建Kafka源、构建数据流、添加打印接收器以及执行作业的步骤。