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

Question

我正在尝试应用一个用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中有变通方法吗？

Answer 1

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

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

sc = spark.sparkContext

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

sorted_RDD = sc.parallelize(sorted_list)

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

sorted_RDD.saveAsSingleTextFile(<output_dir>)

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

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