如何解决将Pyspark数据框保存到Azure存储
我正在将概念证明从AWS / EMR迁移到Azure。
它是用python编写的,并在AWS EMR和S3上使用Spark,Hadoop和Cassandra。它可以计算少量OTC衍生产品的潜在远期敞口。
我目前有一个障碍:如何将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存储帐户和容器。我已启用对此的公开访问。
我探索了各种方法,例如:
- 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 Ubuntu 18.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.access$200(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存储/数据帧问题的已解决帖子。
解决方法
根据我的测试,我们可以使用软件包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
,
我通过将存储帐户更改为Blobstorage类型而不是Storage Gen2解决了该问题。 Windows Azure存储Blob(WASB)驱动程序is unsupported with Data Lake Storage Gen2.
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。