Как применить операцию «Нравится» при объединении нескольких фреймов данных в спарк?

Question

Я пытаюсь объединить два фрейма данных, а затем применить к нему аналогичную операцию.Но это не возвращает никакого значения.Я хочу сделать здесь образец соответствия.Любое предположение, что я делаю здесь неправильно.

import org.apache.spark._
import org.apache.spark.sql.Row

val upcTable = spark.sqlContext.sparkContext.parallelize(Seq(
  Row(1, 50, 100),
  Row(2, 60, 200),
  Row(36, 70, 300),
  Row(45, 80, 400)
))

val lookupUpc = spark.sqlContext.sparkContext.parallelize(Seq(
 Row(3, 70, 300),
 Row(4, 80, 400)
))

val upcDf = spark.sqlContext.createDataFrame(upcTable, StructType(Seq(
  StructField("U_ID", StringType, nullable = false),
  StructField("V_ID", IntegerType, nullable = false),
  StructField("R_ID", IntegerType, nullable = false))))

val lookupDf = spark.sqlContext.createDataFrame(lookupUpc, StructType(Seq(
  StructField("U_ID", StringType, nullable = false),
  StructField("V_ID", IntegerType, nullable = false))))
lookupDf.show()

val joinDf = upcDf.join(lookupDf,Seq("V_ID"),"inner").filter(upcDf("U_ID").like("%lookupDf(U_ID)")).select(upcDf("U_ID"),upcDf("V_ID"),upcDf("R_ID")).show()

Здесь я хотел 36 и 45 от upcDf.

Answer 1

Вместо метода столбца , подобного , который ожидает литерал String, метод содержит , который принимает аргумент типа Any (следовательно, также Column) будет более подходящимв вашем случае:

val joinDf = upcDf.join(lookupDf, Seq("V_ID"), "inner").
  where(upcDf("U_ID").contains(lookupDf("U_ID"))).
  select(upcDf("U_ID"), upcDf("V_ID"), upcDf("R_ID"))

joinDf.show
// +----+----+----+
// |U_ID|V_ID|R_ID|
// +----+----+----+
// |  45|  80| 400|
// |  36|  70| 300|
// +----+----+----+

Обратите внимание, что столбец U_ID в образце набора данных должен иметь тип String на основе перечисленных схем.

[ОБНОВЛЕНИЕ]

Согласно уточненному требованию из комментариев, если вы хотите ограничить совпадение только начальным символом, я бы предложил использовать метод regexp_extract и заменить вышеприведенное предложение where следующим:

where(lookupDf("U_ID") === regexp_extract(upcDf("U_ID"), "^(.)", 1))