scala - 使用 spark 按排序顺序将数据合并到 csv 文件

Question

我有一个这样的数据集：

name  time val
---- ----- ---
fred 04:00 111
greg 03:00 123
fred 01:00 411
fred 05:00 921
fred 11:00 157
greg 12:00 333

以及某个文件夹中的 csv 文件，一个对应于数据集中的每个唯一名称：

fred.csv
greg.csv

例如，fred.csv 的内容如下所示：

00:00 222
10:00 133

我的目标是以排序的时间顺序有效地将数据集合并到 CSV，以便 fred.csv 以如下方式结束：

00:00 222
01:00 411
04:00 111
05:00 921
10:00 133

实际上，有成千上万个独特的名字，而不仅仅是两个。我使用 union 和 sort 函数按顺序添加行，但我没有成功使用 partitionBy、for each 或将行合并到正确的 CSV 文件中。

Answer 1

导入并声明必要的变量

val spark = SparkSession.builder
  .master("local")
  .appName("Partition Sort Demo")
  .getOrCreate;

import spark.implicits._

从源文件创建数据框

val df = spark.read
  .format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .csv("csv/file/location")

//df.show()
+----+-----+---+
|name| time|val|
+----+-----+---+
|fred|04:00|111|
|greg|03:00|123|
|fred|01:00|411|
|fred|05:00|921|
|fred|11:00|157|
|greg|12:00|333|
+----+-----+---+

现在repartition按名称的数据帧和sort每个分区然后save它们

//repartition
val repartitionedDf = df.repartition($"name")

for {
  //fetch the distinct names in dataframe use as filename
  distinctName <- df.dropDuplicates("name").collect.map(_ (0))
} yield {
  import org.apache.spark.sql.functions.lit

  repartitionedDf.select("time", "val")
    .filter($"name" === lit(distinctName)) //filter df by name
    .coalesce(1)
    .sortWithinPartitions($"time") //sort
    .write.mode("overwrite").csv("location/" + distinctName + ".csv") //save
}

笔记：

CSV 文件的内容在突出显示的文件中可用。