Использование значений столбцов в качестве имен файлов для сохранения фрейма данных искры

Question

Как сохранить фрейм данных spark в файл, используя значения столбцов в качестве имен файлов. возможно ли это?

+--------------------------+----------+-----------------+-----------------------------------+
|ID                        |CITY      |DATE             |name                               |
+--------------------------+----------+-----------------+-----------------------------------+
|1                         |          |2011-01-01       |20110101_DATA.snappy.parquet       |
|2                         |          |2011-01-01       |20110101_DATA.snappy.parquet       |
|3                         |          |2011-01-01       |20110101_DATA.snappy.parquet       |
|4                         |Chicago   |2011-01-01       |20110101_DATA.snappy.parquet       |
|5                         |Mansfield |2011-01-02       |20110102_DATA.snappy.parquet       |
|6                         |Pittsburgh|2011-01-02       |20110102_DATA.snappy.parquet       |
|7                         |          |2011-01-02       |20110102_DATA.snappy.parquet       |
|8                         |Clarion   |2011-01-03       |20110103_DATA.snappy.parquet       |
|9                         |Storrs    |2011-01-03       |20110103_DATA.snappy.parquet       |
|10                        |          |2011-01-03       |20110103_DATA.snappy.parquet       |
+--------------------------+----------+-----------------+-----------------------------------+

Ожидаемый результат:

Разделение по дате и использование значения имени в качестве имени файла при сохранении данных в виде паркета. О / р будет 3 файла

/DATE=2011-01-01/20110101_DATA.snappy.parquet
/DATE=2011-01-02/20110102_DATA.snappy.parquet
/DATE=2011-01-03/20110103_DATA.snappy.parquet

Answer 1

Spark не может создавать собственные имена в выходном файле партера, как вы хотите. Вы можете использовать следующий код, но это не масштабируемое решение, потому что вы используете действие .collect().

# In large dataframe maybe it will not work
unique_filename = [row.name for row in df.select('name').distinct().collect()]

for filename in  unique_filenames:
  output_filename = "/DATE=" + filename[0:4] + "-" + filename[4:6] + "-" + filename[6:8] + "/" + filename
  df.select("ID", "CITY", "DATE") 
    .filter(df['name']==filename) \
    .write \
    .parquet(output_filename)

У вас будет именно то, что вы хотите:

/DATE=2011-01-01/20110101_DATA.snappy.parquet
/DATE=2011-01-02/20110102_DATA.snappy.parquet
/DATE=2011-01-03/20110103_DATA.snappy.parquet