在 Spark Dataframe Rows 上并行操作

Question

环境：Scala、spark、结构化流、kafka

我有一个来自具有以下架构的 kafka 流的 DF

DF：

BATCH ID: 0
+-----------------------+-----+---------+------+
|                  value|topic|partition|offset|
+-----------------------+-----+---------+------+
|{"big and nested json"}|  A  |        0|     0|
|{"big and nested json"}|  B  |        0|     0|
+-----------------------+-----+---------+------+

我想使用 spark 并行处理每一行，并且我设法将它们拆分给我的执行者使用

DF.repartition(Number).foreach(row=> processRow(row))

我需要将值列中的值提取到它自己的数据框中来处理它。 我在使用 Dataframe 通用 Row 对象时遇到了困难..

有没有办法将每个执行程序中的单行转换为自己的数据帧（使用固定模式？）并在固定位置写入？ 有没有更好的方法来解决我的问题？

编辑+澄清：

使用自 spark2.4 以来存在的 writeStream 功能的 forEachBatch 函数作为批处理来接收 DF

目前将 DF 拆分为 ROWS 使得行将平均拆分为我的所有执行程序，我想将单个 GenericRow 对象转换为 DataFrame，以便我可以使用我制作的函数进行处理

例如，我会将行发送到函数

processRow(row:row)

取值和主题，转回单行DF

+-----------------------+-----+
|                  value|topic|
+-----------------------+-----+
|{"big and nested json"}|  A  |
+-----------------------+-----+

用于进一步处理

Answer 1

我猜你一次消耗多个 kafka 数据。

首先你需要为所有kafka主题准备schema，例如我在value列中使用了两个不同的JSON。

scala> val df = Seq(("""{"name":"Srinivas"}""","A"),("""{"age":20}""","B")).toDF("value","topic")

scala> df.show(false)
+-------------------+-----+
|value              |topic|
+-------------------+-----+
|{"name":"Srinivas"}|A    |
|{"age":20}         |B    |
+-------------------+-----+

scala> import org.apache.spark.sql.types._

主题 A 的架构

scala> val topicASchema = DataType.fromJson("""{"type":"struct","fields":[{"name":"name","type":"string","nullable":true,"metadata":{}}]}""").asInstanceOf[StructType]

主题 B 的架构

scala> val topicBSchema = DataType.fromJson("""{"type":"struct","fields":[{"name":"age","type":"long","nullable":true,"metadata":{}}]}""").asInstanceOf[StructType]

结合主题及其架构。

scala> val topicSchema = Seq(("A",topicASchema),("B",topicBSchema)) // Adding Topic & Its Schema.

处理数据帧

scala> topicSchema
.par
.map(d => df.filter($"topic" === d._1).withColumn("value",from_json($"value",d._2)))
.foreach(_.show(false)) // Using .par & filtering dataframe based on topic & then applying schema to value column.
+----------+-----+
|value     |topic|
+----------+-----+
|[Srinivas]|A    |
+----------+-----+

+-----+-----+
|value|topic|
+-----+-----+
|[20] |B    |
+-----+-----+

写入 hdfs

scala> topicSchema
.par
.map(d => df.filter($"topic" === d._1).withColumn("value",from_json($"value",d._2)).write.format("json").save(s"/tmp/kafka_data/${d._1}"))

存储在 hdfs 中的最终数据

scala> import sys.process._
import sys.process._

scala> "tree /tmp/kafka_data".!
/tmp/kafka_data
├── A
│   ├── part-00000-1e854106-49de-44b3-ab18-6c98a126c8ca-c000.json
│   └── _SUCCESS
└── B
    ├── part-00000-1bd51ad7-cfb6-4187-a374-4e2d4ce9cc50-c000.json
    └── _SUCCESS

2 directories, 4 files

Answer 2

在这种情况下，更适合使用.map 而不是.foreach。原因是map 返回一个新数据集，而foreach 只是一个函数，不返回任何内容。

可以帮助您的另一件事是解析位于 JSON 中的架构。

我最近有类似的要求。 我的 JSON 对象对于主题 A 和 B 都有一个“相似”模式。如果您不是这种情况，您可能需要在下面的解决方案中通过按主题对它们进行分组来创建多个dataframes。

val sanitiseJson: String => String = value => value
  .replace("\\\"", "\"")
  .replace("\\\\", "\\")
  .replace("\n", "")
  .replace("\"{", "{")
  .replace("}\"", "}")

val parsed = df.toJSON
  .map(sanitiseJson)

这会给你类似的东西：

{
    "value": { ... },
    "topic": "A"
}

然后你可以将它传递给一个新的read 函数：

var dfWithSchema = spark.read.json(parsed)

此时您将访问嵌套 JSON 中的值：

dfWithSchema.select($"value.propertyInJson")

如果需要，您可以对sanitiseJson 进行一些优化。