如何用Spark Java API从配置文件读取表Schema以避免硬编码？

要避免硬编码Schema，你可以用JSON/YAML这类结构化配置文件存储各表的Schema定义，然后通过Java代码解析配置并转换为Spark的StructType。以下是具体实现方案：

1. 定义配置文件 #

先创建一个包含所有表Schema的配置文件，这里以JSON为例（YAML方案见后文）：

schema_config.json

{
  "emp_dept": [
    {"name": "emp_dept", "type": "string", "nullable": true},
    {"name": "empid", "type": "integer", "nullable": true},
    {"name": "empdesignation", "type": "string", "nullable": true},
    {"name": "emp_salary", "type": "integer", "nullable": true}
  ],
  "emp_details": [
    {"name": "emp_details", "type": "string", "nullable": true},
    {"name": "empid", "type": "integer", "nullable": true},
    {"name": "empfistname", "type": "string", "nullable": true},
    {"name": "emplastname", "type": "integer", "nullable": true}
  ]
}

每个表的Schema包含三个字段：

• name：列名
• type：数据类型（对应Spark的类型，如string/integer）
• nullable：是否允许为空

2. 创建字段配置映射类 #

用POJO类来映射配置文件中的字段定义：

public class FieldConfig {
    private String name;
    private String type;
    private boolean nullable;

    // Getters and Setters
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public String getType() { return type; }
    public void setType(String type) { this.type = type; }
    public boolean isNullable() { return nullable; }
    public void setNullable(boolean nullable) { this.nullable = nullable; }
}

3. 编写Schema加载工具类 #

实现读取配置文件、转换为StructType的工具类：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;

public class SchemaLoader {
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static StructType loadSchema(String configPath, String tableName) throws IOException {
        // 读取JSON配置文件，解析为Map结构
        Map<String, List<FieldConfig>> schemaMap = objectMapper.readValue(
                new File(configPath),
                objectMapper.getTypeFactory().constructMapType(Map.class, String.class, List.class)
        );

        // 获取指定表的字段配置
        List<FieldConfig> fieldConfigs = schemaMap.get(tableName);
        if (fieldConfigs == null) {
            throw new IllegalArgumentException("未找到表 " + tableName + " 的Schema配置");
        }

        // 转换为Spark的StructField数组
        StructField[] fields = fieldConfigs.stream()
                .map(config -> DataTypes.createStructField(
                        config.getName(),
                        convertStringToDataType(config.getType()),
                        config.isNullable()
                ))
                .toArray(StructField[]::new);

        return DataTypes.createStructType(fields);
    }

    // 将配置中的字符串类型转换为Spark DataType
    private static org.apache.spark.sql.types.DataType convertStringToDataType(String typeStr) {
        return switch (typeStr.toLowerCase()) {
            case "string" -> DataTypes.StringType;
            case "integer" -> DataTypes.IntegerType;
            case "long" -> DataTypes.LongType;
            case "double" -> DataTypes.DoubleType;
            case "boolean" -> DataTypes.BooleanType;
            // 可根据需求扩展更多类型
            default -> throw new IllegalArgumentException("不支持的数据类型: " + typeStr);
        };
    }
}

4. 使用工具类加载DataFrame #

在业务代码中调用工具类，加载指定表的Schema并读取CSV文件：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;

import java.io.IOException;

public class SparkCsvReader {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("SchemaFromConfig")
                .master("local[*]")
                .getOrCreate();

        String configPath = "src/main/resources/schema_config.json";

        try {
            // 加载emp_dept表的Schema
            StructType deptSchema = SchemaLoader.loadSchema(configPath, "emp_dept");
            Dataset<Row> df1 = spark.read().format("csv")
                    .option("header", "true")
                    .schema(deptSchema)
                    .csv("path/to/emp_dept.csv");

            // 加载emp_details表的Schema
            StructType detailsSchema = SchemaLoader.loadSchema(configPath, "emp_details");
            Dataset<Row> df2 = spark.read().format("csv")
                    .option("header", "true")
                    .schema(detailsSchema)
                    .csv("path/to/emp_details.csv");

            // 验证数据
            df1.printSchema();
            df1.show();
            df2.printSchema();
            df2.show();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            spark.stop();
        }
    }
}

可选：使用YAML配置文件 #

如果你更偏好YAML的简洁格式，可以创建schema_config.yml：

emp_dept:
  - name: emp_dept
    type: string
    nullable: true
  - name: empid
    type: integer
    nullable: true
  - name: empdesignation
    type: string
    nullable: true
  - name: emp_salary
    type: integer
    nullable: true
emp_details:
  - name: emp_details
    type: string
    nullable: true
  - name: empid
    type: integer
    nullable: true
  - name: empfistname
    type: string
    nullable: true
  - name: emplastname
    type: integer
    nullable: true

此时需要修改SchemaLoader的读取逻辑，引入SnakeYAML依赖：

<!-- Maven依赖 -->
<dependency>
    <groupId>org.yaml</groupId>
    <artifactId>snakeyaml</artifactId>
    <version>2.2</version>
</dependency>

修改后的loadSchema方法：

import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream;

public static StructType loadSchema(String configPath, String tableName) throws IOException {
    Yaml yaml = new Yaml();
    Map<String, List<FieldConfig>> schemaMap = yaml.load(new FileInputStream(new File(configPath)));

    // 后续逻辑与JSON版本一致
    List<FieldConfig> fieldConfigs = schemaMap.get(tableName);
    if (fieldConfigs == null) {
        throw new IllegalArgumentException("未找到表 " + tableName + " 的Schema配置");
    }

    StructField[] fields = fieldConfigs.stream()
            .map(config -> DataTypes.createStructField(
                    config.getName(),
                    convertStringToDataType(config.getType()),
                    config.isNullable()
            ))
            .toArray(StructField[]::new);

    return DataTypes.createStructType(fields);
}

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