Java中的Spark記錄鏈接

我需要根據某些欄位的等效性或相似性來記錄兩個資料集的鏈接。例如，假設資料集如下所示（帶有一些隨機資料）：

答：

乙：

當B集中的記錄與A集中的記錄“匹配”時，我希望建立一個鏈接（ID 交換）。

讓我們說：

• 乙記錄必須包含這兩個名字和姓氏從一個在其全名
• B記錄的出生日期必須匹配A的出生日期
• 一的地址必須包含乙的街在它

兩組的約瑟夫·彼得斯都符合所有這些規則。

我將如何用 Spark 做到這一點？一個小例子真的很感激，因為互聯網上沒有很多資訊，它要么是用 Scala 寫的，要么是用 Python 寫的。

UPD：如果有人可以用 Scala 展示 - 好的，我也可以理解 Scala，然后可能會翻譯成 Java。

uj5u.com熱心網友回復：

我沒有用你的資料測驗這段代碼，但希望它有效。在科特林上：

val datasetA: Dataset<Row> = ...
val datasetB: Dataset<Row> = ...

val condition = datasetA.col("BirthDate").equalTo(datasetB.col("BirthDate"))
        .and(datasetB.col("FullName").contains(datasetA.col("FirstName")))
        .and(datasetB.col("FullName").contains(datasetA.col("LastName")))
        .and(datasetB.col("Address").contains(datasetA.col("Street")))
val result = datasetA.join(featuresDF, condition)

uj5u.com熱心網友回復：

您可以加入您的兩個資料框。

最有效的方法是在資料幀A中創建一些列，以便僅使用列相等條件作為連接條件，這將防止 Spark 在連接兩個資料幀時回退到非常低效的笛卡爾積。您可以執行以下操作：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.Arrays;

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.concat_ws;
import static org.apache.spark.sql.functions.element_at;
import static org.apache.spark.sql.functions.split;

import scala.collection.JavaConverters;

...

Dataset<Row> outputDataframe = dataframeA
  .withColumn("FullName", concat_ws(" ", col("FirstName"), col("LastName")))
  .withColumn("Street", element_at(split(col("Address"), ", "), -1))
  .join(dataframeB, JavaConverters.asScalaBuffer(Arrays.asList("Street", "FullName", "BirthDate")), "left_outer")
  .drop("Street", "FullName");

使用您的示例資料框A：

 ---- --------- -------- ---------- ----------------------------------------- 
|A_ID|FirstName|lastName|BirthDate |Address                                  |
 ---- --------- -------- ---------- ----------------------------------------- 
|0   |Vera     |Williams|12.03.1999|Colorado, Greeley, 3774 Stark Hollow Road|
|1   |Joseph   |Peters  |11.10.1988|Florida, Deltona, 4577 Willis Avenue     |
 ---- --------- -------- ---------- ----------------------------------------- 

和資料框B：

 ---- ---------------- ---------- ------------------ 
|B_ID|FullName        |BirthDate |Street            |
 ---- ---------------- ---------- ------------------ 
|37  |Joseph Peters   |11.10.1988|4577 Willis Avenue|
|49  |Valerie J Porter|17.01.2000|2114 Center Street|
 ---- ---------------- ---------- ------------------ 

您將獲得以下output資料框：

 ---------- ---- --------- -------- ----------------------------------------- ---- 
|BirthDate |A_ID|FirstName|lastName|Address                                  |B_ID|
 ---------- ---- --------- -------- ----------------------------------------- ---- 
|12.03.1999|0   |Vera     |Williams|Colorado, Greeley, 3774 Stark Hollow Road|null|
|11.10.1988|1   |Joseph   |Peters  |Florida, Deltona, 4577 Willis Avenue     |37  |
 ---------- ---- --------- -------- ----------------------------------------- ---- 

注意：如果您無法輕松地從 dataframe 中提取精確匹配的資料A，您可以使用Egor 的解決方案。但是，您可能會遇到性能問題，因為 Spark 將執行笛卡爾積。

標籤：爪哇 阿帕奇火花 记录链接

上一篇：Spark訪問Row物件值

下一篇：在pyspark中對RDD的元素進行求和和除法