如何在Databricks用Python把对账失败的Spark DataFrame发至Azure Application Insights

问题描述 #

我有一个存储对账数据的Spark DataFrame，仅在对账失败时才会创建。当该DataFrame在Databricks中生成后，需要将其中的数据发送至Azure Application Insights。目前已写出部分代码，但卡在了发送数据到App Insights的环节，了解到可以用OpenTelemetry库，但微软文档没明确说明我的场景怎么用，求针对reconciliation_DF的Python代码示例。

原代码片段：

try:
    # Check count of reconciliation Data Frame and send to Application Insights if count > 0, meaning there's one or more reconciliation error(s).
    if reconciliation_DF.count() > 0:
       reconciliation_DF_json = reconciliation_DF.toJSON().collect()   # Convert Data Frame to JSON
       reconciliation_DF_json = json.dumps(reconciliation_DF_json)    # Convert list of JSONs to string for logging

       # Some code here to write this "reconciliation_DF_json" data to Azure Application Insights???
       # This is where I am stuck.

    else:
         print("No reconciliation errors found")

except Exception as e:
    print("An error occurred: {}".format(e))
    raise

解决方案：使用OpenTelemetry发送数据到Azure Application Insights #

步骤1：在Databricks中安装必要库 #

在Databricks集群上安装以下Python包：

• opentelemetry-api
• opentelemetry-sdk
• opentelemetry-exporter-azure-monitor

可通过集群库管理界面安装，或在Notebook中运行：

%pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-azure-monitor

步骤2：完整代码实现 #

下面是集成OpenTelemetry的完整代码，可直接替换原代码中的待实现部分：

import json
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.azure.monitor import AzureMonitorTraceExporter

# 替换为你的Application Insights连接字符串
# 生产环境建议用Databricks Secrets存储，避免硬编码
APPINSIGHTS_CONNECTION_STRING = "InstrumentationKey=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx;IngestionEndpoint=https://xxx.applicationinsights.azure.com/"

# 初始化OpenTelemetry追踪配置
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# 配置Azure Monitor导出器
exporter = AzureMonitorTraceExporter(connection_string=APPINSIGHTS_CONNECTION_STRING)
span_processor = BatchSpanProcessor(exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

try:
    # 检查对账失败数据
    error_count = reconciliation_DF.count()
    if error_count > 0:
        # 将DataFrame转为JSON格式的错误列表
        reconciliation_errors = reconciliation_DF.toJSON().collect()
        
        # 创建追踪Span，记录对账失败事件
        with tracer.start_as_current_span("reconciliation-failure") as span:
            # 添加基础属性，便于后续筛选分析
            span.set_attribute("reconciliation.status", "failed")
            span.set_attribute("total_error_count", error_count)
            
            # 逐条发送错误数据到App Insights
            for idx, error_json in enumerate(reconciliation_errors):
                error_data = json.loads(error_json)
                span.add_event(
                    name=f"reconciliation_error_{idx+1}",
                    attributes=error_data
                )
        
        print(f"已向Application Insights发送{error_count}条对账失败数据")
    else:
        print("未发现对账失败数据")

except Exception as e:
    error_msg = str(e)
    print(f"处理过程中出错：{error_msg}")
    # 将异常也发送到App Insights
    with tracer.start_as_current_span("reconciliation_process_error") as span:
        span.set_attribute("error.message", error_msg)
        span.record_exception(e)
    raise

关键说明 #

• 连接字符串：需替换为Azure门户中Application Insights资源的实际连接字符串。生产环境建议用dbutils.secrets.get()从Databricks密钥范围读取，避免硬编码敏感信息。
• 数据发送方式：将每条对账错误以事件形式发送，同时添加总错误数、对账状态等属性，方便在App Insights的日志查询中快速筛选和分析。
• 异常处理：额外添加了流程异常的上报逻辑，便于排查整个对账流程中的问题。

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