将 Pyspark Dataframe 保存到 Azure 存储

Question

我正在将概念验证从 AWS/EMR 迁移到 Azure。

它是用 python 编写的，在 AWS EMR 和 S3 上使用 Spark、Hadoop 和 Cassandra。它计算一小组场外交易衍生品的潜在远期敞口。

我目前有一个障碍：如何将 pyspark 数据帧保存到 Azure 存储？

在 AWS / S3 中这很简单，但我还没有让它在 Azure 上运行。我可能在做傻事！

我已经测试了将文件写入 blob 和 Azure 上的文件存储，但尚未找到指向数据帧的指针。

在 AWS 上，我目前使用以下内容：

npv_dataframe.coalesce(1).saveAsTextFile(output_dir + '/exposure_scenarios/' + str(counterparty))

其中 output_dir 的格式为 s3://s3_bucket_name/directory_name

我设置了一个 Data Lake Storage Gen2 存储帐户和容器。我已启用 public 访问权限。

我探索过各种方法例如：

https://docs.microsoft.com/en-us/python/api/overview/azure/storage-blob-readme?view=azure-python
https://docs.microsoft.com/en-us/azure/storage/common/storage-samples-python?toc=/azure/storage/blobs/toc.json
https://docs.databricks.com/_static/notebooks/data-import/azure-blob-store.html
Write data from pyspark to azure blob?（我相信这是旧的，hadoop 3.2.1 带有 abfs 支持）

其中一些示例使用文件上传模式，但我想要的是直接从 pyspark 数据帧保存。

我使用的测试代码是：

import traceback
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType

try:

    spark = SparkSession.builder.getOrCreate()
    conf = spark.sparkContext._jsc.hadoopConfiguration()
    conf.set("fs.wasbs.impl", "org.apache.hadoop.fs.azure.NativeAzureFileSystem")
    spark.conf.set('fs.azure.account.key.#myaccount#.blob.core.windows.net', '#mykey#')
    df = spark.createDataFrame(["10", "11", "13"], StringType()).toDF("age")
    df.show()
    df \
        .coalesce(1) \
        .write.format('csv') \
        .option('header', True) \
        .mode('overwrite') \
        .save('wasbs://#mycontainer#@#myaccount#.blob.core.windows.net/result_csv')
    print("Hadoop version: " + spark.sparkContext._gateway.jvm.org.apache.hadoop.util.VersionInfo.getVersion())
except Exception as exp:
    print("Exception occurred")
    print(traceback.format_exc())

上面的示例在 df.write 处失败 - 错误是

py4j.protocol.Py4JJavaError: An error occurred while calling o48.save.
: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.azure.NativeAzureFileSystem not found

我在使用 spark-submit 时收到同样的错误

spark-submit --packages org.apache.hadoop:hadoop-azure:3.2.1,com.microsoft.azure:azure-storage:8.6.3 ./test.py

我认为这可能是版本兼容性问题。我注意到 pyspark 中的 hadoop.jars 都是 2.7.4 版本，而我引用的是 3.2.1 安装。

我正在/正在使用：

Java 8 (1.8.0_265) 火花 3.0.0 Hadoop 3.2.1 Python 3.6 Ubuntu18.04

我确保 Spark 目录中的所有 hadoop jar 与 Hadoop jar 目录中的相同。

在跟踪另一个堆栈跟踪错误后，我将命令更新为：spark-submit --packages org.apache.hadoop:hadoop-azure:3.2.1,com.microsoft.azure:azure-storage:8.6.5 test.py

然后我收到了一个不同的 Java 错误，这看起来像是密钥的问题？？

py4j.protocol.Py4JJavaError: An error occurred while calling o48.save.
: java.lang.NoSuchMethodError: 'org.apache.hadoop.conf.Configuration org.apache.hadoop.security.ProviderUtils.excludeIncompatibleCredentialProviders(org.apache.hadoop.conf.Configuration, java.lang.Class)'
        at org.apache.hadoop.fs.azure.SimpleKeyProvider.getStorageAccountKey(SimpleKeyProvider.java:45)
        at org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.getAccountKeyFromConfiguration(AzureNativeFileSystemStore.java:989)
        at org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.createAzureStorageSession(AzureNativeFileSystemStore.java:1078)
        at org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.initialize(AzureNativeFileSystemStore.java:543)
        at org.apache.hadoop.fs.azure.NativeAzureFileSystem.initialize(NativeAzureFileSystem.java:1344)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2669)
        at org.apache.hadoop.fs.FileSystem.access0(FileSystem.java:94)
        at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
        at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2685)
        at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:373)
        at org.apache.hadoop.fs.Path.getFileSystem(Path.java:295)
        at org.apache.spark.sql.execution.datasources.DataSource.planForWritingFileFormat(DataSource.scala:424)
        at org.apache.spark.sql.execution.datasources.DataSource.planForWriting(DataSource.scala:524)
        at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:290)
        at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:271)
        at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:229)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:564)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:238)
        at java.base/java.lang.Thread.run(Thread.java:832)

此外，在将 Azure 帐户安全密钥添加到 hadoop 配置后，如果我尝试：

hdfs dfs -ls wasbs://CONTAINER@ACCOUNT.blob.core.windows.net/

我收到错误：java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.azure.NativeAzureFileSystem$Secure not found

感谢任何帮助！有点卡住了想法。看起来，相对于 AWS，关于 Azure 存储/Dataframe 问题的解决帖子很少。

Answer 1

根据我的测试，我们可以使用包com.microsoft.azure:azure-storage:8.6.3将文件上传到spark中的Azure blob。

例如

我正在使用

Java 8 (1.8.0_265) Spark 3.0.0 Hadoop 3.2.0 Python 3.6.9 Ubuntu 18.04

我的代码

import traceback
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType

try:
     
    spark = SparkSession.builder.getOrCreate()
    conf = spark.sparkContext._jsc.hadoopConfiguration()
    conf.set("fs.wasbs.impl", "org.apache.hadoop.fs.azure.NativeAzureFileSystem")
    spark.conf.set('fs.azure.account.key.jimtestdiag924.blob.core.windows.net', '')
    df = spark.createDataFrame(["10", "11", "13"], StringType()).toDF("age")
    df.show()
    df \
      .coalesce(1) \
      .write.format('csv') \
      .option('header', True) \
      .mode('overwrite') \
      .save('wasbs://testupload@<account name>.blob.core.windows.net/result_csv')  
    print("Hadoop version: " + spark.sparkContext._gateway.jvm.org.apache.hadoop.util.VersionInfo.getVersion())      
except Exception as exp:
    print("Exception occurred")
    print(traceback.format_exc())

我的命令

spark-submit --packages org.apache.hadoop:hadoop-azure:3.2.0,com.microsoft.azure:azure-storage:8.6.3  spark.py

Answer 2

我通过将存储帐户更改为 Blobstorage 类型而不是 Storage Gen2 解决了这个问题。 Windows Azure 存储 Blob (WASB) 驱动程序 is unsupported with Data Lake Storage Gen2.

将 Pyspark Dataframe 保存到 Azure 存储

Saving Pyspark Dataframe to Azure Storage

azure

azure-blob-storage

pyspark-dataframes