如何修改PyFlink Streaming File Sink的输出路径格式？

要把输出路径从{year}-{month}-{day}--{hour}/part-xxx-{commit}.json改成层级化的{year}/{month}/{day}/{hour}/part-xxx-{commit}.json，核心是调整PyFlink文件Sink的分区路径配置，具体操作如下：

1. 确认使用的Sink类型 #

如果你的PyFlink版本是1.13及以上，推荐使用FileSystemSink；旧版本可能使用StreamingFileSink，两种配置逻辑类似，下面以主流的FileSystemSink为例说明。

2. 配置层级化分区路径 #

假设你的数据流中已经包含year、month、day、hour这几个用于分区的字段，在构建FileSystemSink时，通过with_partition_path方法直接指定层级化的路径模板即可：

from pyflink.common import Configuration
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FileSystemSink, OutputFileConfig
from pyflink.common.serialization import JsonRowSerializationSchema

# 初始化流执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 假设data_stream是你要输出的数据流，已包含year、month、day、hour字段
# 构建JSON序列化器，适配你的数据结构
json_serializer = JsonRowSerializationSchema.builder()
    .with_type_info(data_stream.get_type_info())
    .build()

# 配置输出文件的前缀、后缀（对应part-xxx-0.json格式）
output_file_config = OutputFileConfig.builder()
    .with_part_prefix("part")
    .with_part_suffix(".json")
    .build()

# 构建FileSystemSink并设置分区路径
file_sink = FileSystemSink.builder()
    .set_output_path("file:///your/base/output/dir")  # 替换为你的基础输出目录
    .set_serialization_schema(json_serializer)
    .set_output_file_config(output_file_config)
    .with_partition_path("{year}/{month}/{day}/{hour}")  # 指定层级化的分区目录格式
    .build()

# 将Sink绑定到数据流
data_stream.sink_to(file_sink)

# 启动作业
env.execute("Sync Data to File System")

3. 关键注意点 #

• 必须确保数据流中确实存在year、month、day、hour字段，PyFlink会自动将字段值填充到路径模板的对应位置，生成层级目录。
• 如果这些分区字段是从事件时间/处理时间中提取的（比如从Timestamp字段拆分出年月日时），需要先在数据流中添加这些字段，才能被分区路径模板引用。
• 若使用旧版StreamingFileSink，可以自定义BucketAssigner，在getBucketId方法里返回f"{year}/{month}/{day}/{hour}"格式的字符串，同样能实现层级分区。

配置完成后，输出文件会按照基础目录/年/月/日/时/part-xxx-{commit}.json的结构存储，完全符合需求。

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