以有状态的方式处理火花中的网络数据包

Question

我想使用Spark来解析网络消息，并以有状态的方式将它们分组为逻辑实体。

问题描述

假设每条消息都在输入数据帧的一行中，如下所示。

| row   | time | raw payload   |
+-------+------+---------------+
|  1    | 10   | TEXT1;        |
|  2    | 20   | TEXT2;TEXT3;  |
|  3    | 30   | LONG-         |
|  4    | 40   | TEXT1;        |
|  5    | 50   | TEXT4;TEXT5;L |
|  6    | 60   | ONG           |
|  7    | 70   | -TEX          |
|  8    | 80   | T2;           | 

任务是解析原始有效负载中的逻辑消息，并在新的输出数据帧中提供它们。在该示例中，有效载荷中的每个逻辑消息以分号（分隔符）结束。

然后，所需的输出数据框可能如下所示：

| row   | time | message       |
+-------+------+---------------+
|  1    | 10   | TEXT1;        |
|  2    | 20   | TEXT2;        |
|  3    | 20   | TEXT3;        |
|  4    | 30   | LONG-TEXT1;   |
|  5    | 50   | TEXT4;        |
|  6    | 50   | TEXT5;        |
|  7    | 50   | LONG-TEXT2;   |

请注意，某些消息行不会在结果中产生新行（例如，行4,6,7,8），有些甚至会产生多行（例如，行2,5）

我的问题：

• 这是UDAF的用例吗？如果是这样，我应该如何实现merge函数？我不知道它的目的是什么。
• 由于消息排序很重要（我无法在不尊重消息顺序的情况下正确处理LONGTEXT-1，LONGTEXT-2），我可以告诉spark可能在更高级别（例如，每个消息的日历日）并行化，但不能在一天内并行化（例如，在时间50,60,70,80处的事件需要按顺序处理。
• 跟进问题：可以想象解决方案不仅可以用于传统的火花，还可以用于火花结构流媒体吗？或者后者是否需要自己的状态处理方法？

Answer 1

通常，您可以使用mapGroupsWithState的flatMapGroupsWithState在spark流上运行任意有状态聚合。你可以找到一些例子here。这些都不会保证流的处理将按事件时间排序。

如果您需要强制执行数据排序，则应尝试使用window operations on event time。在这种情况下，您需要运行无状态操作，但如果每个窗口组中的元素数量足够小，您可以使用collectList作为实例，然后应用UDF（您可以在其中管理每个窗口组的状态）每个清单。

Answer 2

好吧，我在此期间想到了如何使用UDAF做到这一点。

class TagParser extends UserDefinedAggregateFunction {

  override def inputSchema: StructType = StructType(StructField("value", StringType) :: Nil)

  override def bufferSchema: StructType = StructType(
    StructField("parsed", ArrayType(StringType)) ::
      StructField("rest", StringType)
      :: Nil)

  override def dataType: DataType = ArrayType(StringType)

  override def deterministic: Boolean = true

  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = IndexedSeq[String]()
    buffer(1) = null
  }

  def doParse(str: String, buffer: MutableAggregationBuffer): Unit = {

    buffer(0) = IndexedSeq[String]()

    val prevRest = buffer(1)
    var idx = -1
    val strToParse = if (prevRest != null) prevRest + str else str

    do {
      val oldIdx = idx;
      idx = strToParse.indexOf(';', oldIdx + 1)

      if (idx == -1) {
        buffer(1) = strToParse.substring(oldIdx + 1)
      } else {
        val newlyParsed = strToParse.substring(oldIdx + 1, idx)
        buffer(0) = buffer(0).asInstanceOf[IndexedSeq[String]] :+ newlyParsed
        buffer(1) = null
      }

    } while (idx != -1)
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

    if (buffer == null) {
      return
    }

    doParse(input.getAs[String](0), buffer)
  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = throw new UnsupportedOperationException

  override def evaluate(buffer: Row): Any = buffer(0)
}

这里有一个演示应用程序使用上面的UDAF来解决上面的问题：

case class Packet(time: Int, payload: String)

object TagParserApp extends App {

  val spark, sc = ... // kept out for brevity

  val df = sc.parallelize(List(
    Packet(10, "TEXT1;"),
    Packet(20, "TEXT2;TEXT3;"),
    Packet(30, "LONG-"),
    Packet(40, "TEXT1;"),
    Packet(50, "TEXT4;TEXT5;L"),
    Packet(60, "ONG"),
    Packet(70, "-TEX"),
    Packet(80, "T2;")
  )).toDF()

  val tp = new TagParser
  val window = Window.rowsBetween(Window.unboundedPreceding, Window.currentRow)
  val df2 = df.withColumn("msg", tp.apply(df.col("payload")).over(window))
  df2.show()
}

这会产生：

+----+-------------+--------------+
|time|      payload|           msg|
+----+-------------+--------------+
|  10|       TEXT1;|       [TEXT1]|
|  20| TEXT2;TEXT3;|[TEXT2, TEXT3]|
|  30|        LONG-|            []|
|  40|       TEXT1;|  [LONG-TEXT1]|
|  50|TEXT4;TEXT5;L|[TEXT4, TEXT5]|
|  60|          ONG|            []|
|  70|         -TEX|            []|
|  80|          T2;|  [LONG-TEXT2]|
+----+-------------+--------------+

我的主要问题是弄清楚如何实际应用这个UDAF，即使用这个：

df.withColumn("msg", tp.apply(df.col("payload")).over(window))

我现在唯一需要弄清楚的是并行化的各个方面（我只希望在不依赖于排序的情况下发生）但对我来说这是一个单独的问题。