Flink批处理模式下写入Parquet时如何限制单文件大小

针对批处理模式下Flink写入Parquet生成超大文件的问题，可通过以下几种方式解决：

1. 配置文件系统连接器的滚动策略参数 #

Flink的filesystem连接器支持通过滚动策略参数控制单文件上限，当文件达到指定大小后自动生成新文件。只需在sink表的WITH配置中添加相关参数：

• sink.rolling-policy.file-size：设置单文件的最大允许大小，支持128MB、1GB这类带单位的配置
• sink.rolling-policy.check-interval：可选，设置检查文件是否达到滚动条件的时间间隔

修改后的sink表DDL示例：

parquet_sink_ddl = f"""
                            CREATE TABLE sink (
                                ...
                            ) WITH (
                                'connector' = 'filesystem',
                                'path' = 's3a://xxx',
                                'format' = 'parquet',
                                'sink.rolling-policy.file-size' = '128MB',
                                'sink.rolling-policy.check-interval' = '10s'
                            )
                        """

2. 手动重分区提升作业并行度 #

批处理模式下如果作业并行度不足，会导致单个任务处理全量数据，最终生成单个大文件。可以通过REPARTITION算子将数据拆分到多个并行子任务中，每个子任务独立生成文件：

修改后的INSERT语句示例：

t_env.execute(f"""
                        INSERT INTO sink
                        SELECT * FROM source REPARTITION(8)
                    """)

这里的8是并行度数值，可根据总数据量和目标单文件大小调整，比如总数据1GB、目标单文件128MB时，设置并行度8-10即可。

3. 配合Parquet格式参数优化存储结构 #

可同时配置Parquet的块大小参数，让文件内部数据块更合理，配合滚动策略实现更均匀的文件拆分：

'format' = 'parquet',
'parquet.block.size' = '128MB'

注意该参数是Parquet的数据块大小，并非文件大小，主要用于优化存储结构，辅助滚动策略实现更均衡的文件拆分。

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