本地Azurite中Azure Table Storage批量处理失败问题求助

需要将超过100个实体拆分批次提交至本地Azure存储账户（Azurite），已实现批次拆分逻辑（按50个实体/批次），但出现部分批次成功、部分失败的情况（例如107个实体时，前50个成功，中间50个失败，剩余7个成功）。尝试添加sleep后问题未解决，报错信息如下：

用户实现代码：

def save_results_in_database(self, operations, table_name):
        table = self.table_service_client.get_table_client(table_name)
        if table_name is None:
            raise Exception(f"Table {table_name} does not exist !")
        try:
            table.submit_transaction(operations)
        except Exception as e:
            logging.exception(
                f"Error while saving results in database for table {table_name}"
            )

if 0 < len(results) < 100:
        helpers.save_results_in_database(results, action)
    else:
        # if the number of results is greater than 100, we need to split them in batches of 50
        # and save them in batches
        batch_size = 50
        for i in range(0, len(results), batch_size):
            batch = results[i : i + batch_size]
            helpers.save_results_in_database(batch, action)
            sleep(10)

报错信息：

Error while saving results in database for table <table name>
File "/usr/local/lib/python3.10/site-packages/azure/data/tables/_table_client.py", line 734, in submit_transaction
return self._batch_send(self.table_name, *batched_requests.requests, **kwargs)  # type: ignore
 File "/usr/local/lib/python3.10/site-packages/azure/data/tables/_base_client.py", line 334, in _batch_send
    raise decoded
azure.data.tables._error.TableTransactionError: 2:An error occurred while processing this request.
ErrorCode:InvalidInput
Content: {"odata.error":{"code":"InvalidInput","message":{"lang":"en-US","value":"2:An error occurred while processing this request.\nRequestId:994b1eca-1109-4ff9-aa56-0dc73ae97a18\nTime:2023-04-06T13:08:29.278Z"}}}

已确认输入Schema有效，请求排查遗漏点。

• 批次内或跨批次存在重复主键组合：Azurite遵循Azure Table Storage规则，每个实体的PartitionKey + RowKey必须全局唯一。若失败批次中存在与已成功提交实体重复的主键组合，会触发InvalidInput错误。需检查所有实体的主键组合，确保无重复。
• Azurite事务处理限制：作为本地模拟器，Azurite在连续批量请求时可能存在资源锁或事务处理瓶颈。可尝试缩小批次大小（如调整为25个实体/批次），或在提交后增加动态等待逻辑（而非固定sleep）。
• 单个实体数据异常：即使Schema有效，个别实体可能存在隐藏格式问题（如字段值超出长度限制、包含特殊字符）。可单独提交失败批次中的每个实体，定位具体问题实体并修正数据。
• Azurite版本兼容性：旧版本Azurite可能存在批量操作的已知bug，建议升级至最新版本后重新测试。

from azure.data.tables._error import TableTransactionError
import time
import logging

def save_results_in_database(self, operations, table_name):
    table = self.table_service_client.get_table_client(table_name)
    if not table_name:
        raise ValueError(f"Table name cannot be None!")
    
    # 提前检查批次内主键重复
    seen_keys = set()
    for op in operations:
        entity = op.entity
        key_pair = (entity["PartitionKey"], entity["RowKey"])
        if key_pair in seen_keys:
            raise ValueError(f"Duplicate entity key detected: {key_pair}")
        seen_keys.add(key_pair)
    
    # 带重试机制的提交逻辑
    max_retries = 3
    retry_delay = 5
    for attempt in range(max_retries):
        try:
            table.submit_transaction(operations)
            logging.info(f"Successfully submitted batch with {len(operations)} entities")
            return
        except TableTransactionError as e:
            logging.warning(f"Batch failed (attempt {attempt+1}/{max_retries}): {str(e)}")
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
            else:
                logging.exception(f"Failed to submit batch after {max_retries} attempts")
                raise
        except Exception as e:
            logging.exception(f"Unexpected error saving results for table {table_name}")
            raise

# 调用逻辑
if results:
    batch_size = 25  # 缩小批次大小测试
    for i in range(0, len(results), batch_size):
        batch = results[i:i+batch_size]
        helpers.save_results_in_database(batch, action)
        time.sleep(2)

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