Каков наилучший способ «кэшировать» наборы данных Spark между запросами?

Question

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

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

Моя программа, приведенная ниже, успешно возвращает результатына 2019 год. Следующий абонент также звонит в города 2019 года: Spark перезапускается против всей работы, которую он делал раньше ...

Какой здесь принцип оптимизации?

Должен ли я хранить в своей программе на том же уровне, где я храню сеанс искры, который я использую для запросов и сборки, что-то вроде Map<Integer, Dataset>, где ключ - это год, а набор данных - тот, который вхотя бы один звонивший спросил за этот год?

Answer 1

Вам нужно будет сохранить набор данных в формате hdf или в любом другом используемом хранилище и загрузить его всякий раз, когда потребуется, вместо повторного повторного вычисления всего набора данных. Это больше о том, как вы будете разрабатывать свое приложение. Вероятно, эти наборы данных должны быть предварительно рассчитаны для определенных последних лет как часть подготовки данных и всегда готовы к использованию. Предполагается, что при следующем запуске оно будет запущено как новое задание, например: задание выполняется один раз в день

Answer 2

Redis - лучший выбор для использования с искрой. Сохраните результаты в Redis и для следующего запроса просто возьмите Redis.

Answer 3

Предполагается, что программа spark-shell или программа, скомпилированная с помощью spark, которая выполняется в одном и том же сеансе, получает запросы:

1. Использовать IGNITE или
2. Положиться на «пропущенные этапы»эффект (также используя .cache для DF).

Последние, например, против RDD, но DF имеют следующие базовые значения:

val d = sc.parallelize(0 until 100000).map(i => (i%10000, i)).cache // or not cached, does not matter for RDD, for DF, DS it does

val c=d.rightOuterJoin(d.reduceByKey(_+_))
val f=d.leftOuterJoin(d.reduceByKey(_+_))

c.count
c.collect // skipped, shuffled 
f.count
f.collect // skipped, shuffled

val g = f.filter(e => e._1%2==0) 
val h = f.filter(e => e._1==657)
val j = f.filter(e => e._1==1657)

g.collect 
h.collect 
j.collect  // these skipped as well

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

Обратите внимание на пропущенные этапы из Spark UI, поэтому не всегда все так плохо, как может показаться. В некоторых случаях ваше «кеширование» достигается таким образом.

Для действий, требующих другой обработки, по крайней мере базовые (промежуточные) источники нуждаются в .cache или .persist.

Если используется новая искра-отправка:

1. используйте IGNITE или
2. Повторно используйте каталог контрольных точек, хотя и очень сложный, см. Spark Checkpointing Non-Streaming - файлы контрольных точекможет использоваться в последующем запуске задания или в программе драйвера , хотя и запутанной, и действительно применимой только в том случае, если возможно несколько действий на том перемешанном СДР, который предварительно читается, иначе эффект не так велик. Или
3. Используйте хороший начальный запрос и сохраните его и перечитайте. См. https://databricks -prod-cloudfront.cloud.databricks.com / public / 4027ec902e239c93eaaa8714f173bcfc / 4861715144695760/2994977456373837/5701837197372837 / latest.html . Особенно удобно при сортировке.