PyFlink1.14 Datastream结合Pandas处理Kafka无界流不生效问题求解

你的实现思路存在两个核心错误，导致作业无法正常运行：

• table.to_pandas()为批处理场景设计，调用时会等待上游全量数据到达后才会返回Pandas DataFrame，在Kafka无界流场景下会永久阻塞，作业执行图根本无法生成提交到集群，因此Flink UI看不到对应作业，也不会有任何输出。
• 你使用SimpleStringSchema读取Kafka数据，读入的Table只有一个字符串类型的字段，存储完整JSON串，就算能走到Pandas处理步骤，也无法直接映射JSON内的字段结构，加的新字段也无法正常序列化输出。

无界流下使用Pandas逻辑处理数据，需要使用PyFlink的向量化Pandas UDF，配合mini-batch触发机制，实现流式批量处理，不会阻塞作业运行。完整可运行代码如下：

import os
import pandas as pd
from pyflink.common.serialization import Encoder
from pyflink.common.serialization import JsonRowDeserializationSchema, JsonRowSerializationSchema
from pyflink.common.typeinfo import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import StreamingFileSink, FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.table import StreamTableEnvironment
from pyflink.table.udf import udf

KAFKA_SERVERS = os.getenv('KAFKA_BS_SERVERS',"kafka:9094").split(',')
KAFKA_USERNAME = "user"
KAFKA_PASSWORD = "123"
KAFKA_SOURCE_TOPIC = 'topic_one'
KAFKA_SINK_TOPIC = 'topic_two'

# 定义输入JSON的字段结构
INPUT_SCHEMA = Types.ROW_NAMED(
    ["Cylinders", "Displacement", "Horsepower", "Weight", "Acceleration", "Model Year", "USA", "Europe", "Japan"],
    [Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.INT(), Types.INT(), Types.INT(), Types.INT()]
)

# 定义输出JSON的字段结构
OUTPUT_SCHEMA = Types.ROW_NAMED(
    ["Cylinders", "Displacement", "Horsepower", "Weight", "Acceleration", "Model Year", "USA", "Europe", "Japan", "Testing"],
    [Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.DOUBLE(), Types.INT(), Types.INT(), Types.INT(), Types.INT(), Types.STRING()]
)

# 定义Pandas处理UDF，输入输出都是Pandas DataFrame，向量化执行
@udf(input_types=INPUT_SCHEMA, result_type=OUTPUT_SCHEMA, func_type="pandas")
def process_pandas(batch_df: pd.DataFrame) -> pd.DataFrame:
    # 自定义Pandas处理逻辑
    batch_df["Testing"] = "new Vals"
    return batch_df

def create_kafka_source(usern: str, password: str, topic: str): 
    kafka_props = {
        'bootstrap.servers': ','.join(KAFKA_SERVERS),
        'group.id': 'testgroup12',
        'auto.offset.reset': 'earliest',
        'sasl.mechanism': 'PLAIN',
        'sasl.jaas.config' : f"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"{usern}\" password=\"{password}\";",
        'security.protocol': 'SASL_PLAINTEXT',
        "enable.auto.commit": "true",
        "auto.commit.interval.ms": "1000",
        "session.timeout.ms": "30000"
    }
    # 使用JSON反序列化Schema直接解析结构化数据
    deserialization_schema = JsonRowDeserializationSchema.builder().type_info(INPUT_SCHEMA).build()
    kafka_source = FlinkKafkaConsumer(
        topics=[topic],
        deserialization_schema=deserialization_schema,
        properties= kafka_props
    )
    kafka_source.set_commit_offsets_on_checkpoints(True)
    return kafka_source

def create_kafka_sink(usern: str, password: str, topic: str):
    # 使用JSON序列化Schema输出结构化数据
    serialization_schema = JsonRowSerializationSchema.builder().with_type_info(OUTPUT_SCHEMA).build()
    kafka_producer = FlinkKafkaProducer(
        topic= topic,
        serialization_schema=serialization_schema,
        producer_config= {
            'bootstrap.servers': ','.join(KAFKA_SERVERS),
            'sasl.mechanism': 'PLAIN',
            'sasl.jaas.config' : f"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"{usern}\" password=\"{password}\";",
            'security.protocol': 'SASL_PLAINTEXT'
        }
    )
    return kafka_producer

def pipeline():
    env = StreamExecutionEnvironment.get_execution_environment()
    env.add_jars("file:///opt/flink/lib_py/kafka-clients-2.4.1.jar")
    env.add_jars("file:///opt/flink/lib_py/flink-connector-kafka_2.12-1.14.0.jar")
    env.set_parallelism(1)
    env.enable_checkpointing(5000)
    t_env = StreamTableEnvironment.create(stream_execution_environment= env)
    # 开启mini-batch配置，每2秒或者满1000条触发一次Pandas处理，可根据业务调整
    t_env.get_config().set("table.exec.mini-batch.enabled", "true")
    t_env.get_config().set("table.exec.mini-batch.allow-latency", "2s")
    t_env.get_config().set("table.exec.mini-batch.size", "1000")
    # 注册自定义Pandas UDF
    t_env.create_temporary_function("process_pandas", process_pandas)

    kafka_source = create_kafka_source(KAFKA_USERNAME, KAFKA_PASSWORD, KAFKA_SOURCE_TOPIC)
    ds = env.add_source(kafka_source)

    # 流转Table后调用Pandas UDF处理
    table = t_env.from_data_stream(ds)
    res_table = t_env.sql_query("SELECT process_pandas(*) FROM %s" % table)
    res_ds = t_env.to_data_stream(res_table, OUTPUT_SCHEMA)

    # 输出到文件
    res_ds.add_sink(StreamingFileSink
                .for_row_format('/opt/flink/outputs_dumps', Encoder.simple_string_encoder())
                .build())
    # 输出到Kafka
    kafka_sink = create_kafka_sink(KAFKA_USERNAME, KAFKA_PASSWORD, KAFKA_SINK_TOPIC)
    res_ds.add_sink(kafka_sink)

    env.execute("proto_1")

if __name__ == '__main__':
    pipeline()

关键修改说明 #

• 使用JsonRowDeserializationSchema直接解析Kafka的JSON数据为结构化Row，不需要手动解析字符串
• 使用向量化Pandas UDF处理数据，支持无界流场景下的批量触发执行，不会阻塞作业
• 开启mini-batch配置，控制Pandas处理的触发频率，平衡延迟和处理效率
• 使用JsonRowSerializationSchema将处理后的结构化数据序列化为JSON字符串写入Kafka，新增的Testing字段会正常输出

内容的提问来源于stack exchange，提问作者Paul