如何计算每一行的值？

Posted 2023-03-31

【中文标题】如何计算每一行的值？

【英文标题】：How to calculate a value for each row?

【发布时间】：2018-02-24 14:50:59

【问题描述】：

我有一个包含 100 多行的输入数据框（从配置单元表创建）。对于数据框的每一行，我需要提取列值（大多数字符串）并将这些值传递给用户定义的函数。对于每一行，该函数使用这些输入值和其他中间数据帧（从配置单元表创建）来计算一组行并存储在结果数据帧中。 我如何实现这一点 - 请帮助。

我试过了：

var df1= hiveContext.sql("Select event_date,channelcode,st,tc,startsec,endsec from program_master")
var count1=df1.count()
df1 = df1.withColumn("INDEX", monotonically_increasing_id())
var i=1
while (i <= count1)
  var ed = df1.filter(df1("INDEX") === s"""$i""").select(to_date(unix_timestamp(df1("ed"), "dd-MM-yy").cast(TimestampType)).cast(DateType)).first().getDate(0)
  var cc = df1.filter(df1("INDEX") === s"""$i""").select(df1("cc")).first().getInt(0)
  var ST = df1.filter(df1("INDEX") === s"""$i""").select(df1("ST")).first().getString(0)
  var TC = df1.filter(df1("INDEX") === s"""$i""").select(df1("TC")).first().getString(0)
  var ss = df1.filter(df1("INDEX") === s"""$i""").select(df1("ss")).first().getInt(0)
  var es = df1.filter(df1("INDEX") === s"""$i""").select(df1("es")).first().getInt(0)
  calculate_values(ed, cc, st, tc, ss, ss, sparkSession)
  i=i+1

calculate_values def

def calculate_values(ed: Date,cc:Integer,ST:String,TC:String,ss:Integer,ss:Integer,sparkSession: SparkSession):Unit=

我尝试过的两个问题：因此没有输出 第 3 行：我希望它给出像 1,2,3,......100.... 这样的数字来使用 i 进行迭代 - 但它会生成非常大的随机数。 第 5 行：它抛出 java.util.NoSuchElementException: next on empty iterator

【问题讨论】：

请分享您到目前为止尝试过的内容？以及一些样本输入和预期输出。但在此之前澄清您的要求

添加了导致问题的代码和细节

【参考方案1】：

monotonically_increasing_id() 会生成随机数，但会以递增的方式生成，因此不能像row_number() 函数那样依赖它来生成序列号。但是row_number() 用于整个数据集的成本很高，因为它将在一个执行器中收集所有数据，除非您通过对数据进行分组来使用row_number()。

monotonically_increasing_id() 在您想要对数据进行排序/排序的情况下会很有帮助。

您似乎正在尝试使用 event_date、channelcode、st、tc 逐行计算一些值em>、startsec 和 endsec。

如果它是逐行计算，那么我建议您使用udf 函数。因此，您可以将 calculate_value 函数转换为 udf 函数，如下所示

import org.apache.spark.sql.functions._
def calculate_value = udf((ed: Date,cc:Int,ST:String,TC:String,ss:Int,es:Int) => //write your calculation part here)

你调用udf函数使用withColumn作为

df1.withColumn("calculated", calculate(col("ed"), col("cc"), col("ST"), col("TC"), col("ss"), col("es"))

将使用计算值创建一个新列

但如果可以按列进行计算，我建议您也查看inbuilt functions

【讨论】：

谢谢。在 calculate_values 函数中，我实际上必须填充一个结果数据帧（我将其定义为全局数据帧 - 这样我就可以在每次迭代中使用 union 添加），因为对于输入数据帧的每一行，将形成一组行，我需要在后面的代码中保存和使用。使用您的方法，我可以在我的计算函数中执行此操作吗？

是的。我建议的方式为您提供了一个新列，其中包含逐行计算的值，如果您想要一个单独的数据框，那么您只能将该列选择到一个新的数据框中。尝试一下。如果答案是有帮助的，我敢肯定，你可以投票并接受答案。 :)

还有一个问题 - 像我使用的那样使用 while 循环是否有问题？说，而不是 monotonically_increasing_id()，我使用 row_number() 并获取索引。我问的原因是在代码的后面部分，我可能有更多的循环会遇到棘手的情况：一行的每个循环迭代都会产生一个结果，该结果将在下一次迭代中使用。我写的那种 while 循环也是考虑到这一点而设计的。

你绝对可以写出来。没有问题。但是使用内置函数会非常有效。如果您想要之前的行值，那么您可以使用 lag 函数，还有更多您可以从中受益的内置函数。