如何根据列中的一系列值拆分数据框并将它们存储在单独的文件中？答案

【问题标题】：How to split dataframe based on a range of values in a column and store them in separate files?如何根据列中的一系列值拆分数据框并将它们存储在单独的文件中？
【发布时间】：2020-01-16 23:10:57
【问题描述】：

从 postgres 读取表格后，我在 spark 中创建了一个数据框，如下所示。

val url    = "jdbc:postgresql://localhost:5432/testdb"
val connectionProperties = new Properties()
connectionProperties.setProperty("Driver", "org.postgresql.Driver")
connectionProperties.setProperty("Username", "testDB")
connectionProperties.setProperty("Password", "123456")
val query  = "select * from testdb.datatable"
val dataDF = spark.read.jdbc(url, query1, connectionProperties)

我可以看到数据框中的数据计数：

scala> dataDF.count
count: 3907891

样本输出：

    scala> dataDF.take(5)
------------|----|--------|
|source_name|id  |location|
|-----------|----|--------|
|   DB2     | 10 |Hive    |
|   SAP     | 20 |Hive    |
| SQL Server| 17 |Hive    |
|   Oracle  | 21 |Hive    |
|   DB2     | 33 |Hive    |
|-----------|----|--------|

数据框包含“整数”类型的“ID”列，其中包含 10 到 50 范围内的数据无论如何我可以将数据帧分成4个不同的分区，并根据每个文件的列ID将每个分区写入一个文件，其中每个文件包含file1: 10-20, file2: 21-30, file3: 31-40, file4: 41-50中的ID数据

【问题讨论】：

能否请您添加数据框示例输出。
在问题中添加了它

标签： scala apache-spark apache-spark-sql

【解决方案1】：

如果你知道 ids 范围，我会选择简单的。

val data = Seq(
      ("db2", 10, "Hive"),
      ("sap", 20, "Hive"),
      ("sql", 17, "Hive"),
      ("oracle", 21, "Hive"),
      ("server", 33, "Hive"),
      ("risk", 43, "Hive"),
    ).toDF("source_name", "id", "location")

val bucketed = data.withColumn("bucket",
         when($"id".between(0, 10), "1-10")
         .when($"id".between(11, 20), "11-20")
         .when($"id".between(21, 30), "21-30")
         .when($"id".between(31, 40), "31-40")
         .when($"id".between(41, 50), "41-50")
         .otherwise("50+"))

bucketed.write.option("header", true)
      .mode(SaveMode.Overwrite)
      .partitionBy("bucket")
      .csv("bucketing")

【讨论】：

我看到 4 个桶按 id 分区。 bucket=1-10 有 1 文件，ids 1 到 10，bucket=11-20 有 1 文件，ids 11 到 20。我有点困惑。
我的错，我错过了“何时”案例中的一些数据。但这很棒，对我有用。