Java中使用Beam TextIO.read读取文件时无法识别分隔符问题

为什么分隔符配置不生效 #

• 入参类型不匹配：TextIO.read().withDelimiter()方法仅接收单个分隔符对应的字节数组，你尝试传入{"\n","\n"}属于字符串数组，类型不匹配会直接编译报错。
• 版本兼容缺陷：2.25版本之前的Apache Beam对多字节自定义分隔符的匹配逻辑存在bug，即使文件小于默认读取块大小，也可能出现分隔符识别失败的情况。
• 换行符格式不匹配：Windows环境下编辑的文件默认换行符为\r\n，文件中的连续空行对应的字节序列是\r\n\r\n，你配置的\n\n无法匹配Windows格式的换行空行，最终导致整个文件被识别为单个元素。

可行解决方案 #

方案1：修复分隔符配置 #

首先将GCS中存储的文件统一转换为Unix格式换行符\n，同时将Beam依赖版本升级到2.30及以上，再调整代码配置：

PCollection<String> rows= pipeline.apply("Read From GCS", 
    TextIO.read()
        .from(options.getBucketName())
        .withDelimiter(new byte[] {'\n', '\n'})
        .withAutoSplit(true));

如果确认文件为Windows换行符，将分隔符配置修改为：

.withDelimiter(new byte[] {'\r', '\n', '\r', '\n'})

方案2：后处理分组（兼容性最高） #

如果不想纠结换行符格式和版本兼容问题，可以先按默认行规则读取全量内容，再自行按空行拆分块，代码实现如下：

// 先按行读取所有内容，TextIO默认会自动适配不同系统的换行符
PCollection<String> lines = pipeline.apply("Read All Lines", TextIO.read().from(options.getBucketName()));
PCollection<String> blocks = lines.apply("Split Block By Empty Line", ParDo.of(new DoFn<String, String>() {
    @StateId("currentBlock")
    private final StateSpec<StringBuilder> currentBlock = StateSpecs.stringBuilder();

    @ProcessElement
    public void processElement(@Element String line, 
                               @StateId("currentBlock") State<StringBuilder> blockState,
                               ProcessContext ctx) {
        if (line.trim().isEmpty()) {
            // 遇到空行输出已积累的块
            StringBuilder block = blockState.read();
            if (block != null && block.length() > 0) {
                ctx.output(block.toString());
                blockState.clear();
            }
            return;
        }
        // 非空行追加到当前块
        StringBuilder current = blockState.read() == null ? new StringBuilder() : blockState.read();
        if (current.length() > 0) {
            current.append("\n");
        }
        current.append(line);
        blockState.write(current);
    }

    @FinishBundle
    public void finishBundle(@StateId("currentBlock") State<StringBuilder> blockState,
                             ProcessContext ctx) {
        // 输出最后一个没有空行收尾的块
        StringBuilder block = blockState.read();
        if (block != null && block.length() > 0) {
            ctx.output(block.toString());
            blockState.clear();
        }
    }
}));

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