如何使用 Scala 比较 Spark Dataframes 中的两列数据答案

【问题标题】：How to compare two columns data in Spark Dataframes using Scala如何使用 Scala 比较 Spark Dataframes 中的两列数据
【发布时间】：2018-07-20 17:24:09
【问题描述】：

我想比较 Spark DataFrame 中的两列：如果在另一列 (attr_valuelist) 的值中找到列的值 (attr_value)，我只想保留该值。否则，列值应为null。

例如，给定以下输入

id1 id2   attrname  attr_value   attr_valuelist
1   2     test      Yes          Yes, No
2   1     test1     No           Yes, No
3   2     test2     value1       val1, Value1,value2

我希望得到以下输出

id1 id2   attrname  attr_value   attr_valuelist
1   2     test      Yes          Yes
2   1     test1     No           No
3   2     test2     value1       Value1

【问题讨论】：

如果attr_value的值不在attr_valuelist中，那行应该保持不变吗？
Spark 自定义转换可能会有所帮助
请将第二列的值改为null，第一列的值不变。

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

【解决方案1】：

根据您的示例输入，我假设带有搜索项的列包含一个字符串，而搜索目标是一个字符串序列。另外，我假设您对不区分大小写的搜索感兴趣。

这将是输入（我添加了一个可以产生 null 的列来测试我编写的 UDF 的行为）：

+---+---+--------+----------+----------------------+
|id1|id2|attrname|attr_value|attr_valuelist        |
+---+---+--------+----------+----------------------+
|1  |2  |test    |Yes       |[Yes, No]             |
|2  |1  |test1   |No        |[Yes, No]             |
|3  |2  |test2   |value1    |[val1, Value1, value2]|
|3  |2  |test2   |value1    |[val1, value2]        |
+---+---+--------+----------+----------------------+

你可以用一个非常简单的 UDF 来解决你的问题。

val find = udf {
  (item: String, collection: Seq[String]) =>
    collection.find(_.toLowerCase == item.toLowerCase)
}

val df = spark.createDataFrame(Seq(
  (1, 2, "test", "Yes", Seq("Yes", "No")),
  (2, 1, "test1", "No", Seq("Yes", "No")),
  (3, 2, "test2", "value1", Seq("val1", "Value1", "value2")),
  (3, 2, "test2", "value1", Seq("val1", "value2"))
)).toDF("id1", "id2", "attrname", "attr_value", "attr_valuelist")

df.select(
  $"id1", $"id2", $"attrname", $"attr_value",
  find($"attr_value", $"attr_valuelist") as "attr_valuelist")

showing 最后一个命令的输出将产生以下输出：

+---+---+--------+----------+--------------+
|id1|id2|attrname|attr_value|attr_valuelist|
+---+---+--------+----------+--------------+
|  1|  2|    test|       Yes|           Yes|
|  2|  1|   test1|        No|            No|
|  3|  2|   test2|    value1|        Value1|
|  3|  2|   test2|    value1|          null|
+---+---+--------+----------+--------------+

您可以在任何spark-shell 中执行此代码。如果您在提交到集群的作业中使用它，请记住import spark.implicits._。

【讨论】：

【解决方案2】：

你能试试这个代码吗？我认为它适用于 SQL contains case when。

val emptyRDD = sc.emptyRDD[Row] 

var emptyDataframe = sqlContext.createDataFrame(emptyRDD, your_dataframe.schema)

your_dataframe.createOrReplaceTempView("tbl")  

emptyDataframe = sqlContext.sql("select id1, id2, attrname, attr_value, case when
attr_valuelist like concat('%', attr_value, '%') then attr_value else
null end as attr_valuelist from tbl") 

emptyDataframe.show

【讨论】：