scala – 一个RDD中的部分/完全匹配值到另一个RDD中的值

1：使用CrossJoin

import spark.implicits._

val df1 = Seq(
  (1,"2017-2-13","ABX-3354 gsfette"),(2,"2017-3-18","TYET-3423 asdsad"),(3,"2017-2-09","TYET-3423 rewriu"),(4,"ABX-335442324"),//changed from "ABX-3354 42324"
  (5,"2017-4-01","aerrTYET-3423") //changed from "TYET-3423 aerr"
).toDF("id","dt","txt")

val df2 = Seq(
  ("mfr1","ABX-3354"),("mfr2","TYET-3423")
).toDF("col1","key")

//match function for filter
def matcher(row: Row): Boolean = row.getAs[String]("txt")
  .contains(row.getAs[String]("key"))

val join = df1.crossJoin(df2)

import org.apache.spark.sql.functions.count

val result = join.filter(matcher _)
  .groupBy("key")
  .agg(count("txt").as("count"))

2：使用Broadcast变量

import spark.implicits._

val df1 = Seq(
  (1,"ABX-3354 42324"),(5,"aerrTYET-3423"),(6,"aerrYET-3423")
).toDF("id","pattern")

//small dataset to broadcast
val df2 = Seq(
  ("mfr1","TYET-3423")
).map(_._2) // considering only 2 values in pair

//Lookup to use in UDF
val lookup = spark.sparkContext.broadcast(df2)

//Udf
import org.apache.spark.sql.functions._
val matcher = udf((txt: String) => {
  val matches: Seq[String] = lookup.value.filter(txt.contains(_))
  if (matches.size > 0) matches.head else null
})

val result = df1.withColumn("match",matcher($"pattern"))
  .filter($"match".isNotNull) // not interested in non matching records
  .groupBy("match")
  .agg(count("pattern").as("count"))

两种解决方案都会产生相同

result.show()

+---------+-----+
|      key|count|
+---------+-----+
|TYET-3423|    3|
| ABX-3354|    2|
+---------+-----+