使用其他现有列 Spark/Scala 添加新列

Posted

技术标签:

【中文标题】使用其他现有列 Spark/Scala 添加新列【英文标题】:Adding new column using other existing columns Spark/Scala 【发布时间】:2021-05-12 15:08:00 【问题描述】:

我想使用其他现有列添加新列。这必须在有条件的情况下释放。这是我的数据框的一个例子:

val data = Seq(("WHT20177", "CTHT WO/MTR# : WHT20212/BTI0426; WHT20177/BTH0393"),
                ("WHT55637", "CTHT WO/MTR# : WHT50747/BTI2699; WHT55637/SQL1239"))

val dataFrame = data.toDF("prev_wo", "ref_wo")

+--------+-------------------------------------------------+
|prev_wo |ref_wo                                           |
+--------+-------------------------------------------------+
|WHT20177|CTHT WO/MTR# : WHT20212/BTI0426; WHT20177/BTH0393|
|WHT55637|CTHT WO/MTR# : WHT50747/BTI2699; WHT55637/SQL1239|
+--------+-------------------------------------------------+

“ref_wo”列必须包含“prev_wo”,仅在这种情况下,我必须将以下元素放入我将命名为“col1”的新列中。

第一行要提取的值为"BTH0393",第二行要提取的值为"SQL1239" 我正在使用两种不同的方法在 Spark Scala 中尝试这个。第一个只对第一行有反应,第二个只对第二行有反应。

第一种方法:

def addNewColumn(df: DataFrame): DataFrame = 

  val prev_wo = dataFrame.select("prev_wo").collectAsList().get(0).mkString(",")

  val regex_extract = ("(?<=" + prev_wo + "\\/)(.7)").r

  df
    .withColumn("col1",
      when($"ref_wo".contains(col("prev_wo")),
        regexp_extract(col("ref_wo"), regex_extract.toString(), 1))
        .otherwise(null)
    )



val new_dataFrame = dataFrame
  .transform(addNewColumn)

输出:

+--------+-------------------------------------------------+-------+
|prev_wo |ref_wo                                           |col1   |
+--------+-------------------------------------------------+-------+
|WHT20177|CTHT WO/MTR# : WHT20212/BTI0426; WHT20177/BTH0393|BTH0393|
|WHT55637|CTHT WO/MTR# : WHT50747/BTI2699; WHT55637/SQL1239|       |
+--------+-------------------------------------------------+-------+

第二种方法:

def addColumn(df: DataFrame): DataFrame = 
  var out = df

  df.collect().foreach(row => 

    val prev_wo = row.getValuesMap(Seq("prev_wo")).get("prev_wo").getOrElse("")

    val regex_extract = ("(?<=" + prev_wo + "\\/)(.7)").r

    out = out
      .withColumn("col1",
        when($"ref_wo".contains(col("prev_wo")),
          regexp_extract(col("ref_wo"), regex_extract.toString(), 1))
          .otherwise(null)
      )
  )

  out


val new_dataFrame = dataFrame
  .transform(addColumn)

输出

+--------+-------------------------------------------------+-------+
|prev_wo |ref_wo                                           |col1   |
+--------+-------------------------------------------------+-------+
|WHT20177|CTHT WO/MTR# : WHT20212/BTI0426; WHT20177/BTH0393|       |
|WHT55637|CTHT WO/MTR# : WHT50747/BTI2699; WHT55637/SQL1239|SQL1239|
+--------+-------------------------------------------------+-------+

【问题讨论】:

【参考方案1】:

您可以将regexp_extract 与从prev_wo 动态生成的模式一起使用:

dataFrame.withColumn("col1", expr("regexp_extract(ref_wo, concat(prev_wo, '/(.7)'), 1)")).show(false)
+--------+-------------------------------------------------+-------+
|prev_wo |ref_wo                                           |col1   |
+--------+-------------------------------------------------+-------+
|WHT20177|CTHT WO/MTR# : WHT20212/BTI0426; WHT20177/BTH0393|BTH0393|
|WHT55637|CTHT WO/MTR# : WHT50747/BTI2699; WHT55637/SQL1239|SQL1239|
+--------+-------------------------------------------------+-------+

【讨论】:

以上是关于使用其他现有列 Spark/Scala 添加新列的主要内容,如果未能解决你的问题,请参考以下文章

spark scala - UDF 用于创建新列

避免指定模式两次(Spark/scala)

如何基于两个现有列使用 pandas 正确添加新列?

解决如何在现有表中添加列

如何使用微软互操作将工作表开头的新列添加到现有 excel

如何将新列添加到现有表 symfony - orocommerce