如何将每个窗口的Parquet文件名统一写入单个GCS元数据文件？

问题背景 #

业务需求为：将数据写入GCS的Parquet文件后，在每个窗口结束时，把该窗口生成的所有Parquet文件名汇总写入单个元数据文件。实际执行中，单个窗口的Parquet文件名被分散到多个元数据文件，无法满足需求。

期望输出示例：

Metadata Filename: gs://my-bucket/path/to/my/metadata-file/metadata-20240117T12:40-20240117T12:45.txt

Metadata File Content:
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-0.parquet
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-1.parquet
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-2.parquet
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-3.parquet
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-4.parquet
gs://my-bucket/path/to/my/parquet-file/parquet-20240117T12:40-20240117T12:45-5.parquet

错误原因 #

从代码逻辑来看，核心问题在于：

1. 缺少全局聚合步骤：FileSink的并行子任务会各自生成Parquet文件，原代码中收集文件名的逻辑是每个子任务单独输出自身生成的文件名，未将同一窗口下所有子任务的文件名做全局汇总。
2. 元数据输出未做并行度控制：输出元数据时使用默认并行度，多个并行任务同时写入，导致同一窗口的文件名被拆分到多个文件中。

解决方案 #

1. 按窗口聚合所有文件名 #

首先确保每个Parquet文件名包含明确的窗口标识（比如窗口起止时间），然后通过Flink的窗口算子将同一窗口下的所有文件名聚合为一个列表：

// 假设parquetFilePaths是包含Parquet文件路径的数据流，路径中带有窗口起止时间信息
DataStream<String> parquetFilePaths = ...;

// 按窗口ID分组，聚合同一窗口的所有文件名
DataStream<String> aggregatedMetadata = parquetFilePaths
    .keyBy(filePath -> extractWindowIdentifier(filePath)) // 从文件名提取窗口标识（如20240117T12:40-20240117T12:45）
    .window(TumblingEventTimeWindows.of(Time.minutes(5))) // 匹配业务使用的窗口大小
    .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
        @Override
        public void process(String windowId, Context ctx, Iterable<String> paths, Collector<String> out) {
            // 构造完整的元数据内容
            String metadataFileName = String.format("gs://my-bucket/path/to/metadata/metadata-%s.txt", windowId);
            StringBuilder content = new StringBuilder();
            content.append("Metadata Filename: ").append(metadataFileName).append("\n\n");
            content.append("Metadata File Content:\n");
            for (String path : paths) {
                content.append(path).append("\n");
            }
            out.collect(content.toString());
        }
    });

2. 单任务输出元数据文件 #

为避免多个并行任务拆分文件，将元数据输出的并行度设置为1，确保同一窗口的元数据只由一个任务写入：

aggregatedMetadata
    .setParallelism(1)
    .addSink(FileSink.forRowFormat(
            new Path("gs://my-bucket/path/to/metadata"),
            new SimpleStringEncoder<String>("UTF-8")
        )
        .withOutputFileConfig(OutputFileConfig.builder()
                .withPartSuffix(".txt")
                .build())
        .withBucketAssigner(new BucketAssigner<String, String>() {
            @Override
            public String getBucketId(String metadataContent, Context ctx) {
                // 从元数据内容中提取窗口ID，确保同一窗口的元数据写入同一个文件
                return extractWindowIdFromContent(metadataContent);
            }

            @Override
            public SimpleVersionedSerializer<String> getSerializer() {
                return SimpleVersionedStringSerializer.INSTANCE;
            }
        })
        .build());

关键注意事项 #

• 文件名必须包含可提取的窗口标识，这是按窗口聚合的基础。
• 元数据输出的并行度必须设为1，或使用全局窗口，防止多任务写入导致文件拆分。
• 使用BucketAssigner确保同一窗口的元数据内容写入同一个目标文件，避免生成多个碎片文件。

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