使用 Spark 和 Scala 从两个表中提取所需信息

Posted 2023-04-15

【中文标题】使用 Spark 和 Scala 从两个表中提取所需信息

【英文标题】：Extracting required information from two tables using Spark and Scala

【发布时间】：2017-08-07 18:33:57

【问题描述】：

我必须连接两个表并将 col3 保留在 table1 中，如果 table1 中不存在数据但存在于 table2 中，则可以这样处理。

表 1

col1  col2   col3

,,,,  ,,,,   ,,,,
a      p     d
b      q     e
c      r     f
d      s     g

表 2

col1  col2   col3

,,,,  ,,,,   ,,,,
a      m     s
e      q     l

需要的输出

col1  col2   col3
,,,,  ,,,,   ,,,,
a      m     d
b      q     e
c      r     f
d      s     g
e      q     l

我尝试了以下代码

import org.apache.spark.sql.Row, SQLContext
import org.apache.spark.SparkConf, SparkContext

object Tasktr 
  def main(args: Array[String]): Unit = 
    val sparkConf = new SparkConf().setAppName("Myapp").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)

    val df1 = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("/home/magnetto/Desktop/ip.csv")
    val df2= sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("/home/magnetto/Desktop/del.csv")
    df1.registerTempTable("tab1")
    df2.registerTempTable("tab2")
    val df3=sqlContext.sql("select * from tab1 FULL OUTER JOIN tab2 ON tab1.val1=tab2.val1").show()

我得到以下输出

val1|val2|val3|val1|val2|val3|
+----+----+----+----+----+----+
|  k1|  l1|   a|null|null|null|
|  k2|  l2|   b|null|null|null|
|  k3|  l3|   c|null|null|null|
|  k4|  l4|   d|  k4|  l4|   e|
|null|null|null|  k5|  l9|   f|
+----+----+----+----+----+----+

我已尝试对两个表进行完全外部联接，但无法获得所需的输出。

【问题讨论】：

发布您的代码和想要的输出

我用过上面的代码

【参考方案1】：

给定数据框为

表1

+----+----+----+
|col1|col2|col3|
+----+----+----+
|a   |p   |d   |
|b   |q   |e   |
|c   |r   |f   |
|d   |s   |g   |
+----+----+----+

和

表2

+----+----+----+
|col1|col2|col3|
+----+----+----+
|a   |m   |s   |
|e   |q   |l   |
+----+----+----+

您可以通过执行以下操作来实现您想要的结果。重命名是为了避免在join之后应用函数时发生冲突

val tempTable2 = table2.select('col1, 'col2.as("col22"), 'col3.as("col23"))

import org.apache.spark.sql.functions._
table1.join(tempTable2, Seq("col1"), "outer")
  .withColumn("col2", when('col22.isNotNull, 'col22).otherwise('col2))
  .withColumn("col3", when('col3.isNull, 'col23).otherwise('col3))
  .drop("col22", "col23")

这会给你结果

+----+----+----+
|col1|col2|col3|
+----+----+----+
|e   |q   |l   |
|d   |s   |g   |
|c   |r   |f   |
|b   |q   |e   |
|a   |m   |d   |
+----+----+----+

【讨论】：

有什么方法可以在不重命名列的情况下做...类似于 df1.col1,df1.col2 和 df2.col1,df2.col2