使用 Spark 进行 Mongodb 聚合

Question

我在使用 Mongodb 和 Spark 时遇到了聚合问题。我不是这方面的专家，我什至不知道是否存在我需要的操作。

我有几条记录，已经按用户名汇总。然后是这个特殊的用户名，叫做“-”。

username1 data1:100
username2 data1:100
username3 data1:100
username4 data1:100
-         data1:55

现在，我需要将用户名 "-" 中的 data1 sum 到所有其他 data1。

username1 data1:155
username2 data1:155
username3 data1:155
username4 data1:155
-         data1:55

如何使用 mongodb spark 制作这个？

其实我有

rawDataRows.///some stuff//.groupBy("username")

这会产生我写的输出，有人可以帮我将用户名“-”中的数据“合并”到所有其他用户吗？

Answer 1

将 MongoDB 放在一边（JSON 文档 - 我现在没有这样的资源可以使用，请参阅 https://docs.mongodb.com/spark-connector/current/python/read-from-mongodb/），然后：

• 如果您已经汇总数据并且只有 1 条“-”记录，
  • 您所要做的就是将此单条记录过滤到另一个数据帧 DF1，
    • 然后将原始数据 '-' 过滤为 DF2，
      • 然后加入并添加值，
        • 选择所需的列。

不需要 GROUP BY。

像这样：

// Simple data gen: JSON - need to read your MongoDB doc in yourself
import scala.collection.mutable.ListBuffer
val json_content1 = "{'username': 'hello', 'data1': 32}"
val json_content2 = "{'username': 'hello2', 'data1': 64}"
val json_content3 = "{'username': '-', 'data1': 100}"


var json_seq = new ListBuffer[String]()
json_seq += json_content1
json_seq += json_content2
json_seq += json_content3

// JSON in
val json_ds = json_seq.toDS()
json_ds.show(false)

// Make a DF
val df= spark.read.json(json_ds).cache()
df.show(false)

// Get the 2 sets of data
val df1 = df.filter($"username" === "-" ).toDF("dataNull","usernameNull")
val df2 = df.filter($"username" =!= "-" )

df1.show(false)
df2.show(false)

// Add together and Bob's your uncle
val res = df1.join(df2).withColumn("data1", 'data1 + 'dataNull).drop('dataNull).drop('usernameNull)
res.show(false)

返回：

+-----+--------+
|data1|username|
+-----+--------+
|132  |hello   |
|164  |hello2  |
+-----+--------+

对于我的示例数据。