将 Spark 数据框列与其在 Scala 中的行连接起来答案

【问题标题】：Concatenate spark data frame column with its rows in Scala将 Spark 数据框列与其在 Scala 中的行连接起来
【发布时间】：2019-03-13 11:01:06
【问题描述】：

我正在尝试通过连接数据框中的值来构建字符串。例如：

val df = Seq(
  ("20181001","10"),     
  ("20181002","40"),
  ("20181003","50")).toDF("Date","Key")
df.show

DF 的输出如下。

这里我想根据数据框的值来构建条件，比如：(Date=20181001 and key=10) or (Date=20181002 and key=40) or (Date=20181003 and key=50) ) 等等..生成的条件将作为另一个进程的输入。这里数据框中的列可以是动态的。

下面的 sn-p 我正在尝试，它正在根据需要形成字符串，但它是一个静态字符串。当我必须为超过 10 列生成条件时，我也不太确定它会如何执行。任何建议都受到高度赞赏。

val df = Seq(
  ("20181001","10"),     
  ("20181002","40"),
  ("20181003","50")).toDF("Date","Key")

val colList = df.columns
var cond1 = ""
var finalCond =""
for (row <- df.rdd.collect)
 {
    cond1 = "("
    var pk = row.mkString(",").split(",")(0)
    cond1 = cond1+colList(0)+"="+pk
    var ak = row.mkString(",").split(",")(1)
    cond1 = cond1 +" and " + colList(1)+ "=" +ak +")"
    finalCond = finalCond + cond1 + " or " 
    cond1= ""    
 }
 print("Condition:" +finalCond.dropRight(3))

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

检查这个 DF 解决方案。

scala> val df = Seq(
       |   ("20181001","10"),
       |   ("20181002","40"),
       |   ("20181003","50")).toDF("Date","Key")
df: org.apache.spark.sql.DataFrame = [Date: string, Key: string]

scala> val df2 = df.withColumn("gencond",concat(lit("(Date="), 'Date, lit(" and Key=") ,'Key,lit(")")))
df2: org.apache.spark.sql.DataFrame = [Date: string, Key: string ... 1 more field]


scala> df2.agg(collect_list('gencond)).show(false)
+------------------------------------------------------------------------------------+
|collect_list(gencond)                                                               |
+------------------------------------------------------------------------------------+
|[(Date=20181001 and Key=10), (Date=20181002 and Key=40), (Date=20181003 and Key=50)]|
+------------------------------------------------------------------------------------+

EDIT1

您可以从 parquet 文件中读取它们，只需像在此解决方案中一样更改名称。在最后一步中，再次替换 parquet 标题中的名称。检查一下。

scala> val df = Seq(("101","Jack"),("103","wright")).toDF("id","name")  // Original names from parquet
df: org.apache.spark.sql.DataFrame = [id: string, name: string]

scala> val df2= df.select("*").toDF("Date","Key")  // replace it with Date/Key as we used in this question
df2: org.apache.spark.sql.DataFrame = [Date: string, Key: string]

scala> val df3 = df2.withColumn("gencond",concat(lit("(Date="), 'Date, lit(" and Key=") ,'Key,lit(")")))
df3: org.apache.spark.sql.DataFrame = [Date: string, Key: string ... 1 more field]

scala> val df4=df3.agg(collect_list('gencond).as("list"))
df4: org.apache.spark.sql.DataFrame = [list: array<string>]

scala> df4.select(concat_ws(" or ",'list)).show(false)
+----------------------------------------------------+
|concat_ws( or , list)                               |
+----------------------------------------------------+
|(Date=101 and Key=Jack) or (Date=103 and Key=wright)|
+----------------------------------------------------+

scala> val a = df.columns(0)
a: String = id

scala> val b = df.columns(1)
b: String = name

scala>  df4.select(concat_ws(" or ",'list).as("new1")).select(regexp_replace('new1,"Date",a).as("colx")).select(regexp_replace('colx,"Key",b).as("colxy")).show(false)
+--------------------------------------------------+
|colxy                                             |
+--------------------------------------------------+
|(id=101 and name=Jack) or (id=103 and name=wright)|
+--------------------------------------------------+


scala>

【讨论】：

太棒了！但是如何使列动态而不是硬编码？
您在询问“df”值吗？ - 您可以拥有一个 csv 文件，然后使用 spark.read.csv("csv_file") 读取它们，它会为您提供数据帧
我曾经根据 parquet 文件创建 df 并从 json 文件中获取这些条件，并且假设 JSON 中提到的列（在本例中为 Date 和 Key）在df。所以我需要根据 JSON 中可用的列来构建条件，这些列可以作为列表提供给我。
得到你..检查我的EDIT1

【解决方案2】：

调用 collect 会将结果拉回驱动程序，因此如果您有一个巨大的 DataFrame，您很可能会耗尽内存。

如果您确定只处理少量的行，这不是问题。

你可以这样做：

df.map(row => s"($Date={row.getString(0)} and Key=${row.getString(1)})").collect.mkString("Condition: ", " or ", "")

输出：

res2: String = Condition: (Date=20181001 and Key=10) or (Date=20181002 and Key=40) or (Date=20181003 and Key=50)

【讨论】：

谢谢..您的解决方案运行良好。胸围改变了一点点。df.rdd.map(row => s"({Date=${row.getString(0)} 和 Key= ${row.getString(1)})").collect.mkString("条件：", " or ", "")

【解决方案3】：

使用udf，您可以对columns 的可变数量进行如下操作

val list=List("Date","Key")

def getCondString(row:Row):String={
    "("+list.map(cl=>cl+"="+row.getAs[String](cl)).mkString(" and ")+")"
  }

val getCondStringUDF=udf(getCondString _)
df.withColumn("row", getCondStringUDF(struct(df.columns.map(df.col(_)):_*))).select("row").rdd.map(_(0).toString()).collect().mkString(" or ")

【讨论】：