Apache Beam序列化报错：DoFnWithExecutionInformation因PipelineOptions不可序列化

错误信息 #

java.lang.IllegalArgumentException: 无法序列化DoFnWithExecutionInformation{doFn=WriteWithAppendToGoFile$CreateTrailerDoFn@57b711b6, mainOutputTag=Tag, sideInputMapping={}, schemaInformation=DoFnSchemaInformation{elementConverters=[], fieldAccessDescriptor=*}}

原因：java.io.NotSerializableException: PipelineOptions对象不可序列化，不应嵌入到转换操作中（您是否在字段或匿名类中捕获了PipelineOptions对象？）。如果您使用DoFn，请在运行时通过ProcessContext/StartBundleContext/FinishBundleContext.getPipelineOptions()访问PipelineOptions，或者在管道构建时提前从PipelineOptions中提取必要字段。

问题代码 #

private class CreateTrailerDoFn<T extends Extract> extends DoFn <T,String> {
    @ProcessElement
    public void processElement(ProcessContext context) {
        final int[] count = {0};
        String timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new java.util.Date());
        T data = context.element();
        data.getContent().forEach(row -> {
            LOG.info(String.format("data : %s", row));
            count[0]++;
        });
        String trailerRow = String.format("%s,%s", count[0], timeStamp);
        LOG.info(trailerRow);
        context.output(trailerRow);
    }
}

问题根源 #

你的CreateTrailerDoFn是非静态内部类，Java中这类类会隐式持有外部类的实例引用。如果外部类中持有了PipelineOptions对象（或其他不可序列化的成员），当Beam尝试序列化DoFn时，会连带序列化外部类实例，而PipelineOptions本身不可序列化，就会触发这个错误。

解决方案 #

1. 将DoFn改为静态内部类 #

把private class改成private static class，切断DoFn对外部类的隐式引用，避免序列化外部类带来的问题：

private static class CreateTrailerDoFn<T extends Extract> extends DoFn<T, String> {
    @ProcessElement
    public void processElement(ProcessContext context) {
        final int[] count = {0};
        String timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new java.util.Date());
        T data = context.element();
        data.getContent().forEach(row -> {
            LOG.info(String.format("data : %s", row));
            count[0]++;
        });
        String trailerRow = String.format("%s,%s", count[0], timeStamp);
        LOG.info(trailerRow);
        context.output(trailerRow);
    }
}

2. 优化线程安全与代码效率（可选但推荐） #

SimpleDateFormat不是线程安全的，在Beam并行执行环境中可能引发异常。建议改用Java 8+的DateTimeFormatter，同时简化计数逻辑：

private static class CreateTrailerDoFn<T extends Extract> extends DoFn<T, String> {
    private DateTimeFormatter formatter;

    @Setup
    public void setup() {
        formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
    }

    @ProcessElement
    public void processElement(ProcessContext context) {
        // 直接用集合size替代循环计数，更高效
        int count = context.element().getContent().size();
        String timeStamp = formatter.format(LocalDateTime.now());
        String trailerRow = String.format("%s,%s", count, timeStamp);
        LOG.info(trailerRow);
        context.output(trailerRow);
    }
}

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