检查加入流是否成功使用 Apache Spark - Scala

Question

我是使用 Scala 的 Apache Spark 的新手。我可以使用以下命令将表加入流：

Updated_DF = Inbound_DF.join(colToAdd, colToAdd("key") <=> Inbound_DF("key"), "left")
    .withColumnRenamed("Data_DF","site").drop("Id","key")

现在我想检查colToAdd("key") 和Inbound_DF("key") 是否匹配并且加入是否成功。例如，colToAdd：

Id   key     Data_DF
S31  S3   {"name":"nick","region":"IN"}
S21  S2   {"name":"john","region":"CA"}
S11  S1   {"name":"ashley","region":"CA"}
S51  S5   {"name":"bella","region":"UK"}
S41  S4   {"name":"kumar","region":"In"}
S6   S6   {"name":"ben","region":"US"}
P11  P1   {"name":"MKD","region":"UAE"}
P21  P2   {"name":"ahmad","region":"UAE"}

来自传入流的消息如下所示：

cusId  key  item  price
1897   S2   book   54

加入后，更新后的消息应如下所示：

cusId  key  item  price    site
1897   S2   book   54    {"name":"john","region":"CA"}

但如果我收到一条带有key = S9 的流消息，则不会发生加入，然后我想记录一条消息：

------- join failed, key not found ---------

据我所知，这可以使用filter 方法来实现，但我不确定如何实现。请帮助我如何做到这一点，或者有没有更好的方法来做到这一点。

Answer 1

有多种方法可以做到这一点。我只是向您提供如何做到这一点的想法，您可以根据自己的用例进行调整。

首先，您进行左连接的方式不正确，您需要交换数据框。 Stream 数据框应该是左数据框。

//Source data
val df = Seq(("S31","S3","""{"name":"nick","region":"IN"}"""),("S21","S2","""{"name":"john","region":"CA"}"""),("S11","S1","""{"name":"john","region":"CA"}""")).toDF("Id","Key","Data_DF")
val df1 = Seq((1897,"S2","book",54),(1920,"S9","movie",200)).toDF("custId","Key","item","price")
//initial join and the count of the records
val df2 = df1.join(df,Seq("Key"),"left").drop("Id").withColumnRenamed("Data_DF","site")
val initialjoincount = df2.count()
//filter and count of the records
val filteredDF = df2.filter($"site".isNotNull)
val filtereddfcount = filteredDF.count()
//compare both the counts and print message/log
if(filtereddfcount == initialjoincount)
{
  println("Join Happened")
}
else
{
  println("Value not found in stream.")
}