如何根据具有相同 id 的另一行更新一行

Question

使用 Spark 数据框，我想根据具有相同 id 的其他行更新行值。

例如， 我有以下记录，

id,value
1,10
1,null
1,null
2,20
2,null
2,null

我想得到如下结果

id,value
1,10
1,10
1,10
2,20
2,20
2,20

总而言之，某些行中的 value 列为 null，如果有另一行具有相同 id 且具有有效值的行，我想更新它们。

在sql中，我可以简单地用inner-join写一个更新语句，但是我在Spark-sql中没有找到同样的方法。

更新 combineCols a 内连接 combineCols b 在 a.id = b.id 设置 a.value = b.value （这就是我在sql中的做法）

Answer 1

让我们使用 SQL 方法来解决这个问题 -

myValues = [(1,10),(1,None),(1,None),(2,20),(2,None),(2,None)]
df = sqlContext.createDataFrame(myValues,['id','value'])

df.registerTempTable('table_view')
df1=sqlContext.sql(
    'select id, sum(value) over (partition by id) as value from table_view'
)
df1.show()
+---+-----+
| id|value|
+---+-----+
|  1|   10|
|  1|   10|
|  1|   10|
|  2|   20|
|  2|   20|
|  2|   20|
+---+-----+

警告： 这些代码假定任何特定的id 都只有一个non-null 值。当我们groupby 值时，我们必须使用aggregation 函数，而我使用了sum。如果任何id 有2 个non-null 值，则将求和。如果id 可以有多个non-null 值，那么最好使用min/max，这样我们就可以得到其中一个值而不是sum。

df1=sqlContext.sql(
    'select id, max(value) over (partition by id) as value from table_view'
)

Answer 2

您可以使用 window 来执行此操作（在 pyspark 中）：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# create dataframe
df = sc.parallelize([
    [1,10],
    [1,None],
    [1,None],
    [2,20],
    [2,None],
    [2,None],
]).toDF(('id', 'value'))

window = Window.partitionBy('id').orderBy(F.desc('value'))
df \
    .withColumn('value', F.first('value').over(window)) \
    .show()

结果：

+---+-----+
| id|value|
+---+-----+
|  1|   10|
|  1|   10|
|  1|   10|
|  2|   20|
|  2|   20|
|  2|   20|
+---+-----+

您可以在 scala 中使用相同的功能。