Java读取Azure Data Lake Delta数据报错排查求助

代码实现 #

package com.example;

import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public class App 
{
    public static void main( String[] args )
    {
        System.out.println( "Hello World New!" );
        
        String azureClientId = "<azure-client-id>";
        String azureClientSecret = "<azure-client-secret>";
        String azureStorageAccountName = "<storage-name>";

        SparkConf conf = new SparkConf().setAppName("PSDeltaLakeDemoRG");
        SparkSession spark = SparkSession.builder().config(conf).config("spark.master", "local").getOrCreate();

        spark.conf().set("fs.azure.account.auth.type", "OAuth");
        spark.conf().set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider");
        spark.conf().set("fs.azure.account.auth.oauth2.client.id", azureClientId);
        spark.conf().set("fs.azure.account.auth.oauth2.client.secret", azureClientSecret);
        spark.conf().set("fs.azure.account.auth.oauth2.client.endpoint", "https://login.microsoftonline.com/<tenant-id>/oauth2/token");
        spark.conf().set("spark.databricks.delta.preview.enabled", "true");
        
        String containerName = "<container-name>";
        String filePath = "mnt/datalake/Output/YellowTaxis.delta";

        String azurePath = "abfss://" + containerName + "@" + azureStorageAccountName + ".dfs.core.windows.net/" + filePath;
        
        Dataset<Row> data = spark.read().format("delta").load(azurePath);
        data.show();

        spark.stop();
    }
}

错误信息 #

WARNING: All illegal access operations will be denied in a future release
ANTLR Tool version 4.9.3 used for code generation does not match the current runtime version 4.7ANTLR Runtime version 4.9.3 used for parser compilation does not match the current runtime version 4.7ANTLR Tool version 4.9.3 used for code generation does not match the current runtime version 4.7ANTLR Runtime version 4.9.3 used for parser compilation does not match the current runtime version 4.7Exception in thread "main" Unable to load key provider class.
        at org.apache.hadoop.fs.azurebfs.AbfsConfiguration.getTokenProvider(AbfsConfiguration.java:477)
        at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.initializeClient(AzureBlobFileSystemStore.java:814)
        at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.<init>(AzureBlobFileSystemStore.java:149)
        at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.initialize(AzureBlobFileSystem.java:108)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3469)
        at org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174)
        at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3574)
        at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3521)
        at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:540)
        at org.apache.hadoop.fs.Path.getFileSystem(Path.java:365)
        at org.apache.spark.sql.delta.DeltaTableUtils$.findDeltaTableRoot(DeltaTable.scala:164)
        at org.apache.spark.sql.delta.sources.DeltaDataSource$.parsePathIdentifier(DeltaDataSource.scala:267)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.x$1$lzycompute(DeltaTableV2.scala:67)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.x$1(DeltaTableV2.scala:62)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.rootPath$lzycompute(DeltaTableV2.scala:62)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.rootPath(DeltaTableV2.scala:62)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.deltaLog$lzycompute(DeltaTableV2.scala:73)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.deltaLog(DeltaTableV2.scala:73)
        at org.apache.spark.sql.delta.catalog.DeltaTableV2.toBaseRelation(DeltaTableV2.scala:139)
        at org.apache.spark.sql.delta.sources.DeltaDataSource.createRelation(DeltaDataSource.scala:177)
        at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:346)
        at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:229)
        at org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:211)
        at scala.Option.getOrElse(Option.scala:189)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:186)
        at com.example.App.main(App.java:48)

当前依赖配置 #

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.5.0</version> <!-- Use the appropriate Spark version -->
    </dependency>
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-core_2.12</artifactId>
        <version>1.0.0</version> <!-- Use the appropriate Delta Lake version -->
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.12</artifactId>
      <version>3.5.0</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-azure</artifactId>
      <version>3.2.0</version> <!-- Use the appropriate Hadoop Azure version -->
    </dependency>

注：在Azure工作区使用相同配置可正常访问数据，需排查是否遗漏配置或依赖。

排查与解决方法 #

1. 核心问题：依赖版本严重不兼容 #

• Delta Lake版本：Spark 3.5.0对应的Delta Lake最低版本为2.4.0，当前使用的1.0.0版本与Spark 3.5完全不兼容，会导致大量API冲突。
• Hadoop版本匹配：Spark 3.5.0默认依赖Hadoop 3.3.x，当前hadoop-azure 3.2.0版本过低，需升级到3.3.x系列（如3.3.6）。
• spark-sql scope修正：本地运行时需去掉<scope>provided</scope>，否则会缺失SQL模块依赖。

2. 缺失OAuth认证依赖 #

报错Unable to load key provider class是因为缺少Azure AD OAuth认证所需的配套依赖，需添加：

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-azure-datalake</artifactId>
    <version>3.3.6</version>
</dependency>
<dependency>
    <groupId>com.microsoft.azure</groupId>
    <artifactId>azure-storage</artifactId>
    <version>8.6.6</version>
</dependency>

3. 配置项修正 #

OAuth配置的键名有误，需将fs.azure.account.auth.oauth2.*改为fs.azure.account.oauth2.*，正确配置如下：

spark.conf().set("fs.azure.account.auth.type", "OAuth");
spark.conf().set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider");
spark.conf().set("fs.azure.account.oauth2.client.id", azureClientId);
spark.conf().set("fs.azure.account.oauth2.client.secret", azureClientSecret);
spark.conf().set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<tenant-id>/oauth2/token");
// 移除过时配置
// spark.conf().set("spark.databricks.delta.preview.enabled", "true");

4. 完整依赖示例 #

调整后的Maven依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.5.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.5.0</version>
    </dependency>
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-core_2.12</artifactId>
        <version>2.4.0</version>
        <exclusions>
            <exclusion>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-sql_2.12</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-azure</artifactId>
        <version>3.3.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-azure-datalake</artifactId>
        <version>3.3.6</version>
    </dependency>
    <dependency>
        <groupId>com.microsoft.azure</groupId>
        <artifactId>azure-storage</artifactId>
        <version>8.6.6</version>
    </dependency>
</dependencies>

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