【问题标题】:equality of two data frames两个数据帧相等
【发布时间】:2021-09-24 00:57:41
【问题描述】:

我有以下场景:

我有 2 个仅包含 1 列的数据框让我们说

DF1=(1,2,3,4,5)
DF2=(3,6,7,8,9,10)

基本上这些值是键,如果 DF1 中的键不在 DF2 中,我将创建 DF1 的镶木地板文件(在当前示例中,它应该返回 false)。我目前实现我的要求的方法是:

val df1count= DF1.count
val df2count=DF2.count
val diffDF=DF2.except(DF1)
val diffCount=diffDF.count
if(diffCount==(df2count-df1count)) true
else false

这种方法的问题是我调用了 4 次动作元素,这肯定不是最好的方法。有人可以建议我实现这一目标的最佳有效方法吗?

【问题讨论】:

    标签: python scala apache-spark databricks


    【解决方案1】:

    这是一种获取两个数据帧之间不常见行的方法:

    val d1 = Seq((3, "Chennai", "rahman", "9848022330", 45000, "SanRamon"), (1, "Hyderabad", "ram", "9848022338", 50000, "SF"), (2, "Hyderabad", "robin", "9848022339", 40000, "LA"), (4, "sanjose", "romin", "9848022331", 45123, "SanRamon"))
    val d2 = Seq((3, "Chennai", "rahman", "9848022330", 45000, "SanRamon"), (1, "Hyderabad", "ram", "9848022338", 50000, "SF"), (2, "Hyderabad", "robin", "9848022339", 40000, "LA"), (4, "sanjose", "romin", "9848022331", 45123, "SanRamon"), (4, "sanjose", "romino", "9848022331", 45123, "SanRamon"), (5, "LA", "Test", "1234567890", 12345, "Testuser"))
    
    val df1 = d1.toDF("emp_id" ,"emp_city" ,"emp_name" ,"emp_phone" ,"emp_sal" ,"emp_site")
    val df2 = d2.toDF("emp_id" ,"emp_city" ,"emp_name" ,"emp_phone" ,"emp_sal" ,"emp_site")
    
    spark.sql("((select * from df1) union (select * from df2)) minus ((select * from df1) intersect (select * from df2))").show //spark is SparkSession
    

    【讨论】:

      【解决方案2】:

      您可以使用以下功能:

      import org.apache.spark.sql.functions._
      
      def diff(key: String, df1: DataFrame, df2: DataFrame): DataFrame = {
        val fields = df1.schema.fields.map(_.name)
        val diffColumnName = "Diff"
      
        df1
          .join(df2, df1(key) === df2(key), "full_outer")
          .withColumn(
            diffColumnName,
            when(df1(key).isNull, "New row in DataFrame 2")
              .otherwise(
                when(df2(key).isNull, "New row in DataFrame 1")
                  .otherwise(
                    concat_ws("",
                      fields.map(f => when(df1(f) =!= df2(f), s"$f ").otherwise("")):_*
                    )
                  )
              )
          )
          .filter(col(diffColumnName) =!= "")
          .select(
            fields.map(f =>
              when(df1(key).isNotNull, df1(f)).otherwise(df2(f)).alias(f)
            ) :+ col(diffColumnName):_*
          )
      }
      

      在你的情况下运行这个:

      diff("emp_id", df1, df2)
      

      示例

      import org.apache.spark.sql.{DataFrame, SparkSession}
      import org.apache.spark.sql.functions._
      
      object DiffDataFrames extends App {
        val session = SparkSession.builder().master("local").getOrCreate()
      
        import session.implicits._
      
        val df1 = session.createDataset(Seq((1,"a",11),(2,"b",2),(3,"c",33),(5,"e",5))).toDF("n", "s", "i")
        val df2 = session.createDataset(Seq((1,"a",11),(2,"bb",2),(3,"cc",34),(4,"d",4))).toDF("n", "s", "i")
      
        def diff(key: String, df1: DataFrame, df2: DataFrame): DataFrame =
        /* above definition */
      
        diff("n", df1, df2).show(false)
      }
      

      【讨论】:

      • 请告诉我如何声明 df1 和 df2。我已经声明如下 sqlContext = SQLContext(sc) df = sqlContext.sql("select * from table1") df2 = sqlContext.sql("select * from table2") 然后按原样处理上面的代码....获取语法错误....我对这个 spark scala 代码很陌生
      • 你能纠正我做错了什么吗,当我尝试运行下面的代码时,我得到一个错误:未找到:值 df1,未找到 df2 .. import org.apache.spark.sql .{DataFrame, SQLContext} import org.apache.spark.sql.functions._ val sc: SparkContext val sqlContext = new org.apache.spark.sql.SQLContext(sc) sqlContext = SQLContext(sc) df1 = sqlContext.sql( "select * from table1") df2 = sqlContext.sql("select * from table2") diff("tenant",df1,df2) def diff(key: String, df1: DataFrame, df2: DataFrame): DataFrame = { . ..... } /// 提供的 diff 有趣的代码
      • 嗨,我添加了简短的例子。
      • 当数据帧没有 key_column 时,如何连接 Df1 和 DF2 的多个列。如何更新上述 Diff 函数以处理多个键。
      【解决方案3】:

      坚持 DF1 和 DF2。 与非持久化数据相比,持久化将数据保存在内存中(缓存),并且可以在相同的 DF 上执行多个操作(如计数),而计算量(物理执行计划)更少。

      DF1.persist
      DF2.persist
      val df1count= DF1.count
      val df2count=DF2.count
      val diffDF=DF2.except(DF1)
      val diffCount=diffDF.count
      if(diffCount==(df2count-df1count)) true
      else false
      

      【讨论】:

        猜你喜欢
        • 2013-02-17
        • 2013-10-07
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        相关资源
        最近更新 更多