如何在批量模式下运行Spark 3.3.0 Cosmos OLTP连接器并解决写入问题

问题背景 #

在Spark 3.3.0、Scala 2.12、Java 11环境中使用cosmos.oltp连接器，已成功实现批量读取Cosmos DB数据，但写入操作时作业持续处于运行状态无法完成。推测连接器默认采用流模式，需确认是否可切换为批量模式或禁用默认流行为。

当前使用的Maven依赖：

<dependency>
   <groupId>com.azure.cosmos.spark</groupId>
   <artifactId>azure-cosmos-spark_3-3_2-12</artifactId>
   <version>4.19.0</version>
</dependency> 

读取代码（已成功运行）：

Dataset<Row> cosmosDB = spark.read()
        .format("cosmos.oltp")
        .option("spark.cosmos.accountEndpoint","Endpoint")
        .option("spark.cosmos.accountKey","master_key")
        .option("spark.cosmos.database","database")
        .option("spark.cosmos.container","container")
        .option("preferredRegions","Region")
        .option("spark.cosmos.read.inferSchema.enabled","true")
        .option("spark.cosmos.read.inferSchema.includeSystemProperties","true")
        .option("spark.cosmos.read.partitioning.strategy","Restrictive")
        .load();

解决方案：明确配置批量写入策略 #

cosmos.oltp连接器默认即为批量模式，流模式需显式配置为cosmos.changeFeed。作业挂起并非因为流模式，而是写入时缺少必要的批量配置参数。需在写入代码中添加以下关键配置：

完整写入示例代码

cosmosDB.write()
        .format("cosmos.oltp")
        .option("spark.cosmos.accountEndpoint","Endpoint")
        .option("spark.cosmos.accountKey","master_key")
        .option("spark.cosmos.database","database")
        .option("spark.cosmos.container","container")
        .option("preferredRegions","Region")
        // 明确启用批量写入策略
        .option("spark.cosmos.write.strategy", "ItemBatch")
        // 控制每个批量请求的文档数量（可根据数据大小调整，默认100）
        .option("spark.cosmos.write.batch.size", "100")
        // 指定写入模式（必须设置，避免行为不确定）
        .mode(SaveMode.Append)
        .save();

关键参数说明 #

• spark.cosmos.write.strategy: 设置为ItemBatch强制使用批量写入策略，将多个文档打包为单个请求发送，提升写入效率
• spark.cosmos.write.batch.size: 调整每个批量请求包含的文档数量，若单文档较大可适当减小数值，避免请求超时
• SaveMode: 必须指定写入模式（Append/Overwrite/Upsert/Ignore），明确数据写入的行为逻辑

额外排查方向 #

若配置后仍出现作业挂起，可检查：

• Cosmos DB容器的吞吐量（RU/s）是否足够，写入时是否出现限流
• 数据中的分区键是否正确设置，避免热点分区导致写入阻塞
• 文档是否存在格式错误（如缺失必填字段、数据类型不匹配）

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