spark在S3上的分区内创建分区

Question

我有以下制表符分隔的示例数据集：

col1  period  col3  col4  col5  col6  col7  col8  col9  col10 col11 col12 col13 col14 col15 col16 col17 col18 col19 col20 col21 col22
ASSDF 202001  A B BFGF  SSDAA WDSF  SDSDSD  SDSDSSS SDSDSD  E F FS  E CURR1 CURR2 -99 CURR3 -99 -99 -99 -99
ASSDF 202002  A B BFGF  SSDAA WDSF  SDSDSD  SDSDSSS SDSDSD  E F FS  E CURR1 CURR2 -99 CURR3 -99 -99 -99 -99
ASSDF 202003  A B BFGF  SSDAA WDSF  SDSDSD  SDSDSSS SDSDSD  E F FS  E CURR1 CURR2 -99 CURR3 -99 -99 -99 -99
ASSDF 202004  A B BFGF  SSDAA WDSF  SDSDSD  SDSDSSS SDSDSD  E F FS  E CURR1 CURR2 -99 CURR3 -99 -99 -99 -99
...
...
ASSDF 202312  A B BFGF  SSDAA WDSF  SDSDSD  SDSDSSS SDSDSD  E F FS  E CURR1 CURR2 -99 CURR3 -99 -99 -99 -99

我正在对此数据运行一些转换，最终数据位于spark dataset“ds1”中。之后，我用“period”分区将该数据集写入s3。因为我也希望在s3文件中使用period,所以我正在从from period列创建另一列“datasetperiod”。

我的scala函数来保存TSV数据集。

def saveTsvDataset(dataframe: DataFrame, outputFullPath: String, numPartitions: Integer, partitionCols: String*): Unit = {
    dataframe
      .repartition(numPartitions)
      .write
      .partitionBy(partitionCols:_*)
      .mode(SaveMode.Overwrite)
      .option("sep", "\t")
      .csv(outputFullPath)
  }

在S3上保存数据集的Scala代码。为S3上的分区添加新列datasetPeriod。

 saveTsvDataset(
      DS1.withColumn("datasetPeriod",$"period")
      , "s3://s3_path"
      , 100
      , "period"
    )

现在，我的问题是我的周期从202001到202312，当我在s3上使用“DataSetPeriod”上的分区编写时，有时它会在任何随机的周期内创建分区。所以这在任何时期都是随机发生的。我从来没有看到这种情况发生了多个时期。它创建类似“s3://s3_path/datasetperiod=202008/datasetperiod=202008”的路径。

Answer 1

DataFrame中已经有句点列。因此不需要再创建一个新的重复的DataSetPeriode列。

当您使用.PartitionBy（“period”）将DataFrame写入S3:/../ParentFolder时，它将创建如下所示的文件夹：

df.write.partitionBy("period").csv("s3://../parentFolder/")
s3://.../parentFolder/period=202001/
s3://.../parentFolder/period=202002/
s3://.../parentFolder/period=202003/
...
s3://.../parentFolder/period=202312/

在回读数据时，只需提到路径tillparentfolderonly，它将自动读取periode作为列之一。

val df = spark.read.csv("s3://../parentFolder/")
//df.schema will give you `period` as one of the column
df.printSchema
root
 |-- col1: string (nullable = true)
 |-- .... //other columns go here
 |-- period: string (nullable = true)

也就是说，无论您在partition列中获得多个分区，都只是由于使用PartitionBy写入数据时使用了错误的路径。