重命名由于 partitionBy 而创建的文件夹

Question

我正在为作业在胶水上运行的时间戳添加一列。我想使用partitionBy(load_timestamp) 保存它。创建了一个文件夹，例如load_timestamp=2020-04-27 03:21:54. 我希望文件夹命名为table_name=2020-04-27 03:21:54. 这可能吗？

enriched = df.withColumn("load_timestamp", unix_timestamp(lit(timestamp),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))
enriched.write.partitionBy("load_timestamp").format("parquet").mode("append").save("s3://s3-enriched-bucket/" + job_statement[0])

Answer 1

默认情况下，Spark 根据分区列创建directories，即

<partition_column_name>=<value>

Easiest way 解决方法是将列名保留为table_name 并在partition by 子句中使用。

enriched = df.withColumn("table_name", unix_timestamp(lit(timestamp),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))

enriched.write.partitionBy("table_name").format("parquet").mode("append").save("s3://s3-enriched-bucket/" + job_statement[0])

Other way would be:

通过使用hadoop.fs file API 迭代重命名目录并将load_timestamp 更改为table_name。