如何使用Python向Azure集群导入单条数据以用于KQL查询

前提准备 #

• 安装必要的Python包：

  pip install azure-kusto-ingest azure-kusto-data azure-identity
  

• 拥有Azure Data Explorer集群的访问权限，明确目标数据库名、表名，以及可用的认证方式（如Azure AD应用、用户凭据等）

方法1：使用Kusto摄入客户端（推荐，适配高频小数据场景） #

该方式依托ADX的摄入服务做优化，即使单条数据也能高效处理，后续连续产生的数据还会自动合并缓存，适合每秒生成数据的场景。

示例代码：

from azure.kusto.ingest import KustoIngestClient, IngestionProperties, DataFormat
from azure.identity import DefaultAzureCredential

# 替换为你的集群、数据库、表信息
cluster_uri = "https://<你的集群名>.kusto.windows.net"
database_name = "<你的数据库名>"
table_name = "<你的表名>"

# 初始化认证凭据（支持本地开发、Azure托管环境等多种场景）
credential = DefaultAzureCredential()

# 创建摄入客户端
ingest_client = KustoIngestClient(cluster_uri, credential)

# 配置摄入属性，匹配数据格式
ingestion_props = IngestionProperties(
    database=database_name,
    table=table_name,
    data_format=DataFormat.JSON  # 可根据实际调整为CSV、TEXT等
)

# 单条数据示例（需与目标表schema完全匹配）
single_data = '{"id": 123, "timestamp": "2024-05-20T10:00:00", "value": 45.6}'

# 执行单条数据导入
ingest_client.ingest_from_string(single_data, ingestion_properties=ingestion_props)

print("单条数据导入完成")

方法2：直接通过Kusto查询客户端插入（适合极低频率单条数据） #

如果数据产生频率极低，可直接用KQL的.set-or-append命令写入表，无需经过摄入服务。

示例代码：

from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.identity import DefaultAzureCredential

# 替换为你的集群、数据库、表信息
cluster_uri = "https://<你的集群名>.kusto.windows.net"
database_name = "<你的数据库名>"
table_name = "<你的表名>"

# 构建连接并初始化客户端
kcsb = KustoConnectionStringBuilder.with_azure_identity(cluster_uri)
client = KustoClient(kcsb)

# 构造单条数据插入的KQL命令
insert_command = f"""
.set-or-append {table_name} <|
print id=123, timestamp=datetime(2024-05-20T10:00:00), value=45.6
"""

# 执行插入命令
response = client.execute(database_name, insert_command)
print("插入结果:", response.primary_results[0].to_dict())

关键注意事项 #

• Schema匹配：确保单条数据的格式、字段与目标表结构完全一致，否则会出现导入失败或数据错位。
• 认证选型：生产环境推荐使用Azure AD服务主体认证，避免用户凭据过期导致服务中断。
• 高频优化：若每秒都有数据产生，建议攒小批量（如10条/1秒）再调用导入接口，或传入包含多条数据的DataFrame，减少请求次数提升效率。
• 错误处理：实际部署时需添加异常捕获逻辑，处理网络、权限、数据格式等各类错误。

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