Apache Spark 在withinColumn中访问新建列

mum43rcc 于 12个月前发布在 Apache

关注(0)|答案(3)|浏览(167)

我有以下数据集：

|value|
+-----+
|    1|
|    2|
|    3|

字符串
我想创建一个新列 newValue，它从前一行获取 newValue 的值并对其执行一些操作。为了简单起见，只需递增3。如果没有前一列，则在第一行的情况下，应该获取 value。结果如下所示：

|value|newValue|
+-----+--------+
|    1|       1|
|    2|       4| # newValue previous row (1) + 3
|    3|       7| # newValue previous row (4) + 3

型
我尝试了下面的代码，但是在尝试访问前一行时，新列 newValue 似乎还不存在。如何在withColumn中访问新创建的列？

val data = Seq(1, 2, 3)
val dataset: Dataset[Int] = data.toDS()
val windowSpec = Window.orderBy("value")

val result = dataset.withColumn("newValue", coalesce(lag("newValue", 1).over(windowSpec) + 3, $"value"))

型
这将导致以下错误消息：
org.apache.spark.sql.AnalysisException：[UNRESOLVED_COLUMN.WITH_SUGGESTION]无法解析名为“newValue”的列或函数参数。是否为以下之一？[“value”]

apache-spark

来源：https://stackoverflow.com/questions/77357325/access-newly-created-column-in-withincolumn

3条答案

按热度按时间

zfycwa2u1#

我不确定是否可以使用窗口函数来实现这一点。你可以通过将DataFrame转换为RDD或List来进行计算。我在一个例子中测试了它，它似乎可以工作，但不确定这是否是一种优化的方法。

df = spark.sql("select id as value, row_number() over(order by id) as rn from range(1, 4)")

firstVal = df.select("value").first()[0]
dfCount = df.count()
dataList = [[firsrtVal, 1]]

for i in range(2, dfCount + 1):
    dataList.append(list([dataList[-1][0] + 3, i]))

df2 = spark.createDataFrame(dataList, ["newValue", "rn"])

finalDF = df.join(df2, 'rn', 'inner').select("value", "newValue")

字符串

赞(0）回复(0）举报 12个月前

rwqw0loc2#

无法访问以前的聚合值。对于这种特定情况，newValue可以计算为-所有以前值的总和，加上当前位置乘以三：

val dataset: Dataset[Int] = Seq(1, 2, 3, 12, 43).toDS()
val rangeWindow = Window.orderBy("value").rangeBetween(Window.unboundedPreceding, Window.currentRow - 1)
val formula = sum("value").over(rangeWindow) + count("*").over(rangeWindow) * 3

val result = dataset
  .withColumn("newValue", coalesce(formula, lit(1)))

字符串
输出量：

+-----+--------+
|value|newValue|
+-----+--------+
|1    |1       |
|2    |4       |
|3    |9       |
|12   |15      |
|43   |30      |
+-----+--------+

型

赞(0）回复(0）举报 12个月前

fumotvh33#

我相信你所需要的就是running sum和一个常数3

scala> df.show(false)
+-----+
|value|
+-----+
|1    |
|2    |
|3    |
|4    |
|5    |
+-----+

个字符
输出

+-----+--------+
|value|newValue|
+-----+--------+
|1    |1       |
|2    |4       | -- 1  + 3
|3    |7       | -- 4  + 3
|4    |10      | -- 7  + 3
|5    |13      | -- 10 + 3
+-----+--------+

型

赞(0）回复(0）举报 12个月前

我来回答

Apache Spark 在withinColumn中访问新建列

3条答案

相关问题

热门标签

最新问答