如何判断Azure Kusto队列式数据导入操作是否完成？

使用QueuedIngestClient进行数据导入时，它采用异步队列机制——调用ingest_from_dataframe后，数据仅被提交到导入队列，返回的IngestionResult的status始终会是QUEUED，不会自动跟踪后续处理状态。要在程序中确认导入是否完成，可以通过以下两种方式实现：

1. 通过操作ID查询导入状态 #

每个IngestionResult都会包含唯一的operation_id，你可以用这个ID通过Kusto管理命令查询具体的处理状态：

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.exceptions import KustoServiceError

# 构建集群管理连接（需使用集群URI，格式为https://<cluster-name>.kusto.windows.net）
cluster_uri = "<你的集群URI>"
kcsb = KustoConnectionStringBuilder.with_aad_device_authentication(cluster_uri)
admin_client = KustoClient(kcsb)

# 从导入响应中获取操作ID
operation_id = response.operation_id
# 构造查询命令，指定目标数据库
query = f".show ingestion operations where OperationId == '{operation_id}'"

try:
    # 执行查询
    result = admin_client.execute("<目标数据库名称>", query)
    # 解析查询结果
    for row in result.primary_results[0]:
        print(f"当前状态: {row['Status']}")
        print(f"完成时间: {row['IngestionCompletionTime']}")
        print(f"错误信息: {row['Details']}")
except KustoServiceError as e:
    print(f"查询状态失败: {e}")

查询结果中的Status字段会显示当前导入的实际状态，可能值包括InProgress（处理中）、Succeeded（成功完成）、Failed（失败）等。

2. 查询目标表的导入统计信息 #

你可以通过查询目标表的导入统计，确认是否有对应的数据批次完成导入：

# 构造查询命令
query = ".show table <目标表名称> ingestion statistics"

result = admin_client.execute("<目标数据库名称>", query)
# 解析结果，查看最新的导入记录
for row in result.primary_results[0]:
    print(f"导入批次ID: {row['IngestionId']}")
    print(f"状态: {row['Status']}")
    print(f"导入行数: {row['IngestedRecordsCount']}")

额外提示 #

• 队列导入存在一定延迟，Kusto会批量处理队列中的任务，建议设置合理的轮询间隔（比如30秒以上），避免频繁查询。
• 如果导入失败，可以用.show ingestion failures命令排查具体错误原因。
• 如果需要更实时的状态反馈，可考虑使用ManagedStreamingIngestClient，它更适合低延迟的导入场景。

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