使用列值作为文件名来保存 spark 数据框

Posted 2023-04-15

【中文标题】使用列值作为文件名来保存 spark 数据框

【英文标题】：Using column values as filenames to save spark dataframe

【发布时间】：2020-01-22 18:24:20

【问题描述】：

如何使用列值作为文件名将 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       |
+--------------------------+----------+-----------------+-----------------------------------+

预期输出：

按日期分区并在将数据保存为 parquet 时使用名称值作为文件名。 o/p 将是 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

【问题讨论】：

你能举个例子吗？

@ggeop 已更新示例

还有一个定义，你只需要空文件或者你想要里面有数据？如果是，哪些数据。数据框会很大吗？

是的，文件将包含数据。它将包含除名称列和 DATE 之外的所有列（因为 DATE 将存在于分区中）

【参考方案1】：

Spark 无法根据需要在输出 parquet 文件中本地创建自定义名称。您可以使用以下代码，但它不是一个可扩展的解决方案，因为您使用了.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