使用scala基于Spark DataFrame中现有列的聚合添加新列

Question

我有一个如下的数据框。我需要在现有列的基础上创建一个新列。

col1 col2
a      1
a      2
b      1
c      1
d      1
d      2

输出数据帧看起来像这样

col1  col2 col3 col4
a      1   1      2
a      2   1      2
b      1   0      1
c      1   0      1
d      1   1      2
d      2   1      2

我用来寻找col3的逻辑是如果col1计数

我熟悉如何在sql中实现这一点。但很难找到数据帧DSL的解决方案。任何帮助都将不胜感激。谢谢

Answer 1

要添加col3，您可以在以下情况下使用withcolumn：

val df2 = df.withColumn("col3",when($"col2" > 1, 1).otherwise(0))

要添加col4，前面提到的groupBy/max join应该完成以下工作：

val df3 = df2.join(df.groupBy("col1").max("col2"), "col1")

Answer 2

Sparkdf有一个属性，名为与列。您可以根据需要添加任意多的派生列。但是该列没有添加到现有的DF中，而是创建了一个添加了列的新DF。

向数据添加静态日期

val myFormattedData = myData.withColumn("batchdate",addBatchDate(myData("batchdate")))
val addBatchDate = udf { (BatchDate: String) => "20160101" }

Answer 3

groupBy col1和aggregate得到count和max。然后你可以把它和原始数据帧连接起来，得到你想要的结果

val df2 = df1.groupBy("col1").agg(count() as col3, max("col2") as col4) 

val df3 = df1.join(df2, "col1")