Блог

Главная - Вопросы по программированию - Измените csv.writer(open()) — Python на PySpark

Измените csv.writer(open()) — Python на PySpark

#python #pandas #apache-spark #pyspark

Вопрос:

Я пытаюсь применить функцию, написанную на 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?

Ответ №1:

Я пока не могу опубликовать комментарий, но я хотел бы сказать, что для неструктурированных данных вы можете выбрать Spark RDD.

Когда вы создадите экземпляр сеанса SparkSession, вы сможете извлечь контекст Spark. Например, при условии, что вы правильно создали свой экземпляр:

 sc = spark.sparkContext

Затем вы можете преобразовать свой отсортированный список в RDD, как таковой:

 sorted_RDD = sc.parallelize(sorted_list)

Оттуда вы можете запустить метод saveAsSingleTextFile для записи в текстовый файл.

 sorted_RDD.saveAsSingleTextFile(<output_dir>)

Теперь я не уверен, что у вас есть какая-либо дальнейшая обработка. Другой вариант, который вы можете сделать, — определить максимальный размер списка, а затем создать фрейм данных со значениями null, чтобы создать четный размер. На самом деле все зависит от того, что вы хотите извлечь из этих данных.

**Пожалуйста, обратите внимание, что я использовал метод saveAsSingleTextFile (), потому что я предполагаю, что вы хотите сохранить это как один текстовый файл. Если вы хотите разделить этот файл, что обычно рекомендуется для больших наборов данных, вы можете сделать saveAsTextFile().

Метки: