Spark обновление кэшированного набора данных - PullRequest
Новая
       12

Spark обновление кэшированного набора данных

0 голосов
/ 04 мая 2018

У меня есть отсортированный набор данных, который обновляется (фильтруется) внутри цикла в соответствии со значением заголовка набора данных.

Если я кэширую набор данных каждые n (например, 50) циклов, я получаю хорошую производительность.

Однако после определенного количества циклов кеш, похоже, не работает, так как программа тормозит (наверное, потому, что память, выделенная для кеширования, заполнена).

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

dataset = dataset.sort(/* sort condition */)
dataset.cache()
var head = dataset.head(1)
var count = 0
while (head.nonEmpty) {
  count +=1
  /* custom operation with the head */
  dataset = dataset.filter(/* filter condition based on the head of the dataset */
  if (count % 50 == 0) {
    dataset.cache()
  }
  head = dataset.head(1)
}

Ответы [ 2 ]

0 голосов
/ 04 мая 2018

Попробуйте кэшировать набор данных, прежде чем кэшировать таким образом, вы удалите старую копию набора данных из памяти и сохраните только самую последнюю, избегая нескольких копий в памяти. Ниже приведен пример, но вы должны держать dataset.unpersist () в правильном месте, основываясь на вашей логике кода 

  if (count % 50 == 0) {
    dataset.cache()
  }

  dataset = dataset.filter(/* filter condition based on the head of the dataset */

  if (count % 50 == 0) {
    dataset.cache()
  }
0 голосов
/ 04 мая 2018

cache один не поможет вам здесь. С каждой итерацией происхождение и план выполнения растут, и это не то, что можно решить с помощью одного только кэширования.

Вы должны хотя бы разорвать родословную: 

if (count % 50 == 0) {
  dataset.cache()
  dataset.checkpoint
}

хотя лично я бы также записал данные в распределенное хранилище и прочитал их обратно: 

if (count % 50 == 0) {
  dataset.write.parquet(s"/some/path/$count")
  dataset = spark.read.parquet(s"/some/path/$count")
}

это может быть неприемлемо в зависимости от вашего развертывания, но во многих случаях ведет себя более предсказуемо, чем кэширование и контрольные точки

...