从数据集中的地图按键排序

Question

我想通过时间戳订购一些我从HDFS检索的avro文件。

我的avro文件的架构是：

headers：Map [String，String]，body：String

现在棘手的部分是时间戳是地图中的键/值之一。所以我在地图中包含了这样的时间戳：

key_1 - > value_1，key_2 - > value_2，timestamp - > 1234567，key_n - > value_n

请注意，值的类型是String。

我创建了一个case类来使用这个模式创建我的数据集：

case class Root(headers : Map[String,String], body: String)

创建我的数据集：

val ds = spark
          .read
          .format("com.databricks.spark.avro")
          .load(pathToHDFS)
          .as[Root]

我真的不知道如何开始这个问题，因为我只能得到列标题和正文。如何使嵌套值最终按时间戳排序？

我想做这样的事情：

ds.select("headers").doSomethingToGetTheMapStructure.doSomeConversionStringToTimeStampForTheColumnTimeStamp("timestamp").orderBy("timestamp")

有点精确：我不想从我的初始数据集中丢失任何数据，只是一个排序操作。

我使用Spark 2.3.0。

Answer 1

加载的Dataset看起来应该类似于下面的示例数据集：

case class Root(headers : Map[String, String], body: String)

val ds = Seq(
  Root(Map("k11"->"v11", "timestamp"->"1554231600", "k12"->"v12"), "body1"),
  Root(Map("k21"->"v21", "timestamp"->"1554134400", "k22"->"v22"), "body2")
).toDS

您可以通过Map键查找timestamp，cast值Long，并按如下方式执行orderBy：

ds.
  withColumn("ts", $"headers"("timestamp").cast("Long")).
  orderBy("ts").
  show(false)
// +-------------------------------------------------+-----+----------+
// |headers                                          |body |ts        |
// +-------------------------------------------------+-----+----------+
// |[k21 -> v21, timestamp -> 1554134400, k22 -> v22]|body2|1554134400|
// |[k11 -> v11, timestamp -> 1554231600, k12 -> v12]|body1|1554231600|
// +-------------------------------------------------+-----+----------+

请注意，$"headers"("timestamp")与使用apply柱方法（即$"headers".apply("timestamp")）相同。

或者，您也可以使用getItem按键访问Map，例如：

$"headers".getItem("timestamp")

Answer 2

您可以使用Scala的sortBy，它接受一个函数。我建议你明确地将val ds声明为Vector（或其他集合），这样你就会在IntelliJ中看到适用的函数（如果你使用的是IntelliJ），它肯定会编译。

根据您的代码查看下面的示例：

  case class Root(headers : Map[String,String], body: String)

  val ds: Vector[Root] = spark
    .read
    .format("com.databricks.spark.avro")
    .load(pathToHDFS)
    .as[Root]

  val sorted = ds.sortBy(r => r.headers.get("timestamp").map(PROCESSING) ).reverse

编辑：添加反向（假设您希望它下降）。在作为参数传递的函数内部，您还可以将处理放入时间戳。

Answer 3

import org.apache.spark.sql.{Encoders, Encoder, Dataset}
import org.apache.spark.sql.functions.{col, desc}
import java.sql.Timestamp

case class Nested(key_1: String,key_2: String,timestamp: Timestamp,key_n: String)
case class Root(headers:Nested,body:String)

implicit val rootCodec: Encoder[Root] = Encoders.product[Root]

val avroDS:Dataset[Root] = spark.read
                                .format("com.databricks.spark.avro")
                                .load(pathToHDFS)
                                .as[Root]

val sortedDF: DataFrame = avroDS.orderBy(desc(col("timestamp")))

此代码段会直接将您的Avro数据投射到Dataset[Root]。您不必依赖导入sparksession.implicits并且将消除将timestamp字段转换为TimestampType的步骤。在内部，Spark的Timestamp数据类型使用java.sql.Timestamp实现。