如何将pandas DataFrame列值传入KQL查询验证是否存在于ADX中

方案1：使用azure-kusto-data SDK参数化批量查询（推荐） #

该方案适合数据量较大的场景，一次性提交所有待校验值查询，避免逐行请求的性能损耗和限流风险。
首先安装依赖：

pip install azure-kusto-data pandas

实现代码示例：

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.helpers import dataframe_from_result_table
import pandas as pd

# 初始化ADX客户端，根据实际认证方式替换连接串构造逻辑
cluster = "你的ADX集群地址（例：https://xxx.region.kusto.chinacloudapi.cn）"
db = "你的数据库名称"
kcsb = KustoConnectionStringBuilder.with_aad_device_authentication(cluster)
client = KustoClient(kcsb)

# 提取pandas中待校验的列值，去重去空减少无效查询
check_values = df["col_name"].dropna().unique().tolist()

# 构造参数化KQL
query = """
table_name
| where value in dynamic(@check_values)
| distinct value
"""
# 传入参数执行查询
response = client.execute(db, query, parameters={"check_values": check_values})
# 提取ADX中存在的目标值
adx_exists_values = dataframe_from_result_table(response.primary_results[0])["value"].tolist()

# 对比得到ADX中不存在的值
not_exist_values = [v for v in check_values if v not in adx_exists_values]
print("ADX中不存在的取值：", not_exist_values)

方案2：Jupyter %%kql 魔法命令变量传递 #

如果使用Jupyter环境的kqlmagic扩展，直接通过$变量名即可引用Python上下文的变量，你的原始写法只需要补充结果接收逻辑即可实现逐行校验：

# 提前配置好kql魔法的ADX集群、数据库默认连接
for idx, val in enumerate(df["col_name"].dropna()):
    # -o 参数指定将查询结果赋值给Python变量query_result
    %%kql -o query_result
    table_name
    | where value == $val
    | limit 1

    if len(query_result) == 0:
        print(f"第{idx}行取值{val}不存在于ADX中")

注意事项 #

• 逐行查询仅适合数据量小于100条的场景，数据量较大时必须使用批量查询方案
• 所有参数化查询场景都不需要手动拼接字符串加引号，SDK会自动处理类型匹配，避免注入风险
• 待校验列存在空值时建议提前过滤，避免产生无效查询

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