Synapse Spark日志配置求助：ADLS日志写入与REST异常捕获

我用PySpark写了一段代码，功能是调用REST API提取XML格式内容，然后把数据写入数据湖容器的Parquet文件里。现在想给这段代码加日志功能，既要记录错误信息，也要记录执行的操作和流程更新。

作为Spark新手，我查的线上示例大多用1/0这种简单场景演示错误处理和日志，而且日志默认存在本地目录，没法写到ADLS的存储账户/容器/目录层级里，纯Python代码也没法直接运行。我试过用ABFSS和ADL路径往ADLS指定目录写日志文件，但都失败了（日志文件已经提前在对应路径创建好了）。

• 将错误信息写入ADLS存储账户/容器/目录层级下指定文件夹的日志文件；
• 捕获REST相关的特定异常。

LogFilepath = "abfss://raw@<存储账户名>.dfs.core.windows.net/Data/logging/data.log"
#LogFilepath2 = "adl://<存储账户名>.azuredatalakestore.net/raw/Data/logging/data.log"
print(LogFilepath)
try:
    1/0
except Exception as e:
    print('My Error...' + str(e))
with open(LogFilepath, "a") as f:
    f.write("An error occured: {}
".format(e))

一、解决ADLS日志写入问题 #

Python的open()函数只能操作本地文件系统，无法直接访问ABFSS/ADL这类分布式存储路径。在PySpark环境中，要实现ADLS日志写入，可通过以下两种方式：

方法1：使用Azure Data Lake Storage SDK写入日志 #

1. 先安装依赖包（环境未预装时执行）：

pip install azure-storage-file-datalake azure-identity

2. 代码示例：

from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential
import datetime

# 配置ADLS参数
storage_account_name = "<你的存储账户名>"
container_name = "raw"
log_directory = "Data/logging"
log_file_name = "data.log"

# 初始化ADLS客户端（采用DefaultAzureCredential，支持MSI、Azure CLI等身份验证方式）
service_client = DataLakeServiceClient(
    account_url=f"https://{storage_account_name}.dfs.core.windows.net",
    credential=DefaultAzureCredential()
)

# 获取文件系统和目录客户端
file_system_client = service_client.get_file_system_client(file_system=container_name)
directory_client = file_system_client.get_directory_client(log_directory)
file_client = directory_client.get_file_client(log_file_name)

# 日志写入函数
def write_log(message, log_type="INFO"):
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] [{log_type}] {message}\n"
    
    # 追加模式写入：先读取现有内容，再合并写入
    try:
        existing_data = file_client.download_file().readall()
        file_client.upload_data(existing_data + log_entry.encode("utf-8"), overwrite=True)
    except Exception:
        # 文件不存在时直接创建写入
        file_client.upload_data(log_entry.encode("utf-8"), overwrite=False)

# 测试错误捕获与日志写入
try:
    # 替换为你的REST API调用和XML处理逻辑
    1/0
except ZeroDivisionError as e:
    error_msg = f"计算错误: {str(e)}"
    print(error_msg)
    write_log(error_msg, log_type="ERROR")
except Exception as e:
    error_msg = f"未知错误: {str(e)}"
    print(error_msg)
    write_log(error_msg, log_type="ERROR")

方法2：使用Spark RDD写入日志（适合批量日志） #

如果日志量不大，可将日志内容转为RDD，通过Spark的saveAsTextFile写入ADLS：

from pyspark.sql import SparkSession
import datetime

spark = SparkSession.builder.appName("LogToADLS").getOrCreate()

def write_log_spark(message, log_type="INFO"):
    timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] [{log_type}] {message}"
    
    # 创建单元素RDD，设置分区数为1保证单文件输出
    log_rdd = spark.sparkContext.parallelize([log_entry], numSlices=1)
    # 以追加模式写入ADLS
    log_rdd.saveAsTextFile("abfss://raw@<存储账户名>.dfs.core.windows.net/Data/logging/", mode="append")

# 测试
try:
    1/0
except Exception as e:
    write_log_spark(f"错误发生: {str(e)}", "ERROR")

注意：saveAsTextFile会生成分区命名的文件（如part-00000），若需统一文件名，可后续通过ADLS SDK合并文件。

二、捕获REST相关特定异常 #

REST API调用常用requests库，可精准捕获以下特定异常：

import requests

try:
    # 调用REST API
    response = requests.get("https://your-api-url.com/xml-data")
    # 触发HTTP状态码错误（4xx、5xx）
    response.raise_for_status()
    # 处理XML内容
    xml_content = response.text
except requests.exceptions.HTTPError as e:
    error_msg = f"API HTTP错误: {str(e)}"
    write_log(error_msg, "ERROR")
except requests.exceptions.ConnectionError as e:
    error_msg = f"API连接错误: {str(e)}"
    write_log(error_msg, "ERROR")
except requests.exceptions.Timeout as e:
    error_msg = f"API请求超时: {str(e)}"
    write_log(error_msg, "ERROR")
except requests.exceptions.RequestException as e:
    error_msg = f"API请求异常: {str(e)}"
    write_log(error_msg, "ERROR")

这些异常覆盖了REST调用中常见的HTTP错误、连接失败、超时等场景，可精准定位问题。

注意事项 #

• 确保Spark作业的运行身份（服务主体/用户）拥有ADLS容器的写入权限（至少为Storage Blob Data Contributor角色）；
• 使用Azure SDK时优先采用DefaultAzureCredential，避免硬编码密钥，提升安全性；
• 若在Databricks环境运行，ABFSS路径的身份验证会自动处理，无需额外配置。

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