は、二つのデータフレームが誤って2つのレコード間

Question

差があるスパークに新しい列として値を一致追加比較します変更した場合、および&のいずれの列に変更がある場合は、この値を列として追加します。この

Answer 1

のような予想される出力は、UDFのデータを比較し、値を返し、両方の列の値（古いものと新しい値）を渡し、後で列とUDFを使用して、プライマリキー上の2つのデータフレームに参加します同じでない場合。

val check = udf ((old_val:String,new_val:String) => if (old_val == new_val) new_val else "") 

df_check= df 
    .withColumn("Check_Name",check(df.col("name"),df.col("new_name"))) 
    .withColumn("Check_Namelast",check(df.col("lastname"),df.col("new_lastname"))) 

はdef機能

  def fn(old_df:Dataframe,new_df:Dataframe) : Dataframe = 
      { 
      val old_df_array = old_df.collect() //make df to array to loop thru 
      val new_df_array = new_df.collect() //make df to array to loop thru 
      var value_change : Array[String] = "" 

      val count = old_df.count 
      val row_count = old_df.coloumn 
      val row_c = row.length 
      val coloumn_name = old_df.coloumn 

      for (i to count) //loop thru all rows 
      { 
      var old = old_df_array.Map(x => x.split(",")) 
      var new = new_df_array.Map(x => x.split(",")) 
      for (j to row_c) //loop thru all coloumn 
      { 
      if(old(j) != new(j)) 
      { 
      value_change = value_change + coloumn_name(j) " has value changed" ///this will add all changes in one full row 
      } 
      //append to array 
      append j(0) //primary key 
      append value_change //Remarks coloumn 
      } 
      } 
      //convert array to df 
      }