Как экспортировать таблицу данных в PySpark в CSV?

Я использую Spark 1.3.1 (PySpark) и сгенерировал таблицу с помощью SQL-запроса. Теперь у меня есть объект, который является DataFrame. Я хочу экспортировать этот объект DataFrame (я назвал его "таблица") в CSV файл, чтобы я мог управлять им и строить столбцы. Как экспортировать таблицу DataFrame в файл csv?

Спасибо!

Ответ 1

Если кадр данных помещается в память драйвера и вы хотите сохранить его в локальной файловой системе, вы можете преобразовать Spark DataFrame в локальный Pandas DataFrame с помощью метода toPandas, а затем просто используйте to_csv

df.toPandas().to_csv('mycsv.csv')

В противном случае вы можете использовать spark-csv:

Spark 1.3

df.save('mycsv.csv', 'com.databricks.spark.csv')

Искра 1.4 +

df.write.format('com.databricks.spark.csv').save('mycsv.csv')

В Spark 2.0+ вы можете напрямую использовать источник данных csv:

df.write.csv('mycsv.csv')

Ответ 2

Для Apache Spark 2+, чтобы сохранить данные в одном файле csv. Используйте следующую команду

query.repartition(1).write.csv("cc_out.csv", sep='|')

Здесь 1 указывает, что мне нужен только один раздел csv. вы можете изменить его в соответствии с вашими требованиями.

Ответ 3

Если вы не можете использовать spark-csv, вы можете сделать следующее:

df.rdd.map(lambda x: ",".join(map(str, x))).coalesce(1).saveAsTextFile("file.csv")

Если вам нужно обрабатывать строки с разрывами строк или запятыми, которые не будут работать. Используйте это:

import csv
import cStringIO

def row2csv(row):
    buffer = cStringIO.StringIO()
    writer = csv.writer(buffer)
    writer.writerow([str(s).encode("utf-8") for s in row])
    buffer.seek(0)
    return buffer.read().strip()

df.rdd.map(row2csv).coalesce(1).saveAsTextFile("file.csv")

Ответ 4

Как насчет этого (вам не нужен один лайнер)?

for row in df.collect():
    d = row.asDict()
    s = "%d\t%s\t%s\n" % (d["int_column"], d["string_column"], d["string_column"])
    f.write(s)

f - открытый дескриптор файла. Также разделителем является TAB char, но его легко изменить на все, что вы хотите.

Ответ 5

Вам нужно перераспределить Dataframe в одном разделе, а затем определить формат, путь и другие параметры для файла в формате файловой системы Unix, и вот, пожалуйста,

df.repartition(1).write.format('com.databricks.spark.csv').save("/path/to/file/myfile.csv",header = 'true')

Подробнее о функции перераспределения Подробнее о функции сохранения