Windows系统下Apache Beam流式窗口输出文件名含冒号报错求助

Windows系统不允许文件名包含冒号，而Apache Beam默认生成窗口输出文件时，会直接使用带冒号的ISO格式时间戳（比如2020-12-12T15:00:10.000Z）作为文件名的一部分，导致抛出无效文件名错误。要解决这个问题，你需要自定义文件名生成逻辑，把时间戳里的冒号替换成点。

实现步骤 #

1. 自定义FilenamePolicy类，重写窗口文件名的生成逻辑，将时间戳中的冒号替换为点
2. 在TextIO.write()中使用这个自定义的策略，替代默认的文件名生成规则

完整代码示例 #

自定义FilenamePolicy类

import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.ResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.values.PaneInfo;

import java.time.Instant;
import java.time.format.DateTimeFormatter;

public class CustomWindowFilenamePolicy extends TextIO.FilenamePolicy {
    private final ResourceId baseOutputDir;
    private final String prefix;

    public CustomWindowFilenamePolicy(ResourceId baseOutputDir, String prefix) {
        this.baseOutputDir = baseOutputDir;
        this.prefix = prefix;
    }

    @Override
    public ResourceId windowedFilename(
            int shardNumber,
            int numShards,
            BoundedWindow window,
            PaneInfo paneInfo,
            FileBasedSink.OutputFileHints outputFileHints) {
        // 固定窗口属于IntervalWindow类型，强转后获取起止时间
        IntervalWindow intervalWindow = (IntervalWindow) window;
        // 自定义时间格式，把冒号替换成点
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH.mm.ss.SSS'Z'");
        String startTimestamp = formatter.format(intervalWindow.start().toInstant());
        String endTimestamp = formatter.format(intervalWindow.end().toInstant());

        // 拼接符合要求的文件名
        String filename = String.format("%s%s-%s-pane-%d-%s-%05d-of-%05d%s",
                prefix,
                startTimestamp,
                endTimestamp,
                paneInfo.getIndex(),
                paneInfo.getTiming().toString().toLowerCase(),
                shardNumber,
                numShards,
                outputFileHints.getSuggestedFilenameSuffix());

        return baseOutputDir.resolve(filename, ResolveOptions.StandardResolveOptions.RESOLVE_FILE);
    }

    @Override
    public ResourceId unwindowedFilename(
            int shardNumber, int numShards, FileBasedSink.OutputFileHints outputFileHints) {
        // 非窗口场景的文件名生成，按需调整
        return baseOutputDir.resolve(
                String.format("%s-%05d-of-%05d%s", prefix, shardNumber, numShards, outputFileHints.getSuggestedFilenameSuffix()),
                ResolveOptions.StandardResolveOptions.RESOLVE_FILE);
    }
}

修改后的主代码

output.apply(ParDo.of(new DoFn<KV<String, Long>, String>() {
    @ProcessElement
    public void processElement(ProcessContext c, BoundedWindow window) {
        c.output(String.format("%s %s %s",
                window.maxTimestamp(),
                Objects.requireNonNull(c.element()).getKey(),
                Objects.requireNonNull(c.element()).getValue()));
    }
}))
.apply(TextIO.write()
        .to("src/main/resources/Sink/FixedWindow") // 这里传入目标目录，而非带前缀的路径
        .withFilenamePolicy(new CustomWindowFilenamePolicy(
                FileBasedSink.convertToFileResourceIfPossible("src/main/resources/Sink/FixedWindow"),
                "output")) // 指定文件名前缀为"output"
        .withWindowedWrites()
        .withNumShards(1));

关键说明 #

• 自定义策略中，通过DateTimeFormatter将时间戳的HH:mm:ss格式改为HH.mm.ss，彻底避免冒号出现在文件名中
• TextIO.write().to()方法现在只传入目录路径，文件名的前缀和格式由CustomWindowFilenamePolicy控制
• 生成的文件名会和你期望的格式完全一致：output2020-12-12T15.00.10.000Z-2020-12-12T15.00.15.000Z-pane-0-last-00000-of-00001

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