Как эффективно сопоставлять ключи из одного набора данных на основе значений из другого набора данных

Предполагая, что фрейм данных 1 представляет целевую страну и список исходных стран, а фрейм данных 2 представляет доступность для всех стран, найдите все пары из фрейма данных 1, где сопоставление целевой страны ИСТИНА, а сопоставление исходной страны равно ЛОЖЬ:

Dataframe 1 (targetId, sourceId):
США: Китай, Россия, Индия, Япония
Китай: США, Россия, Индия
Россия: США, Япония

Dataframe 2 (идентификатор, доступен):
США: true
Китай: false
Россия: true
Индия: false
Япония: true

Результирующий набор данных должен выглядеть следующим образом:
(США, Китай),
(США, Индия)

Моя идея состоит в том, чтобы сначала взорвать набор данных1, создать новый фрейм данных (скажем, tempDF), добавить к нему 2 новых столбца: targetAvailable, sourceAvailable и, наконец, отфильтровать для targetAvailable = false и sourceAvailable = true, чтобы получить желаемый результирующий фрейм данных.

Ниже приведен фрагмент моего кода:

 val sourceDF = sourceData.toDF("targetId", "sourceId")
 val mappingDF = mappingData.toDF("id", "available")
 val tempDF = sourceDF.select(col("targetId"), 
                explode(col("sourceId")).as("source_id_split"))

 val resultDF = tempDF.select("targetId")
         .withColumn("targetAvailable", isAvailable(tempDF.col("targetId")))
         .withColumn("sourceAvailable", isAvailable(tempDF.col("source_id_split")))


 /*resultDF.select("targetId", "sourceId").
  filter(col("targetAvailable") === "true" and col("sourceAvailable") 
  === "false").show()*/


// udf to find the availability value for the given id from the mapping table
val isAvailable = udf((searchId: String) => {
val rows = mappingDF.select("available")
          .filter(col("id") === searchId).collect()

if (rows(0)(0).toString.equals("true")) "true" else "false"  })

Вызов isAvailable UDF при вычислении resultDF вызывает у меня какое-то странное исключение. Я делаю что-то неправильно? есть ли лучший/более простой способ сделать это?


apache-spark apache-spark-sql scala data-transform apache-spark-dataset
person Uday    schedule 22.03.2020    source источник
comment
Привет Удай, добро пожаловать в SO. Что вы уже пробовали? Не могли бы вы показать нам код, который у вас есть, и вашу текущую проблему? В настоящее время это выглядит не как проблема, с которой вы столкнулись, а как часть вашей домашней работы. Спасибо!   -  person regina_fallangi    schedule 23.03.2020
comment
привет @regina_fallangi: спасибо за указание. Отредактировал описание и добавил свой код.   -  person Uday    schedule 23.03.2020
comment
вызов isAvailable UDF при вычислении результата DF вызывает какое-то странное исключение. -› Не могли бы вы опубликовать странное исключение?   -  person regina_fallangi    schedule 24.03.2020

Ответы (1)


arrow_upward
1
arrow_downward

В вашем UDF вы ссылаетесь на другой кадр данных, что невозможно, поэтому вы получаете «странное» исключение.

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

Однако идея использовать explode очень хороша. Вот способ добиться желаемого:

// generating data, please provide this code next time ;-)
val sourceDF = Seq("USA" ->  Seq("China", "Russia", "India", "Japan"),
                   "China" -> Seq("USA", "Russia", "India"),
                   "Russia" -> Seq("USA", "Japan"))
               .toDF("targetId", "sourceId")
val mappingDF = Seq("USA" -> true, "China" -> false,
                    "Russia" -> true, "India" -> false,
                    "Japan" -> true)
               .toDF("id", "available")

sourceDF
    // we can filter available targets before exploding.
    // let's do it to be more efficient.
    .join(mappingDF.withColumnRenamed("id", "targetId"), Seq("targetId"))
    .where('available)
    // exploding the sources
    .select('targetId, explode('sourceId) as "sourceId")
    // then we keep only non available sources
    .join(mappingDF.withColumnRenamed("id", "sourceId"), Seq("sourceId"))
    .where(! 'available)
    .select("targetId", "sourceId")
    .show(false)

который дает

+--------+--------+
|targetId|sourceId|
+--------+--------+
|USA     |China   |
|USA     |India   |
+--------+--------+
person Oli    schedule 24.03.2020
comment
Круто, это именно то, что я искал. Спасибо большое! - person Uday; 26.03.2020
comment
Оли: здесь мы дважды соединяемся с таблицей сопоставления, один раз, чтобы отфильтровать целевые сопоставления, а затем отфильтровать исходные сопоставления. просто любопытно, есть ли способ решить это, используя только одно соединение? - person Uday; 26.03.2020
comment
Извините, я пропустил ваш комментарий в то время. Это на самом деле зависит. Если оба кадра данных большие, я не вижу никакого решения, потому что, поскольку вы объединяете два разных столбца, вам нужно по-разному перетасовывать данные, чтобы совместить совпадающие ключи. Однако, если один из них мал (достаточно мал, чтобы его можно было транслировать), мы могли бы добиться большего. Так ли это? - person Oli; 11.05.2020