Apache Beam序列化DoFn失败：TableSchema无法序列化问题求助

问题背景 #

使用Apache Beam 2.41 + Java 11，尝试将String类型的PCollection转换为BigQuery TableRow类型的PCollection。从Avro文件加载TableSchema并通过ValueProvider传入DoFn，运行时出现DoFn无法序列化的错误。

错误原因 #

报错核心是java.io.NotSerializableException: com.google.api.services.bigquery.model.TableSchema：

• com.google.api.services.bigquery.model.TableSchema类本身未实现Serializable接口，即使将其包装在StaticValueProvider中，DoFn序列化时仍会尝试序列化该对象，导致失败。
• 原代码中DoFn直接持有ValueProvider<TableSchema>，而TableSchema实例无法被序列化，进而导致整个DoFn序列化失败。
• 附加问题：原buildTableRow方法存在逻辑错误，如重复复用同一个TableCell实例、switch case无break、错误使用TableRow的setF方法（该方法用于批量设置字段列表，而非单个键值对）。

修复方案 #

1. 避免直接持有不可序列化的TableSchema：改为传入Avro schema文件路径的ValueProvider，在DoFn的@Setup初始化方法或apply方法中动态加载并转换为TableSchema，跳过序列化TableSchema的步骤。
2. 修复TableRow构建逻辑：循环内新建TableCell（或直接使用TableRow的键值对设置），为switch case添加break，修正数据类型转换逻辑。

完整修复代码 #

Main方法 #

public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.create();
    options.setRunner(DirectRunner.class);
    options.setTempLocation("data/temp/");
    Pipeline p = Pipeline.create(options);

    // 传入Avro schema文件路径的ValueProvider，而非已转换的TableSchema
    ValueProvider<String> schemaPath = ValueProvider.StaticValueProvider.of("data/ship_data_schema.avsc");

    PCollection<String> pc1 = p.apply(TextIO.read().from("data/ship_data.csv"));
    PCollection<TableRow> pc2 = pc1.apply(MapElements.via(new ConvertStringToTableRow(schemaPath)));
    
    PipelineResult result = p.run();
    result.waitUntilFinish();
}

修复后的ConvertStringToTableRow类 #

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.sql.Date;
import java.sql.Timestamp;
import java.util.List;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;

public static class ConvertStringToTableRow extends SimpleFunction<String, TableRow> {
    private final ValueProvider<String> schemaPath;
    // 用transient避免序列化，在@Setup中初始化提升性能
    private transient TableSchema tableSchema;

    public ConvertStringToTableRow(ValueProvider<String> schemaPath) {
        this.schemaPath = schemaPath;
    }

    @DoFn.Setup
    public void setup() {
        // 在初始化阶段加载一次schema，避免每次apply都重复加载
        if (schemaPath.isAccessible()) {
            BeamShemaUtil beamShemaUtil = new BeamShemaUtil(schemaPath.get());
            this.tableSchema = beamShemaUtil.convertBQTableSchema();
        }
    }

    private TableRow buildTableRow(TableSchema sc, String[] arr) {
        List<TableFieldSchema> fieldSchemaList = sc.getFields();
        TableRow row = new TableRow();

        for (int i = 0; i < fieldSchemaList.size(); i++) {
            TableFieldSchema fieldSchema = fieldSchemaList.get(i);
            String fieldName = fieldSchema.getName();
            String fieldType = fieldSchema.getType();
            String fieldValue = arr[i].trim();

            if (fieldValue.isEmpty()) {
                row.set(fieldName, null);
                continue;
            }

            switch (fieldType.toUpperCase()) {
                case "STRING":
                    row.set(fieldName, fieldValue);
                    break;
                case "BYTES":
                    row.set(fieldName, fieldValue.getBytes(StandardCharsets.UTF_8));
                    break;
                case "INT64":
                case "INTEGER":
                    row.set(fieldName, Long.valueOf(fieldValue)); // INT64对应Long，避免溢出
                    break;
                case "FLOAT64":
                case "FLOAT":
                    row.set(fieldName, Double.valueOf(fieldValue)); // FLOAT64对应Double
                    break;
                case "BOOL":
                case "BOOLEAN":
                    row.set(fieldName, Boolean.valueOf(fieldValue));
                    break;
                case "NUMERIC":
                    row.set(fieldName, new BigDecimal(fieldValue)); // NUMERIC需要用BigDecimal
                    break;
                case "TIMESTAMP":
                    // 根据实际时间格式转换，示例假设为ISO格式
                    row.set(fieldName, Timestamp.valueOf(fieldValue));
                    break;
                case "DATE":
                    row.set(fieldName, Date.valueOf(fieldValue));
                    break;
                // 其他类型可根据需求扩展
                default:
                    row.set(fieldName, fieldValue);
                    // 可选：抛出异常提示未支持类型
                    // throw new UnsupportedOperationException("Unsupported BigQuery type: " + fieldType);
            }
        }
        return row;
    }

    @Override
    public TableRow apply(String element) {
        String[] arr = element.split(",");
        // 如果@Setup未初始化（比如分布式环境中），则在apply中加载
        if (tableSchema == null && schemaPath.isAccessible()) {
            BeamShemaUtil beamShemaUtil = new BeamShemaUtil(schemaPath.get());
            tableSchema = beamShemaUtil.convertBQTableSchema();
        }
        return buildTableRow(tableSchema, arr);
    }
}

翻译后的错误信息 #

线程"main"中的异常：java.lang.IllegalArgumentException: 无法序列化DoFnWithExecutionInformation{doFn=org.apache.beam.sdk.transforms.MapElements$1@270a620, mainOutputTag=Tag<output>, sideInputMapping={}, schemaInformation=DoFnSchemaInformation{elementConverters=[], fieldAccessDescriptor=*}}
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:59)
    ...（省略中间栈跟踪）
Caused by: java.io.NotSerializableException: com.google.api.services.bigquery.model.TableSchema
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1185)
    ...（省略中间栈跟踪）
进程退出码1

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