Azure ML Studio中使用数据资产无法读取Delta表最新版本的解决方法

核心思路 #

直接通过Delta Lake原生API读取，绕过Azure ML数据资产的文件/表类型限制——普通文件数据集仅读取Parquet文件本身，不会解析_delta_log中的版本控制信息，这是导致读取到已删除行的根本原因。

具体实现步骤 #

1. 配置ADLS Gen2访问权限 #

确保Azure ML计算实例/集群具备ADLS Gen2的访问权限：

• 为计算资源的托管标识赋予存储Blob数据Contributor角色
• 或使用服务主体，在代码运行环境中配置环境变量AZURE_CLIENT_ID、AZURE_TENANT_ID、AZURE_CLIENT_SECRET

2. 安装依赖 #

在Azure ML计算环境中安装Delta Lake与Azure存储相关依赖：

pip install delta-spark azure-storage-file-datalake

3. 读取最新版本的Delta表 #

使用PySpark或Delta Lake Python API读取，示例代码：

from delta.tables import DeltaTable
from pyspark.sql import SparkSession

# 初始化SparkSession（Azure ML计算集群环境已预装Spark，可直接调用）
spark = SparkSession.builder \
    .appName("DeltaTableReader") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# ADLS Gen2路径格式：abfss://<容器名>@<存储账户名>.dfs.core.windows.net/my-data
delta_table_path = "abfss://my-container@mystorageaccount.dfs.core.windows.net/my-data"

# 读取最新版本Delta表，自动过滤已删除行
delta_table = DeltaTable.forPath(spark, delta_table_path)
latest_df = delta_table.toDF()

# 验证数据
latest_df.show()

4. 切换Delta表版本 #

通过指定版本号或时间戳读取历史版本，示例：

# 读取版本2的Delta表（替换为实际版本号）
version_2_df = spark.read.format("delta").option("versionAsOf", 2).load(delta_table_path)
version_2_df.show()

# 通过时间戳读取（格式：yyyy-MM-dd HH:mm:ss）
timestamp_df = spark.read.format("delta").option("timestampAsOf", "2024-05-20 10:00:00").load(delta_table_path)
timestamp_df.show()

5. （可选）在Azure ML中注册Delta表资产（v2 API） #

若需在Azure ML Studio中可视化管理，可通过v2 API注册自定义数据资产，并在元数据中记录Delta表路径：

from azure.ai.ml import MLClient
from azure.ai.ml.entities import Data
from azure.identity import DefaultAzureCredential

ml_client = MLClient(DefaultAzureCredential(), subscription_id="<订阅ID>", resource_group_name="<资源组>", workspace_name="<工作区名>")

delta_data = Data(
    name="my-delta-table",
    version="1",
    path=delta_table_path,
    type="custom",
    description="ADLS Gen2中的Delta表，支持版本切换"
)

ml_client.data.create_or_update(delta_data)

后续使用时，直接从资产中获取路径，再通过Delta Lake API读取即可。

关键说明 #

• 禁止使用Azure ML的文件数据集或表数据集直接读取Delta表，这类资产类型无法解析Delta事务日志，会读取到原始Parquet文件中的已删除行。
• 版本控制功能必须依赖Delta Lake原生API实现，它会自动解析_delta_log中的事务记录，过滤无效数据。

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