pyspark 将数据框拆分为两列，而不为第二列创建文件夹结构

Question

两部分问题。

我有一个 pyspark 数据框，我正在从我的 azure blob 存储中的 JSON 文件列表中读取它。

在一些简单的 ETL 之后，我需要将它作为 parquet 文件从 blob 存储移动到数据湖，到目前为止很简单。

我没有成功地尝试有效地将其写入由两列定义的文件夹中，一列是日期列，另一列是 ID

使用partitionBy 让我接近

id | date                | nested_json_data | path
1  | 2019-01-01 12:01:01 | {data : [data]}  | dbfs:\mnt\..
1  | 2019-01-01 12:01:02 | {data : [data]}  | dbfs:\mnt\..


df.write.partitionBy("id","date").('dbfs:\mnt\curated\...') 

这给了我一个文件夹结构如下：

mnt --- |
         -- ...\1--|
                    --...\1\date=2019-01-01%2012:01:01\{file}.pq
                    --...\1\date=2019-01-01%2012:01:02\{file}.pq

我所追求的是每个唯一 id 列的单个文件夹，每个文件按日期拆分。

mnt --- |
         -- ...\1--|
                    --...\1 -- |
                               |filename_2019_01_01_12_0_01.pq
                               |filename_2019_01_01_12_0_02.pq

第二个问题是我的日期文件夹名称就像Date=2019-12-23 13%3A26%3A00

有没有一种方法可以在不更改我的 spark 数据帧的架构的情况下不使用它？如果我必须创建一个临时列，那很好。

Answer 1

您不能仅将 partitionBy 与 Spark 一起使用。据我所知，Spark 总是将分区写为partition=value 文件夹。见spark partition discovery。

您可以做的是在一个临时文件夹中写入partitionBy("id", "date")，然后递归地列出文件并移动/重命名它们以获得您想要的结构。

对于日期格式，可以在写之前进行转换：

df = df.withColumn("date") date_format(col("date"), "yyyy_MM_dd_HH_mm_ss")

这里有一些代码递归地列出临时文件夹中的所有零件文件，并将它们复制到目标文件夹并重命名。它使用 Hadoop FS，您可以根据自己的情况进行调整：

conf = sc._jsc.hadoopConfiguration()
Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileUtil = sc._gateway.jvm.org.apache.hadoop.fs.FileUtil


# list all files from staging folder
staging_folder_path = Path("/tmp/folder/")
fs_staging_f = staging_folder_path.getFileSystem(conf)
staging_files = fs_staging_f.listFiles(staging_folder_path, True)

# filter files that have names starting with 'part"
part_files = []
while staging_files.hasNext():
    part_file_path = staging_files.next().getPath()
    if part_file_path.getName().startswith("part"):
        part_files.append(part_file_path)

# add some logic here to flatten the date folders and add the date pattern to filenames

# move files
FileUtil.copy(....)