向Cosmos Gremlin上传JSON时分区键为空错误的解决求助

尝试将JSON文件上传至Azure Cosmos DB - Gremlin API，已配置分区键为/LOCATIONSTATE，确认JSON中该字段无空值，但仍报错：Cannot add a vertex where the partition key property has value 'null'。更换分区键为/LOCATIONZIP或/LOCATIONLOOKUPID后错误依旧，怀疑是空字符串导致，寻求解决方法。

核心原因 #

• JSON数据读取路径不匹配：脚本中遍历json_data["XYZ"]，但示例JSON的顶级键是"XYZ Grocery"，导致无法读取有效数据，后续分区键变量实际未获取到有效值，最终传递空值。
• 分区键属性名错误：Cosmos DB Gremlin API要求分区键的属性名必须与容器配置的实际字段名一致，而非固定写partitionKey。比如配置的分区键是LOCATIONSTATE，就需要设置.property('LOCATIONSTATE', '{value}')，而非.property('partitionKey', ...)。
• 重复创建同类型顶点：脚本中重复创建location_state类型顶点，且顶点ID重复使用location_state值，引发逻辑冲突，同时干扰分区键的正常传递。

修复步骤 #

1. 修正JSON读取路径：将遍历语句改为for record in json_data["XYZ Grocery"]，匹配实际JSON结构。
2. 对齐分区键属性名：所有addV语句中，将.property('partitionKey', '{location_state}')替换为.property('LOCATIONSTATE', '{location_state}')（与容器配置的分区键字段名保持一致）。
3. 移除重复顶点创建逻辑：删除最后一段重复创建location_state顶点的代码，避免重复操作和ID冲突。
4. 优化空值处理：确保分区键值经过严格非空校验，避免空字符串或null传递。

修改后的Python脚本 #

import json
import requests
from gremlin_python.driver import client, serializer
import uuid  # To generate unique IDs if 'id' is missing
from azure.storage.filedatalake import DataLakeServiceClient

# Replace with your own details
endpoint = ""
database = ""
graph = ""
primary_key = ""

# ADLS connection details
adls_account_url = ""
adls_file_system = "json"
adls_file_path = ""

# Azure Data Lake Storage Client
def get_adls_client():
    service_client = DataLakeServiceClient(account_url=adls_account_url, credential="")
    return service_client

def read_json_file_from_adls(file_name):
    service_client = get_adls_client()
    file_system_client = service_client.get_file_system_client(file_system=adls_file_system)
    file_client = file_system_client.get_file_client(file_name)

    file_content = file_client.download_file()
    return json.loads(file_content.readall().decode('utf-8'))

# Initialize the Gremlin client
gremlin_client = client.Client(
    endpoint,
    'g',
    username=f"/dbs/{database}/colls/{graph}",
    password=primary_key,
    message_serializer=serializer.GraphSONSerializersV2d0()
)

# Function to process a single JSON file and create the graph
def process_json_file(file_name):
    # Read the JSON file from ADLS
    json_data = read_json_file_from_adls(file_name)

    print(f"Processing file: {file_name}")

    # 修正：匹配JSON实际顶级键名
    for record in json_data["XYZ Grocery"]:
        # Use LOCATIONSTATE as the partition key
        location_state = record.get("LOCATIONSTATE", "").strip()
        if not location_state:
            location_state = str(uuid.uuid4())  # Fallback to a generated unique ID
            record["LOCATIONSTATE"] = location_state  # Optionally update the record with the new LOCATIONSTATE

        location_name = record.get("LOCATIONNAME", "").replace("'", "\\'")
        location_city = record.get("LOCATIONCITY", "").replace("'", "\\'")
        # 直接使用已处理好的location_state，避免重复读取原数据
        location_state_clean = location_state.replace("'", "\\'")

        # Create LOCATIONSTATE vertex
        try:
            create_location_state_query = (
                f"g.addV('location_state').property('id', '{location_state_clean}')"
                f".property('state', '{location_state_clean}')"
                # 修正：分区键属性名与容器配置一致
                f".property('LOCATIONSTATE', '{location_state_clean}')"
            )
            gremlin_client.submit(create_location_state_query).all().result()
            print(f"Created location_state vertex: {location_state_clean}")
        except Exception as e:
            print(f"Error creating location_state vertex {location_state_clean}: {e}")
            continue

        # Create LOCATIONNAME vertex and add edge
        location_name_id = str(uuid.uuid4())
        try:
            create_location_name_query = (
                f"g.addV('location_name').property('id', '{location_name_id}')"
                f".property('name', '{location_name}')"
                # 修正：分区键属性名与容器配置一致
                f".property('LOCATIONSTATE', '{location_state_clean}')"
            )
            gremlin_client.submit(create_location_name_query).all().result()

            edge_location_name_query = f"g.V('{location_state_clean}').addE('name').to(g.V('{location_name_id}'))"
            gremlin_client.submit(edge_location_name_query).all().result()
            print(f"Created edge: {location_state_clean} -> {location_name}")
        except Exception as e:
            print(f"Error creating location_name or edge for {location_state_clean}: {e}")

        # Create LOCATIONCITY vertex and add edge
        location_city_id = str(uuid.uuid4())
        try:
            create_location_city_query = (
                f"g.addV('location_city').property('id', '{location_city_id}')"
                f".property('city', '{location_city}')"
                # 修正：分区键属性名与容器配置一致
                f".property('LOCATIONSTATE', '{location_state_clean}')"
            )
            gremlin_client.submit(create_location_city_query).all().result()

            edge_location_city_query = f"g.V('{location_state_clean}').addE('city').to(g.V('{location_city_id}'))"
            gremlin_client.submit(edge_location_city_query).all().result()
            print(f"Created edge: {location_state_clean} -> {location_city}")
        except Exception as e:
            print(f"Error creating location_city or edge for {location_state_clean}: {e}")

# Process the JSON file
try:
    process_json_file(adls_file_path)
except Exception as e:
    print(f"Error processing the file: {e}")

# Close the Gremlin client
gremlin_client.close()

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