Отфильтруйте один фрейм данных, используя другой фрейм данных в spark scala

Question

Я собираюсь продемонстрировать свой вопрос, используя следующие два фрейма данных.

val datF1= Seq((1,"everlasting",1.39),(1,"game", 2.7),(1,"life",0.69),(1,"learning",0.69),
                  (2,"living",1.38),(2,"worth",1.38),(2,"life",0.69),(3,"learning",0.69),(3,"never",1.38)).toDF("ID","token","value")
    datF1.show()

+---+-----------+-----+
| ID|      token|value|
+---+-----------+-----+
|  1|everlasting| 1.39|
|  1|       game|  2.7|
|  1|       life| 0.69|
|  1|   learning| 0.69|
|  2|     living| 1.38|
|  2|      worth| 1.38|
|  2|       life| 0.69|
|  3|   learning| 0.69|
|  3|      never| 1.38|
+---+-----------+-----+




val dataF2= Seq(("life ",0.71),("learning",0.75)).toDF("token1","val2")
dataF2.show()
+--------+----+
|  token1|val2|
+--------+----+
|   life |0.71|
|learning|0.75|
+--------+----+

Я хочу отфильтровать ID и value из dataF1 на основе token1 из dataF2. Для каждого слова в token1 из dataF2, если есть маркер слова, тогда value должно быть равно значению dataF1, иначе значение должно быть равно нулю. Другими словами, мой желаемый результат должен быть таким:

    +---+----+----+
| ID| val|val2|
+---+----+----+
|  1|0.69|0.69|
|  2| 0.0|0.69|
|  3|0.69| 0.0|
+---+----+----+

Поскольку обучение не представлено в идентификаторе, равном 2, значение val равно нулю. Точно так же, поскольку жизнь не существует для ID, равного 3, val2 равно нулю.

Я сделал это вручную следующим образом:

val newQ61=datF1.filter($"token"==="learning")

val newQ7 =Seq(1,2,3).toDF("ID")
val newQ81 =newQ7.join(newQ61, Seq("ID"), "left")
val tf2=newQ81.select($"ID" ,when(col("value").isNull ,0).otherwise(col("value")) as "val" )

val newQ62=datF1.filter($"token"==="life")

val newQ71 =Seq(1,2,3).toDF("ID")
val newQ82 =newQ71.join(newQ62, Seq("ID"), "left")
val tf3=newQ82.select($"ID" ,when(col("value").isNull ,0).otherwise(col("value")) as "val2" )

val tf4 =tf2.join(tf3 ,Seq("ID"), "left")
tf4.show()

+---+----+----+
| ID| val|val2|
+---+----+----+
|  1|0.69|0.69|
|  2| 0.0|0.69|
|  3|0.69| 0.0|
+---+----+----+

Вместо того, чтобы делать это вручную, есть ли способ сделать это более эффективно, получая доступ к индексам одного фрейма данных в другом фрейме данных? потому что в реальных ситуациях может быть более 2 слов, поэтому ручной доступ к каждому слову может быть очень трудным делом.

Спасибо

ОБНОВЛЕНИЕ Когда я использую leftsemi присоединиться к моему выводу так:

datF1.join(dataF2, $"token"===$"token1", "leftsemi").show()
+---+--------+-----+
| ID|   token|value|
+---+--------+-----+
|  1|learning| 0.69|
|  3|learning| 0.69|
+---+--------+-----+

Answer 1

Я считаю, что здесь может работать левое внешнее соединение, а затем поворот на token:

 val ans = df1.join(df2, $"token" === $"token1", "LEFT_OUTER")
.filter($"token1".isNotNull)
.select("ID","token","value")
.groupBy("ID")
.pivot("token")
.agg(first("value"))
.na.fill(0)

Результат (без обработки нуля):

ans.show

+---+--------+----+
| ID|learning|life|
+---+--------+----+
|  1|    0.69|0.69|
|  3|    0.69|0.0 |
|  2|    0.0 |0.69|
+---+--------+----+

ОБНОВЛЕНИЕ : как следует из ответа Ламануса, внутреннее соединение, возможно, лучше, чем внешнее соединение + фильтр.

Answer 2

Я думаю, что inner объединения достаточно. Кстати, я нашел опечатку в вашем тестовом примере, что делает результат неправильным.

val dataF1= Seq((1,"everlasting",1.39),
                (1,"game", 2.7),
                (1,"life",0.69),
                (1,"learning",0.69),
                (2,"living",1.38),
                (2,"worth",1.38),
                (2,"life",0.69),
                (3,"learning",0.69),
                (3,"never",1.38)).toDF("ID","token","value")
dataF1.show
// +---+-----------+-----+
// | ID|      token|value|
// +---+-----------+-----+
// |  1|everlasting| 1.39|
// |  1|       game|  2.7|
// |  1|       life| 0.69|
// |  1|   learning| 0.69|
// |  2|     living| 1.38|
// |  2|      worth| 1.38|
// |  2|       life| 0.69|
// |  3|   learning| 0.69|
// |  3|      never| 1.38|
// +---+-----------+-----+

val dataF2= Seq(("life",0.71), // "life " -> "life"
                ("learning",0.75)).toDF("token1","val2")
dataF2.show
// +--------+----+
// |  token1|val2|
// +--------+----+
// |    life|0.71|
// |learning|0.75|
// +--------+----+

val resultDF = dataF1.join(dataF2, $"token" === $"token1", "inner")
resultDF.show
// +---+--------+-----+--------+----+
// | ID|   token|value|  token1|val2|
// +---+--------+-----+--------+----+
// |  1|    life| 0.69|    life|0.71|
// |  1|learning| 0.69|learning|0.75|
// |  2|    life| 0.69|    life|0.71|
// |  3|learning| 0.69|learning|0.75|
// +---+--------+-----+--------+----+

resultDF.groupBy("ID").pivot("token").agg(first("value"))
  .na.fill(0).orderBy("ID").show

Это даст вам такой результат, как

+---+--------+----+
| ID|learning|life|
+---+--------+----+
|  1|    0.69|0.69|
|  2|     0.0|0.69|
|  3|    0.69| 0.0|
+---+--------+----+

Answer 3

Похоже, вам нужно "левое полусоединение". Он будет фильтровать один фрейм данных на основе другого. Попробуйте использовать его как

datF1.join(datF2, $"token"===$"token2", "leftsemi")

Вы можете найти немного больше информации здесь - https://medium.com/datamindedbe/little-known-spark-dataframe-join-types-cc524ea39fd5