如何用Dataflow将窗口化压缩字节写入Google Cloud Storage？

因为TextIO仅支持字符串类型写入，要实现窗口划分后写入byte[]到GCS，得换用FileIO搭配自定义Sink来实现，具体步骤如下：

1. 自定义字节流Sink #

实现FileIO.Sink<byte[]>接口，处理byte[]的写入逻辑，还可按需加入压缩（比如GZIP）：

public class ByteSink extends FileIO.Sink<byte[]> {
    private BufferedOutputStream outputStream;

    @Override
    public void open(WritableByteChannel channel) throws IOException {
        // 加入GZIP压缩逻辑，不需要的话直接用Channels.newOutputStream(channel)即可
        OutputStream rawStream = Channels.newOutputStream(channel);
        this.outputStream = new BufferedOutputStream(new GZIPOutputStream(rawStream));
    }

    @Override
    public void write(byte[] element) throws IOException {
        outputStream.write(element);
        // 若需要给每个byte[]添加分隔符（比如换行），可在这里追加：outputStream.write('\n');
    }

    @Override
    public void flush() throws IOException {
        outputStream.flush();
    }

    @Override
    public void close() throws IOException {
        outputStream.close();
    }
}

2. 替换原TextIO逻辑为FileIO #

修改原WriteOneFilePerWindow类，用FileIO实现窗口级别的文件输出，确保每个窗口生成单独文件：

public static PTransform<PCollection<byte[]>, PDone> writeWindowedFiles(String outputDirectory) {
    return FileIO.write()
            .via(new ByteSink())
            .to(outputDirectory)
            .withWindowedWrites() // 关键配置：保证每个窗口输出单独文件
            .withNaming(() -> FileIO.Write.defaultNaming("window-output", ".gz")); // 自定义文件名前缀和后缀
}

如果你的数据流是带窗口信息的WindowedValue类型，也可以用writeDynamic按窗口分区：

public static PTransform<PCollection<WindowedValue<byte[]>>, PDone> writeWindowedFiles(String outputDirectory) {
    return FileIO.<WindowedValue<byte[]>, String>writeDynamic()
            .by(windowedVal -> windowedVal.getWindow().toString()) // 按窗口作为分区键
            .via(Contextful.fn(ByteSink::new), Contextful.fn(WindowedValue::getValue))
            .to(outputDirectory)
            .withNaming((windowKey) -> FileIO.Write.defaultNaming("window-" + windowKey, ".gz"))
            .withWindowedWrites();
}

3. 保持窗口配置一致 #

保留原示例中的窗口划分逻辑（比如固定窗口、触发条件等），确保数据流按预期的窗口分组后进入写入逻辑。

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