如何在Databricks Python Notebook中读取ADLS Blob的.xpt文件并转CSV

1. 读取ADLS Blob中的.xpt文件 #

在Databricks环境中，可通过以下两种常用方式访问ADLS中的.xpt文件：

方式一：使用ABFS路径直接读取 #

前提是Databricks集群已配置好ADLS访问权限（如服务主体、SAS令牌或托管身份），直接通过ABFS路径读取二进制流：

import xport.v56
from pyspark.sql import SparkSession

# Databricks中默认已初始化SparkSession，可省略此行
spark = SparkSession.builder.appName("XptToCsv").getOrCreate()

# 替换为你的ADLS文件路径
xpt_path = "abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<文件子路径>/example.xpt"
# 读取二进制文件并加载为xport库对象
with spark.sparkContext.binaryFiles(xpt_path).collect()[0][1] as f:
    library = xport.v56.load(f)

方式二：使用Azure Storage SDK读取 #

需要更精细控制时，可借助azure-storage-blob库操作：

import xport.v56
from azure.storage.blob import BlobServiceClient

# 替换为你的ADLS连接字符串
connection_string = "<ADLS存储账户连接字符串>"
blob_service_client = BlobServiceClient.from_connection_string(connection_string)

# 替换为对应的容器名和Blob路径
container_name = "<容器名>"
blob_name = "<文件子路径>/example.xpt"
blob_client = blob_service_client.get_blob_client(container=container_name, blob=blob_name)

# 下载Blob并加载为xport库对象
with blob_client.download_blob() as f:
    library = xport.v56.load(f)

2. 转换为CSV格式并保存到ADLS #

xport加载后的library对象包含SAS数据集，将其转为Pandas DataFrame后即可导出为CSV，再保存回ADLS：

import pandas as pd

# 提取库中的数据集（若有多个数据集，需遍历library.values()）
dataset = next(iter(library.values()))
df = pd.DataFrame(dataset)

# 替换为CSV输出路径
csv_output_path = "abfss://<容器名>@<存储账户名>.dfs.core.windows.net/<输出子路径>/output.csv"

# 方式一：用Pandas直接保存（适合小数据量）
df.to_csv(csv_output_path, index=False)

# 方式二：转为Spark DataFrame保存（适合大数据量，支持分布式处理）
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("XptToCsv").getOrCreate()
spark_df = spark.createDataFrame(df)
spark_df.write.mode("overwrite").option("header", "true").csv(csv_output_path)

关键注意事项 #

• 执行代码前，确保集群已安装依赖库：%pip install xport azure-storage-blob。
• 确认Databricks集群对目标ADLS容器拥有读写权限，权限配置方式根据实际场景选择服务主体、SAS令牌或托管身份。

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