如何通过Spark连接器向CosmosDB图顶点的属性追加多个值

问题根因 #

你使用的spark.cosmos.write.strategy = ItemAppend是SQL API的通用文档写入策略，仅支持普通文档的顶级属性追加，无法兼容Gremlin图模型的顶点多值属性结构。
Gremlin顶点的多值属性本质是存储{id: UUID, _value: 实际值}结构的数组，你当前使用的to_cosmosdb_vertices函数每次都会生成单元素的属性数组，直接使用ItemAppend会直接覆盖原有属性数组，不会执行追加逻辑。

解决方案 #

方案1：批量合并属性后写入（适用大规模数据更新） #

先读取需要更新的顶点现有属性，合并新值后使用ItemOverwrite策略回写，兼容性最好：

from pyspark.sql.functions import udf, lit
from pyspark.sql.types import ArrayType, StructType, StructField, StringType
import uuid

# 定义Gremlin属性数组结构
attr_schema = ArrayType(StructType([
    StructField("id", StringType(), True),
    StructField("_value", StringType(), True)
]))

# 属性追加UDF
@udf(returnType=attr_schema)
def append_gremlin_attr(old_attr_arr, new_value):
    old_attr_arr = old_attr_arr or []
    old_attr_arr.append({
        "id": str(uuid.uuid4()),
        "_value": new_value
    })
    return old_attr_arr

# 1. 读取现有待更新顶点
existing_v = spark.read.format("cosmos.oltp")\
    .options(**cosmosDbConfig)\
    .query("SELECT id, name FROM c WHERE c.id = 'a'")\
    .load()

# 2. 待写入的新属性数据
new_v = spark.createDataFrame([("a", "Alice", None)], ["id", "name", "age"])\
    .withColumn("entity", lit("person"))

# 3. 关联合并属性
merged_v = new_v.join(existing_v, on="id", how="left")\
    .withColumn("name", append_gremlin_attr(existing_v.name, new_v.name))\
    .select("id", "entity", "name", "age")

# 4. 转换为顶点格式后覆盖写入
cosmosDbVertices = to_cosmosdb_vertices(merged_v, "entity")
cosmosDbVertices.write.format("cosmos.oltp")\
    .mode("append")\
    .option("spark.cosmos.write.strategy", "ItemOverwrite")\
    .options(**cosmosDbConfig)\
    .save()

方案2：补丁写入（适用3.10+版本Spark连接器，性能最优） #

3.10及以上版本的CosmosDB Spark连接器支持局部补丁更新，可以直接对数组属性执行追加操作，不需要预读现有数据：

# 复制原有配置，替换为补丁写入策略
patch_config = cosmosDbConfig.copy()
patch_config.update({
    "spark.cosmos.write.strategy": "ItemPatch",
    "spark.cosmos.write.patch.defaultOperationType": "Add",
    # /name/- 表示向name数组末尾追加元素
    "spark.cosmos.write.patch.columnMappings": "name=/name/-"
})

# 直接写入待追加的属性数据即可
new_v = spark.createDataFrame([("a", "Alice")], ["id", "name"])
new_v.write.format("cosmos.oltp")\
    .mode("append")\
    .options(**patch_config)\
    .save()

方案3：提交Gremlin语句执行追加（适用小批量数据） #

如果更新量不大，可以直接调用Gremlin端点执行原生追加语句，逻辑最简单：

from gremlin_python.driver import client, serializer

gremlin_client = client.Client(
    "wss://<你的CosmosDB账号名>.gremlin.cosmos.azure.com:443/",
    "g",
    username="/dbs/<数据库名>/colls/<容器名>",
    password="<CosmosDB主键>",
    message_serializer=serializer.GraphSONSerializersV2d0()
)

# 执行原生Gremlin追加语句
gremlin_client.submit("g.V('a').property(list, 'name', 'Alice')").all().result()
gremlin_client.close()

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