Предположим, что нам нужно перераспределить набор данных после фильтрации или для достижения степени параллелизма.
Как мы можем выполнить динамическое перераспределение вместо ручной настройки количества разделов?
Примечание. Ищете решение для RDD, фрейма данных и набора данных.
Вы можете использовать repartition(colname) или partitionBy() для динамического разделения вашего набора данных.
Например, если ваш набор данных выглядит следующим образом
create table sensor_data (
sensor_id bigint,
temp float,
region_id string,
state string,
country string
) partition by ( day string)
Если вы хотите выполнить некоторые расчеты по регионам для определенного дня,
val sensor_data = spark.sql("select * from sensor_data where day='2018-02-10')
val results = sensor_data.
repartition(col("region_id")).
mapPartitions( eventIter => {
processEvent(eventIter).iterator
})
case Event(sensor_id: String, country: String, max_temp: float)
def processEvent(evtIter: Iterator[Row]) : List[Event] = {
val maxTempEvents = ListBuffer[Event]()
while (evtIter.hasNext) {
val evt = evtIter.next()
// do your calculation and add results to maxTempEvents list
}
maxTempEvents
}
Надеюсь это поможет.
Спасибо, Рави.
