SparkSQL создает новый столбец на основе выражения

Question

У меня есть фрейм данных с именем ipTraffic со схемой:

ipTraffic: org.apache.spark.sql.DataFrame = [ip: string, record_count: double]

, и я пытаюсь создать новый столбец, который принимает максимальное значение столбца "record_count" и делит назначение количества записей этой строки.

Я запустил:

val calc = ipTraffic.agg(max("record_count")) / (ipTraffic("record_count"))
ipTraffic = ipTraffic.withColumn("weight", expr(calc))

и

val calc = ipTraffic.agg(max("record_count")).divide(ipTraffic("record_count"))
ipTraffic = ipTraffic.withColumn("weight", expr(calc))`

и получил ошибку

error: value / is not a member of org.apache.spark.sql.DataFrame

Что не имеет смысла для меня с тех порделение определенно в искре (очевидно), но я пошел на https://spark.apache.org/docs/2.3.0/api/sql/ и нашел его в любом случае, и "/" включено.

Answer 1

Вы пытаетесь разделить фрейм данных с помощью столбца:

ipTraffic.agg(max("record_count")):

+-----------------+
|max(record_count)| 
+-----------------+
|              3.0|
+-----------------+ 

, разделенного на:

ipTraffic("record_count"):
+------------+
|record_count|
+------------+
|         1.0|
|         2.0|
|         3.0|
|         1.0|
|         2.0|
|         3.0|
+------------+

Вместо этого вы можете сначала вычислить максимальное значение, получить его как буквальное значение изатем используйте его в своем расчете:

import spark.implicits._     
val maxRecordCount = ipTraffic.agg(max($"record_count")).first.getDouble(0)
val ipTrafficWithWeight = ipTraffic.withColumn("weight", lit(maxRecordCount) / $"record_count")