L3: Apache Spark. Введение

План презентации
Apache Spark обзор
Как работает Spark
RDD
Трансформация и действие
Структура задания Spark
©2020 «Неофлекс». Все права защищены
2

©2020 «Неофлекс». Все права защищены
3
Краткая история Spark

©2020 «Неофлекс». Все права защищены
3
Что такое Apache Spark
Apache Spark – это BigData фреймворк с открытым исходным кодом для распределённой пакетной и потоковой обработки неструктурированных и слабоструктурированных данных, входящий в экосистему проектов Hadoop.
Основным автором Apache Spark считается Матей Захария (Matei Zaharia), румынско-канадский учёный в области информатики. Он начал работу над проектом в 2009 году, будучи аспирантом Университета Калифорнии в Беркли. В 2010 году проект опубликован под лицензией BSD, в 2013 году передан фонду Apache Software Foundation и переведён на лицензию Apache 2.0, а в 2014 году принят в число проектов верхнего уровня Apache. Изначально Спарк написан на Scala.

©2020 «Неофлекс». Все права защищены
3
Преимущества и особенности Apache Spark
Spark — всё-в-одном для работы с большими данными - Spark создан для того, чтобы помогать решать широкий круг задач по анализу данных, начиная с простой загрузки данных и SQL-запросов и заканчивая машинным обучением и потоковыми вычислениями, при помощи одного и того же вычислительного инструмента с неизменным набором API.
Spark оптимизирует своё машинное ядро для эффективных вычислений — то есть Spark только управляет загрузкой данных из систем хранения и производит вычисления над ними, но сам не является конечным постоянным хранилищем.
Библиотеки Spark дарят очень широкую функциональность — сегодня стандартные библиотеки Spark являются главной частью этого проекта с открытым кодом. Ядро Spark само по себе не слишком сильно изменялось с тех пор, как было выпущено, а вот библиотеки росли, чтобы добавлять ещё больше функциональности. И так Spark превратился в мультифункциональный инструмент анализа данных. В Spark есть библиотеки для SQL и структурированных данных (Spark SQL), машинного обучения (MLlib), потоковой обработки (Spark Streaming и аналитики графов (GraphX).
Поддержка нескольких языков разработки - Scala, Java, Python и R

©2020 «Неофлекс». Все права защищены
3
Преимущества и особенности Apache Spark

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark
Преимущество Spark особенно проявляется если необходимо выполнить цепочку задач или итераций

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark
Меньше шагов – Spark job это набор трансформаций (без разделения Mapper - Reducer) разделенных Shuffle.

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark
Жизненный цикл процессов
MapReduce – каждый шаг запускает и удаляет процессы Mapper и Reducer
Spark – каждый Executor (исполнитель) является долгоживущим процессом и может в течение жизни исполнять одну или несколько задач последовательно и параллельно (executor cores)

©2020 «Неофлекс». Все права защищены
3
Развитие MapReduce - Tez
Tez – позволяет запустить цепочку MR на выполнение без промежуточной записи в HDFS

©2020 «Неофлекс». Все права защищены
3
MapReduce: word count
Необходимо написать Mapper и Reducer все остальное обеспечивает MapReduce фреймворк

©2020 «Неофлекс». Все права защищены
3
MapReduce и Spark: упрощение разработки
MapReduce Java
val sc = new SparkContext(new SparkConf().setAppName("Spark Count"))
val textFile = sc.textFile("hdfs://...")
val counts = textFile.flatMap(line => line.split(" "))
.map(word => (word, 1))
.reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")
Word Count для Spark на Scala

Особенности Spark
Для каждого набора данных Spark ведет Lineage и может пересчитать данные с любого момента при сбое/потере узла
Устойчивость к сбоям
Реальная работа начинается только тогда, когда требуются данные (сохранение файл, count, collect, …)
Lazy Evaluation (ленивое исполнение)
Возможность как Batch так и Streaming обработки данных
Потоковая обработка в реальном времени
Spark – eng. «ИСКРА»
Различные архитектурные решения для увеличения скорости (кеширование, долгоживущие executors, …)
Скорость работы
Универсальный фреймворк для разработки широкого спектра задач: batch, streaming, ML, GraphX, SparkSQL. Возможность разработки своих модулей
Универсальность
Scala, Java, Python, R
Поддержка нескольких языков
©2020 «Неофлекс». Все права защищены
6

Основные концепции Spark
©2020 «Неофлекс». Все права защищены
11

©2020 «Неофлекс». Все права защищены
3
RDD
На самом деле внутри
это набор партиций…
Работаем с коллекцией
как с единым целым
… распределенных на рабочих узлах
(в памяти, в кеше, на диске, может и не существовать физически)
val textFile = sc.textFile("hdfs://...")

©2020 «Неофлекс». Все права защищены
3
RDD
RDD - Resilient Distributed Dataset:
Неизменяемая распределенная коллекция (таблица)
Отказоустойчивая - для RDD ведется Lineage – Spark всегда знает как восстановить RDD в случае сбоя
Внутри RDD разбита на партиции — это минимальный объем RDD, который будет обработан каждым рабочим узлом.
RDD распределена по узлам Executors

©2020 «Неофлекс». Все права защищены
3
Трансформация и действие
val textFile = sc.textFile("hdfs://...")
val splits = textFile.flatMap(line => line.split(" "))
val allwords = splits.count()
val tb = splits.filter(_.startsWith("b")))
tb.saveAsTextFile("hdfs://...")
val words = splits.map(word => (word, 1))
val counts = words.reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")
RDD: textFile
RDD: splits
RDD: words
RDD: tb
RDD: counts
Запись в файл
Запись в файл
Подсчет к-ва
Трансформация
не приводит к запуску вычислений
Действие
запускает цепочку вычислений
Lineage
Lazy Evaluation

