如何计算每一行的值？

Question

我有一个包含 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

Answer 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