異なるデータフレームからの行の変化を検出する方法

Question

2つの異なるタイムスタンプの人の値を保持する2つのデータフレームがあります。前後の人の可能な変更は、以下のコードに記載されています。

val before = Seq(
(1, "soccer", "1", "2", "3", "4", ""), 
(2, "soccer", "", "", "", "", ""), 
(3, "soccer", "1", "", "", "", ""), 
(4, "soccer", "1", "", "", "", ""), 
(5, "soccer", "1", "", "", "", ""), 
(6, "soccer", "1", "", "", "", "") 
).toDF("id", "sport", "var1", "var2", "var3", "var4", "var5") 

before.show     //> +---+------+----+----+----+----+----+ 
           //| | id| sport|var1|var2|var3|var4|var5| 
           //| +---+------+----+----+----+----+----+ 
           //| | 1|soccer| 1| 2| 3| 4| | 
           //| | 2|soccer| | | | | | 
           //| | 3|soccer| 1| | | | | 
           //| | 4|soccer| 1| | | | | 
           //| | 5|soccer| 1| | | | | 
           //| | 6|soccer| 1| | | | | 
           //| +---+------+----+----+----+----+----+ 
           //| 

val after = Seq(
(1, "soccer", "1", "2", "3", "4", ""), // Same 
(2, "soccer", "1", "", "", "", ""), // Addition 
(3, "soccer", "1", "1", "", "", ""), // Addition 
(4, "soccer", "", "", "", "", ""), // Remove 
(5, "soccer", "2", "1", "", "", ""), // Slide 
(6, "soccer", "2", "", "", "", "") // Change 
).toDF("id", "sport", "var1", "var2", "var3", "var4", "var5") 

after.show     //> +---+------+----+----+----+----+----+ 
           //| | id| sport|var1|var2|var3|var4|var5| 
           //| +---+------+----+----+----+----+----+ 
           //| | 1|soccer| 1| 2| 3| 4| | 
           //| | 2|soccer| 1| | | | | 
           //| | 3|soccer| 1| 1| | | | 
           //| | 4|soccer| | | | | | 
           //| | 5|soccer| 2| 1| | | | 
           //| | 6|soccer| 2| | | | | 
           //| +---+------+----+----+----+----+----+ 
           //| 

物事は同じままにすることができますが、追加や削除があり、最後に変更やスライドがある可能性があります。

私の理想的な出力を前にし、データフレームの後に、各行に直面し、ラベル添付するものです：

outcome.show     //> +---+------+------+ 
           //| | id| sport| diff| 
           //| +---+------+------+ 
           //| | 1|soccer| same| 
           //| | 2|soccer| add| 
           //| | 3|soccer| add| 
           //| | 4|soccer|remove| 
           //| | 5|soccer| slide| 
           //| | 6|soccer|change| 
           //| +---+------+------+ 
           //| 

この質問はthis 1に関連しているのが、ポイントはどのように多くの違いをカウントするだけでした2つの行の間にありました...今度は、これらの違いを細かい穀粒で理解しようとしていますが、私はさまざまな可能なオプションを定義することに固執しています。

EDIT

私はデータフレームを使用しておりますので、私はこのような構造に固執するのではなくケースクラスを使用したいと思います。私は、代わりにDataFrameを使用して@ibossによって提案されたものを適応させようとしています。

私はすべての作業を行う必要があり、このUDFを持っている：

val diff = udf { (bef:DataFrame, aft:DataFrame) => { 
    "hello" // return just this string for now 
    } : String 
} 

このUDFは@ibossにより示唆されるようにoutcome.showで出力を生成するために、すべての作業を行うので、後の可能な結果になります2つの行を一致させると、より正確には "同じ"、 "追加"、 "削除"、 "スライド"または "変更"のいずれかの文字列になります。

私はその後、2つのデータフレームをマージして新しい列を作成するには、このコードしている。このように文句を言い差分を呼び出すとき

val mydiff = before.join(after, "id") 
    .withColumn("diff", diff(before, after)) 
    .select("id", "diff") 

は、しかし、私はエラーを持っている：

type mismatch; found : org.apache.spark.sql.DataFrame required: org.apache.spark.sql.Column 

何を私は理解していない理由は、それがDataFrameを好きではないとそれを解決する方法です...

Answer 1

私はそれらのvarsは何かをはっきりとはわかりませんが、私はあなたがタプルまたはさらなる処理のために容易なケースクラス。それは次のようになります。

val before = Seq(
    (1, "soccer", ("1", "2", "3", "4", "")), 
    (2, "soccer", ("", "", "", "", "")), 
    (3, "soccer", ("1", "", "", "", "")), 
    (4, "soccer", ("1", "", "", "", "")), 
    (5, "soccer", ("1", "", "", "", "")), 
    (6, "soccer", ("1", "", "", "", "")) 
).toDF("id", "sport", "vars") 


val after = Seq(
    (1, "soccer", ("1", "2", "3", "4", "")), 
    (2, "soccer", ("1", "", "", "", "")), 
    (3, "soccer", ("1", "1", "", "", "")), 
    (4, "soccer", ("", "", "", "", "")), 
    (5, "soccer", ("2", "1", "", "", "")), 
    (6, "soccer", ("2", "", "", "", "")) 
).toDF("id", "sport", "vars") 

次に、あなたがあなたの差分

type MyVars = (String, String, String, String, String) 

val diff = udf { (before_vars: MyVars, after_vars: MyVars) => 
    // your implementation of diff function 
} 

before 
    .join(after) 
    .withColumn("diff", diff(before("vars"), after("vars"))) 
    .select("id", "sport", "diff") 

編集計算するために、ユーザー定義関数を使用することができ

UDFの場合は、通常、彼らはのための型推論を行うだろうあなたのタイプを定義する必要はないかもしれません。しかし、あなたはそれを定義したい場合は、その後、あなたはこのよう

udf { (firstName: String, lastName: String) => s"$firstName $lastName": String } 

またはブロック

udf { (name: String) => { 
    val hello = "hello " 
    "hello, " + name 
}: Int } 

でそれを行うことができますし、defを使用しているので、あなたもdef

def getFullName(firstName: String, lastName: String): String = 
    s"$firstName $lastName" 

udf(getFullName _) 

を使用することができます関数を定義するのではなくメソッドであり、udfは関数を必要とします。したがって、部分的なアプリケーション表記法を使用して変換する必要があります。詳細については

、あなたはこのDifference between method and function in Scala

編集2

を見てとることができ、私は少しあなたの質問を誤解ように思えます。 diff udfは、各行に別々に適用する必要があります。したがって、DataFrame全体に渡すことはできません。

読みやすいので、（各行内の）変数をタプルにグループ化することをお勧めします。それでも、元のフォームを使用したい場合しかし、あなたは2つのMySQLのテーブルから来たこの

val diff = udf { (
    beforeVar1: String, 
    beforeVar2: String, 
    beforeVar3: String, 
    beforeVar4: String, 
    beforeVar5: String, 
    afterVar1: String, 
    afterVar2: String, 
    afterVar3: String, 
    afterVar4: String, 
    afterVar5: String 
) => { 
    "hello" // return just this string for now 
    } : String 
} 

before.join(after, "id") 
    .withColumn("diff", diff(
    before("var1"), 
    before("var2"), 
    before("var3"), 
    before("var4"), 
    before("var5"), 
    after("var1"), 
    after("var2"), 
    after("var3"), 
    after("var4"), 
    after("var5"), 
)) 
    .select("id", "diff")