使用Apache Beam左外连接PCollection时出现Schema相关异常求助

这个错误的核心原因是Apache Beam的Join.leftOuterJoin操作依赖输入PCollection的Schema元数据，而你的ClassA和ClassB对应的PCollection没有被Beam识别为带Schema的类型，导致Join操作无法自动生成输出Row的结构，从而触发Cannot call getSchema when there is no schema异常。

具体原因及修复方案 #

1. 自定义类未添加Schema注解

Beam需要通过注解来识别Java类的字段作为Schema。如果ClassA和ClassB是自定义类，需添加Schema相关注解，让Beam能解析类的结构：

import org.apache.beam.sdk.schemas.JavaBeanSchema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;

// 为ClassA添加Schema注解，类需包含标准getter/setter
@DefaultSchema(JavaBeanSchema.class)
public class ClassA {
    private String id1;
    // 其他字段
    public String getId1() { return id1; }
    public void setId1(String id1) { this.id1 = id1; }
    // 其他字段的getter/setter
}

// 对ClassB做同样处理
@DefaultSchema(JavaBeanSchema.class)
public class ClassB {
    private String id2;
    public String getId2() { return id2; }
    public void setId2(String id2) { this.id2 = id2; }
    // 其他字段的getter/setter
}

2. BigQuery读取时未保留Schema映射

如果BigQueryReader.getClassA是自定义读取逻辑，需确保读取过程中正确将BigQuery的Schema映射到Java类。推荐直接使用Beam内置的BigQueryIO读取方法，它会自动处理Schema映射：

// 替换自定义的BigQueryReader，直接用BigQueryIO读取并绑定到带注解的类
final PCollection<ClassA> classA = pipeline.apply(
    BigQueryIO.read(ClassA.class)
              .from("<你的BigQuery表路径>")
);

final PCollection<ClassB> classB = pipeline.apply(
    BigQueryIO.read(ClassB.class)
              .from("<你的BigQuery表路径>")
);

3. 显式转换为带Schema的Row类型

如果不想修改类注解，也可以手动将PCollection<ClassA>和PCollection<ClassB>转换为带Schema的PCollection<Row>后再执行Join：

import org.apache.beam.sdk.transforms.SchemaTransforms;

final PCollection<Row> classARows = classA.apply(
    SchemaTransforms.toRow(ClassA.class)
);
final PCollection<Row> classBRows = classB.apply(
    SchemaTransforms.toRow(ClassB.class)
);

final PCollection<Row> join = classARows.apply(
    "Apply Join",
    Join.<Row, Row>leftOuterJoin(classBRows).using("id1", "id2")
);

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