Scala - 在访问特定列的每一行时需要帮助，例如 x(i+1) 和 x(i-1)

Question

我有一个 spark 数据框，其中有几个列，如 tin、year、date_begin、date_end、continuous_data

    tin   year    continuous_data
    a1    2017          0
    a1    2017          1
    a1    2017          0
    a1    2017          1
    a1    2017          1
    a1    2017          0
    a1    2017          1
    a1    2017          1
    a1    2017          1
    a1    2017          0
    a1    2017          1

同样，我还有 2 个日期时间格式为 (yyyy-mm-dd HH:mm:ss) 的列。

我需要访问 'continuous_data' 列的每一行，例如 x(i+1) 和 x(i-1)。就我而言，就像

continuous_data(i) - 当前行值
Continuous_data(i-1) - 上一行值
Continuous_data(i+1) - 下一行值

这样我的需求如下所示

    tin   year    continuous_data    prev_data    next_data
    a1    2017          0                null        1
    a1    2017          1                0           0
    a1    2017          0                1           1
    a1    2017          1                0           1    
    a1    2017          1                1           0
    a1    2017          0                1           1
    a1    2017          1                0           1
    a1    2017          1                1           1
    a1    2017          1                1           0
    a1    2017          0                1           1
    a1    2017          1                0           null

我需要在纯 Scala 中解决它，而不是使用 spark 函数，我使用窗口函数来实现它，由于某些原因不需要。
我试图从过去几天解决这个问题，但我还不能解决它。有人可以帮我解决这个问题。

Answer 1

如果您需要在不使用 spark Window 和 Spark sql 函数的情况下处理基于 Spark 窗口的操作，您可以使用 UDAF 来完成。使用 UDAF 和 UDF 在一些博客中鼓励除非必要，否则不要使用它们。但是，如果您可以承受损失一些性能和更大的 GC 暂停，那么您可以尝试使用自定义 Spark 转换/聚合。

例子：

假设你想在你的数据集中执行一些幻灯片窗口，可以表示为：

item: String, key: String, timestamp: Long, field1:String, field2:Int, field3:Int, field4:Int

例如，您希望将 field2 的增量实现为数据帧的新字段，并且您想在不使用 Spark sql 的情况下执行此操作，并且需要使用 Scala 类型系统，例如，您想使用 Monoid 实例在两行之间执行操作。在这种情况下，也许直接使用RDD会更好......下面可能是如何处理Dataframe api的示例。

同时处理数据帧和 Scala 类型有点麻烦，因为您必须同时处理这两种类型：

您必须实现 UDAF 抽象成员：

class GenericAggregate(id: StringType, in: IntegerType, sort: LongType, output:IntegerType)(f: Seq[(String, Int)] => Seq[(String, Int)]) {

  private val mapType = MapType(id, MapType(sort, output, true), true)

  // This is the schema for your UDAF. The aggregation needs three fields from the input dataframe
  override def inputSchema: StructType =
    StructType(
      StructField("id_schema0", id) :: StructField("sort_schema0", sort) :: StructField(
        "input_schema0",
        in) :: Nil)

   // This is the internal fields you keep for computing your aggregate. 
  override def bufferSchema: StructType =
    StructType(StructField("internal_buffer", mapType) :: Nil)
  }

  // This kind of aggregation returns a key-value: key -> delta
  override def dataType: DataType = MapType(id, output)

  override def deterministic: Boolean = true

  // This is the initial value for your buffer schema.
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
   buffer(0) = Map("" -> 0)
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getAs[Map[String, Map[Long, Int]]](0) + (input.getAs[String](0) -> Map(
    input.getAs[Long](1) -> input.getAs[Int](2)))
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getAs[Map[String, Map[Long, Int]]](0) ++ buffer2
     .getAs[Map[String, Map[Long, Int]]](0)
  }


  override def evaluate(buffer: Row): Any = {
    val map = buffer.getAs[Map[String, Map[Long, Int]]](0)

    // You need to create a Seq from the map         

    val toSeq0 = map.mapValues(_.head)
    // As window functions you must order your events before applying the function
    val toSeq1 = toSeq0.toSeq.sortBy(_._2._1)
    val sequence = toSeq1.map(el => (el._1, el._2._2))

    /*
    For example, if your internal map is val in = 
        Map("ke1" -> Map(1L -> 3), "key2" -> Map(2L -> 3))
        you will get ArrayBuffer((ke1,3), (key2,3))
    */

    val result = f(sequence)  

    /*
       As a result you will have another ArrayBuffer with your new data
       Map("ke1" -> Map(1L -> 1), "key2" -> Map(2L -> 1))
    */
    result.toMap - k.initKey
  }

}

在本例中，使用 Map 来创建聚合，但您需要提供对该集合进行操作并返回新字段的函数。

为什么有必要？好吧，如果您需要在 Spark Sql 上抽象或构建具有更复杂类型的 DSL，您可以使用类型类派生来为您的产品构建自定义窗口/聚合函数。但是，正如我之前所说，强烈建议大多数时候直接使用 Spark SQL 函数。这些替代方案有助于更好地理解 Spark 的工作原理，并提供机会使用编译器来构建更通用的数据管道，甚至创建可以在 Spark Sql 上执行的 DSL。

希望这会有所帮助。