我在 s3 上确实有 n 个 .zip 文件,我想处理这些文件并从中提取一些数据。 .zip 文件包含单个 json 文件。在 Spark 中,我们可以读取 .gz 文件,但我没有找到任何方法来读取 .zip 文件中的数据。有人可以帮我解决如何使用 python 通过 spark 处理大型 zip 文件。我遇到了一些像 newAPIHadoopFile 这样的选项,但没有得到任何运气,也没有找到在 pyspark 中实现它们的方法。请注意 zip 文件 >1G,有些也是 20G。
以下是我使用的代码:
import zipfile
import io
file_name = "s3 file path for zip file"
def zip_extract(x):
in_memory_data = io.BytesIO(x[1])
file_obj = zipfile.ZipFile(in_memory_data, "r")
files = [i for i in file_obj.namelist()]
return dict(zip(files, [file_obj.open(file).read() for file in files]))
zips = sc.binaryFiles(file_name)
files_data = zips.map(zip_extract)
但由于以下原因而失败。我使用的实例是 r42x.large。
Exit code: 52
Stack trace: ExitCodeException exitCode=52:
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0
最佳答案
我确实以块的形式读取了 zip 文件的内容,并使用 spark 处理了这些块。这对我有用,并帮助我阅读大小超过 10G 的 zip 文件。下面是示例集:
max_data_length=10000
z = zipfile.ZipFile(zip_file)
data = []
counter=1
with z.open(z.infolist()[0]) as f:
line_counter=0
for line in f:
# Append file contents to list
data.append(line)
line_counter=line_counter+1
# Reset counters if record count hit max-data-length threshold
# Create spark dataframes
if not line_counter % max_data_length:
# Spark processing like:
df_rdd = spark.sparkContext.parallelize(data)
# Reset Counters and data-list
counter=counter+1
line_counter=0
data= []
关于python - 如何在pyspark中读取大型zip文件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55400119/