©2020 «Неофлекс». Все права защищены
3
Трансформация и действие
Трансформация
не приводит к запуску вычислений
Действие
запускает цепочку вычислений
Примеры:
map(func)
filter(func)
union(otherDataset)
reduceByKey(func)
join(otherDataset)
Примеры:
collect()
count()
take(n)
saveAsTextFile(path)

Плюсы и минусы Lazy Evaluation
Улучшает читаемость кода, можно разбивать на небольшие куски, потом все соберется в единый DAG.
Удобство написания программ
Обрабатываются только те данные, которые реально нужны. (take(10))
Строится единый план выполнения.
Избежание ненужных вычислений и трафика между драйвером и узлами
Построенный план запроса оптимизируется Spark, сдвигая например некоторые фильтры ближе к началу
Оптимизация
Каждое действие выполняется без оглядки на другое. Необходимо заботится об избежании повторных вычислений.
cache(), persist()
Необходимо заботиться о повторном вычислении
Особенно в итерационных алгоритмах. Здесь может помочь savepoint(), который сохраняет данные на диск и очищает lineage.
Разрастание плана запросов
©2020 «Неофлекс». Все права защищены
6

©2020 «Неофлекс». Все права защищены
3
Lazy Evaluation кэширование
RDD: textFile
RDD: splits
Подсчет к-ва
RDD: textFile
RDD: splits
RDD: tb
Запись в файл
RDD: textFile
RDD: splits
RDD: words
RDD: counts
Запись в файл
Job 1
Job 2
Job 3
Повторное вычисление и чтение из файла

©2020 «Неофлекс». Все права защищены
3
Lazy Evaluation кэширование
RDD: textFile
RDD: splits
Подсчет к-ва
RDD: textFile
RDD: splits
RDD: t10
Запись в файл
RDD: textFile
RDD: splits
RDD: words
RDD: counts
Запись в файл
Job 1
Job 2
Job 3
val textFile = sc.textFile("hdfs://...")
val splits = textFile.flatMap(line => line.split(" ")).cache()
val allwords = splits.count()
val tb = splits.filter(_.startsWith("b"))
tb.saveAsTextFile("hdfs://...")
val words = splits.map(word => (word, 1))
val counts = words.reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")
Кэш в памяти и/или на локальном диске на узле обработки
Повторное вычисление ТОЛЬКО в случае сбоя или если кэша партиции на узле не осталось
Кэширование позволяет избежать повторного вычисления ветки графа
Иногда кэширование может занять много памяти и времени и быстрее будет повторно произвести вычисления

Как устроено приложение Spark
©2020 «Неофлекс». Все права защищены
11

©2020 «Неофлекс». Все права защищены
3
Приложение Spark
executors
cores
Каждая задача получает для выполнения:
num_executors – к-во отдельных процессов JVM, в которых будут запущена потоки обработки данных(они могут быть расположены как на одном узле, так и на разных). Процессы будут работать до конца работы приложения.
executor_cores – к-во параллельных потоков выполняемых в каждом executor. Обработка данных идет в потоках.
executor-memory – к-во памяти выделяемое каждому Executor
driver-memory – к-во памяти выделяемое драйверу.

©2020 «Неофлекс». Все права защищены
3
Приложение Spark
Для каждого действия строится DAG выполнения
DAG отправляется в DAGScheduler
DAGScheduler разбивает его на этапы (stages) и отправляет на выполнение на TaskScheduler
TaskScheduler использует менеджер кластера (Yarn,Mesos, Spark Standalone) для выделения ресурсов
Каждый Executor получает от Driver задание (Tasks) и выполняет его над своей порцией данных
Данные отсылаются на Driver или сохраняются в файл или кэшируются в памяти Executor

©2020 «Неофлекс». Все права защищены
3
Приложение Spark
Этап это последовательность трансформаций разделенных Shuffle

Звучит интересно, хочу попробовать !!!
©2020 «Неофлекс». Все права защищены
11

©2020 «Неофлекс». Все права защищены
3
Как можно попробовать Spark
3. Распаковать архив в любую папку
4. Установить переменную окружения SPARK_HOME на эту папку
…
1. Установить Java и Python, если будете работать в PySpark
2. Скачать Spark: https://spark.apache.org/downloads.html

©2020 «Неофлекс». Все права защищены
3
Как можно попробовать Spark
5. Запустить:
bin/spark-shell - интерпретатор Scala
или
bin/pyspark – интерпретатор Python
:q для выхода ☺

©2020 «Неофлекс». Все права защищены
3
Как можно попробовать Spark
6. Во время работы интерпретатора будет доступен Spark History Server http://localhost:4040/, где можно изучить как работает приложение Spark