含CQL Map与嵌套UDT的Cassandra表批量插入失败排查

错误原因 #

你遇到的KeyError: 0是因为UDT字段名不匹配：

• 你的Cassandra UDT UDTVehicle 定义的第一个字段是num_vehicles（全小写+下划线）
• 但你传入的Vehicles字典里的对应键是num_Vehicles（驼峰式，V大写）
  Cassandra Python驱动在序列化UDT时会严格匹配字段名（包括大小写），找不到对应字段时就会抛出索引0的KeyError（因为num_vehicles是UDT的第一个字段，索引为0）。

修复步骤 #

1. 修正UDT字段名的大小写 #

在构造写入Cassandra的UDT数据时，将字典的键名改为和UDT定义完全一致的小写格式：

# 复制原Vehicle数据，避免修改原record
vehicle_udt = record.get('Vehicles').copy()
# 将驼峰式的num_Vehicles替换为UDT定义的num_vehicles
vehicle_udt['num_vehicles'] = vehicle_udt.pop('num_Vehicles')

2. 优化字段校验逻辑（可选） #

原代码的字段校验逻辑all(record.get(field) is not None for field in list(record.keys()))存在问题——它会检查record里所有的键是否都不为空，但实际上只需要校验表要求的必填字段，改成：

required_fields = ['road_id', 'timestamp', 'radar_id', 'Vehicles']
if not all(record.get(field) is not None for field in required_fields):
    logger.warning("Record is missing required fields")
    return

修改后的完整代码 #

def write_to_cassandra(self,session, record):
        insert_query = """
        INSERT INTO road_traffic (road_id, timestamp, radar_id, vehicles)
        VALUES (?, ?, ?, ?);
        """
        prepared_insert = session.prepare(insert_query)
        batch = BatchStatement()
        batch_size = 0
        batches= 0
        print(record)
        
        # 校验必填字段
        required_fields = ['road_id', 'timestamp', 'radar_id', 'Vehicles']
        if not all(record.get(field) is not None for field in required_fields):
            logger.warning("Record is missing required fields")
            return
        
        # 修正UDT字段名，匹配Cassandra定义
        vehicle_udt = record['Vehicles'].copy()
        if 'num_Vehicles' in vehicle_udt:
            vehicle_udt['num_vehicles'] = vehicle_udt.pop('num_Vehicles')
        
        vehicles_map = {record['road_id']: vehicle_udt}
        print(record['road_id'])
        batch.add(prepared_insert, (record['road_id'], record['timestamp'], record['radar_id'], vehicles_map))
        batch_size += 1

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