HDInsights 上的 Spark - 方案没有文件系统：adl

Question

我正在编写一个处理来自 ADLS 的文件的应用程序。当尝试通过在 spark-shell 中运行代码从集群中读取文件时，访问文件没有问题。但是，当我尝试在集群上运行项目时，它给了我：

[error] java.io.IOException: No FileSystem for scheme: adl

implicit val spark = SparkSession.builder().master("local[*]").appName("AppMain").getOrCreate()
import spark.implicits._

val listOfFiles = spark.sparkContext.binaryFiles("adl://adlAddressHere/FolderHere/")

val fileList = listOfFiles.collect()

这是基于 HDI 3.6 的 spark 2.2

Answer 1

在你的build.sbt 添加：

libraryDependencies += "org.apache.hadoop" % "hadoop-azure-datalake" % "2.8.0" % Provided

我使用 Spark 2.3.1 而不是 2.2。该版本适用于hadoop-azure-datalake 2.8.0。

然后，配置您的 spark 上下文：

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate()
import spark.implicits._

val hadoopConf = spark.sparkContext.hadoopConfiguration
hadoopConf.set("fs.adl.impl", "org.apache.hadoop.fs.adl.AdlFileSystem")
hadoopConf.set("fs.AbstractFileSystem.adl.impl", "org.apache.hadoop.fs.adl.Adl")
hadoopConf.set("dfs.adls.oauth2.access.token.provider.type", "ClientCredential")
hadoopConf.set("dfs.adls.oauth2.client.id", clientId)
hadoopConf.set("dfs.adls.oauth2.credential", clientSecret)
hadoopConf.set("dfs.adls.oauth2.refresh.url", s"https://login.microsoftonline.com/$tenantId/oauth2/token")

---

TL;DR;

如果您通过 spark 上下文使用 RDD，您可以告诉 Hadoop Configuration 在哪里可以找到您的 org.apache.hadoop.fs.adl.AdlFileSystem 的实现。

键的格式为fs.<fs-prefix>.impl，值是实现类org.apache.hadoop.fs.FileSystem的完整类名。

在您的情况下，您需要由org.apache.hadoop.fs.adl.AdlFileSystem 实现的fs.adl.impl。

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate()
import spark.implicits._

val hadoopConf = spark.sparkContext.hadoopConfiguration
hadoopConf.set("fs.adl.impl", "org.apache.hadoop.fs.adl.AdlFileSystem")

我通常使用 Spark SQL，所以我也需要配置 spark session：

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate()
spark.conf.set("fs.adl.impl", "org.apache.hadoop.fs.adl.AdlFileSystem")
spark.conf.set("dfs.adls.oauth2.access.token.provider.type", "ClientCredential")
spark.conf.set("dfs.adls.oauth2.client.id", clientId)
spark.conf.set("dfs.adls.oauth2.credential", clientSecret)
spark.conf.set("dfs.adls.oauth2.refresh.url", s"https://login.microsoftonline.com/$tenantId/oauth2/token")

Answer 2

好吧，我发现如果我将 jar 打包并通过 spark-submit 提交，它可以正常工作，因此可以同时工作。我仍然很惊讶它在本地 [*] 模式下不起作用。