将 csv.writer(open()) - Python 更改为 PySpark答案

【问题标题】：Change csv.writer(open()) - Python to PySpark将 csv.writer(open()) - Python 更改为 PySpark
【发布时间】：2021-12-03 23:51:39
【问题描述】：

我正在尝试应用一个用 Python 制作的函数，以便在 PySpark 上可读。

open() 函数在 PySpark 中不可读。在这种情况下如何更换？

我有一个需要排序的非结构化列表：

import csv
from operator import itemgetter
my_list = [
        [3, 'ab','ac','ad'],
        [4, 'ae','af','at','aj','aa'],
        [1, 'ar','aa','at','as'],
        [2, 'ay','au','aa','ar','aa','a1']
        ]
# Sorting this list:
sorted_list =  sorted(my_list, key=itemgetter(0))

这是我需要更改的在 Python 中工作的函数：


with open('file.txt', 'w', newline='') as myfile:
      wr = csv.writer(myfile, quoting=csv.QUOTE_NONE, delimiter='|')
      wr.writerows(sorted_list)

由于列表列表的非结构化类型，我无法在下面使用此功能：

df = pd.DataFrame(sorted_list)
spark_df = spark.createDataFrame(df)
spark_df.write.csv("file.txt".format(root),sep="|", mode='overwrite', quote="\u0000")

在 Python 或 PySpark 中有解决方法吗？

【问题讨论】：

标签： python pandas apache-spark pyspark

【解决方案1】：

我还不能发表评论，但我想说，对于非结构化数据，您可以选择 Spark RDD。

当您实例化您的 SparkSession 时，您可以检索 Spark 上下文。例如，假设您正确实例化了您的实例：

sc = spark.sparkContext

然后您可以将排序列表转换为 RDD，如下所示：

sorted_RDD = sc.parallelize(sorted_list)

您可以从那里运行 saveAsSingleTextFile 方法来写入文本文件。

sorted_RDD.saveAsSingleTextFile(<output_dir>)

现在，我不确定您是否有任何进一步的处理。您可以做的另一个选择是找出列表的最大大小，然后创建一个具有空值的数据框以创建一个均匀的大小。这完全取决于您想从这些数据中提取什么。

**请注意，我使用了 saveAsSingleTextFile() 方法，因为我假设您希望将其存储为单个文本文件。如果你想分割这个文件，通常推荐用于较大的数据集，你可以做 saveAsTextFile()。

【讨论】：