如何在Apache Spark Java中配置partitionExprs并行读取SQL Server

问题1：VARCHAR类型列的lower/upper bound配置说明 #

• Spark JDBC原生的partitionColumn+lowerBound/upperBound机制仅支持**数值类型（int、long等）**列，无法直接用于VARCHAR类型的主键。因此不需要为VARCHAR列配置这两个参数，强行配置会导致分区失败。
• 替代方案：可以通过两种方式实现并行读取：
  1. 基于哈希函数将VARCHAR列映射为数值分区键
  2. 手动定义分区表达式，拆分数据范围

问题2：Java示例代码 #

方案1：利用哈希函数实现分区读取 #

通过SQL Server的HASHBYTES函数将VARCHAR主键转换为数值，以此作为分区依据。这种方式不需要提前知道数据边界，适合分布均匀的字母数字主键。

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

public class SparkMSSQLVarcharPartition {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("ReadMSSQLVarcharParallel")
                .master("local[*]") // 生产环境请移除该配置
                .getOrCreate();

        String connectionUrl = "jdbc:sqlserver://<server-name>:<port>;databaseName=<db-name>";
        String user = "<user-name>";
        String password = "<password>";
        int numPartitions = 5;

        // 定义带哈希分区逻辑的查询语句
        String query = String.format(
                "SELECT *, ABS(CAST(HASHBYTES('SHA1', <varchar-primary-key>) AS BIGINT)) %% %d AS partition_key " +
                "FROM dbo.<table-name>",
                numPartitions
        );

        Dataset<Row> dataset = spark.read()
                .format("com.microsoft.sqlserver.jdbc.spark")
                .option("url", connectionUrl)
                .option("user", user)
                .option("password", password)
                .option("fetchsize", 10000)
                .option("dbtable", "(" + query + ") AS partitioned_data")
                .option("partitionColumn", "partition_key")
                .option("lowerBound", 0)
                .option("upperBound", numPartitions - 1)
                .option("numPartitions", numPartitions)
                .load();

        // 导出为多个CSV文件（每个分区对应一个文件）
        dataset.drop("partition_key") // 移除临时分区列
                .write()
                .mode("overwrite")
                .option("header", "true")
                .csv("<output-path>");

        spark.stop();
    }
}

方案2：手动定义分区表达式并行读取 #

如果字母数字主键有可拆分的规则（比如按首字符范围），可以手动定义多个分区查询，并行读取后合并数据集。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import java.util.ArrayList;
import java.util.List;

public class SparkMSSQLManualPartition {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("ReadMSSQLManualParallel")
                .master("local[*]") // 生产环境请移除该配置
                .getOrCreate();

        String connectionUrl = "jdbc:sqlserver://<server-name>:<port>;databaseName=<db-name>";
        String user = "<user-name>";
        String password = "<password>";

        // 定义分区规则：按VARCHAR主键的首字符范围拆分，可根据实际数据调整
        List<String> partitionFilters = new ArrayList<>();
        partitionFilters.add("<varchar-primary-key> >= 'A' AND <varchar-primary-key> < 'F'");
        partitionFilters.add("<varchar-primary-key> >= 'F' AND <varchar-primary-key> < 'K'");
        partitionFilters.add("<varchar-primary-key> >= 'K' AND <varchar-primary-key> < 'P'");
        partitionFilters.add("<varchar-primary-key> >= 'P' AND <varchar-primary-key> < 'U'");
        partitionFilters.add("<varchar-primary-key> >= 'U' AND <varchar-primary-key> <= 'Z'");
        partitionFilters.add("<varchar-primary-key> >= '0' AND <varchar-primary-key> <= '9'");

        // 并行读取每个分区的数据
        List<Dataset<Row>> partitionDatasets = new ArrayList<>();
        for (String filter : partitionFilters) {
            Dataset<Row> partDs = spark.read()
                    .format("com.microsoft.sqlserver.jdbc.spark")
                    .option("url", connectionUrl)
                    .option("user", user)
                    .option("password", password)
                    .option("fetchsize", 10000)
                    .option("query", "SELECT * FROM dbo.<table-name> WHERE " + filter)
                    .load();
            partitionDatasets.add(partDs);
        }

        // 合并所有分区数据集
        Dataset<Row> fullDataset = spark.emptyDataFrame();
        for (Dataset<Row> ds : partitionDatasets) {
            fullDataset = fullDataset.union(ds);
        }

        // 导出为多个CSV文件
        fullDataset.write()
                .mode("overwrite")
                .option("header", "true")
                .csv("<output-path>");

        spark.stop();
    }
}

注意事项 #

• 方案1的哈希分区需确保数据分布均匀，避免部分分区数据量过大
• 方案2的手动分区需根据实际数据的字符分布调整规则，保证各分区数据量大致相当
• 生产环境中移除master("local[*]")配置，由集群管理器分配资源

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