Блог

У меня очень огромный набор данных, для которого мне нужно выполнить объединение, чтобы обогатить его дополнительными столбцами. Набор данных A содержит структуру:- origin|destination|segment1_origin|segment2_origin|segment3_origin|segment4_origin|segment5_origin|segment6_origin|segment1_destination|segment2_destination|segment3_destination|segment4_destination|segment5_destination|segment6_destination

и содержит около 5 миллиардов строк

Набор данных B содержит структуру:- origin|destination|stops|route

Набор данных B фактически содержит информацию о каждом сегменте в наборе данных A и почти в 6 раз превышает размер набора данных A

Для того, чтобы обогатить детали остановки и маршрута, я сейчас делаю:-

 for (x <- 1 to 6){
    DatasetB.withColumnRenamed("stops", s"segment${x}_stops").withColumnRenamed("route", s"segment${x}_route")
    DatasetA.join(DatasetB, (col(s"segment${x}departure") === col("origin"))
            amp;amp; (col(s"segment${x}Arrival") === col("destination")), "left").drop("origin", "destination")
}
 

И это решение работает нормально. Но меня беспокоит то, что я присоединяюсь к нему 6 раз. Мне просто было любопытно узнать, есть ли какой-нибудь способ оптимизировать это? Это вызывает асимметрию, и на более поздних этапах работа замедляется.

Есть ли способ в Scala / Spark dataframe написать это лучше?

вы можете выполнить spark.sql соединение с or условием, чтобы вам не нужно было их зацикливать. Также, если ваши данные искажены и занимают время на нескольких разделах, вы можете попробовать salting технику, которая увеличит размер данных, но соединения не будут застревать дольше.