S3 存储桶 .txt.gz 通过 PySpark 复制

Posted 2023-04-15

【中文标题】S3 存储桶 .txt.gz 通过 PySpark 复制

【英文标题】：S3 Bucket .txt.gz Copy Via PySpark

【发布时间】：2017-09-25 12:04:21

【问题描述】：

我正在使用 Python 2（在 EMR 上运行 PySpark 的 Jupyter notebook）。我正在尝试将一些数据作为数据帧加载，以便映射/减少它并将其输出到我自己的 S3 存储桶。

我通常使用这个命令：

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('file:///home/path/datafolder/data2014/*.csv')

当文件位于 S3 而不是我自己的存储桶时（因为我不确定如何格式化 .load 命令），这是我现在的大多数用例。我的文件也是 .csv 和 .txt.gz 的混合文件，我希望在复制时都采用 csv 格式（解压缩）。

我在 google 上查看并在 Python 2 (Jupyter notebook) 中尝试了以下命令：

import os
import findspark
findspark.init('/usr/lib/spark/')

from pyspark import SparkContext, SQLContext
sc = SparkContext.getOrCreate()
sqlContext = SQLContext(sc)

import sys

if sys.version_info[0] >= 3:
    from urllib.request import urlretrieve
else:
    from urllib import urlretrieve

# Get file from URL like this:
urlretrieve("https://s3.amazonaws.com/bucketname/path/path2/path3/path4/path3/results.txt.gz")

这只是输出：('/tmp/tmpmDB1EC.gz', <httplib.HTTPMessage instance at 0x7f54db894758>) 所以我不确定现在该做什么。

我已经阅读了文档，并在这个网站和谷歌上搜索了形成 df 的简单方法，但我被卡住了。我还阅读了有关使用我的 AWS 密钥/秘密密钥（我有）的信息，但我找不到可以遵循的示例。

有人可以帮帮我吗？

【参考方案1】：

您需要使用 spark 上下文加载它

data_file = urlretrieve("https://s3.amazonaws.com/bucketname/path/path2/path3/path4/path3/results.txt.gz")

raw_data = sc.textFile(data_file)

【讨论】：

感谢您的快速回复。我试过这个，但我得到了这个错误：Py4JError：调用 o29.textFile 时发生错误。此外，我不确定这是否可扩展，即我可以像这样读取多个路径：***.com/questions/46240271/…?

sc.textFile("s3.amazonaws.com/bucketname/path/path2/path3/path4/path3/results.txt.gz") 也许？

然后我得到这个错误：NameError: name 'urlretrieve' is not defined。 urllib 在 Python 2 上不可用。问题是我需要能够复制多个 S3 路径（理想情况下转换为解压缩的 CSV）并在单个数据帧中合并在一起，以便进行映射/减少并输出到我的 S3 存储桶.我似乎找不到解决方案。

为什么你会从运行 sc.textFile("s3.amazonaws.com/bucketname/path/path2/path3/pa‌​th4/path3/results.tx‌​t.gz") 得到它不使用 urlretrieve？

我意识到了这个问题。安装 EMR 时，我打开的是 Python 2 笔记本，而不是 Spark 笔记本。这导致了这个问题。它基本上是试图通过 Python 而不是 PySpark 运行命令。