Spark предоставляет быструю и универсальную платформу для обработки данных. По сравнению с Hadoop Spark ускоряет работу программ в памяти более чем в 100 раз, а на диске – более чем в 10 раз. Spark дает больше возможностей для работы с данными. Его синтаксис не так сложен, чтобы начать погружение, для сравнения приведу пример из Pandas.
Для работы с Spark, нужно создать сессию.
«` spark = SparkSession.builder.getOrCreate() «`
Во время создания сессии, происходит кластеризация.
Pandas «` data = pd.read_csv(‘data.csv’) ««
Spark «« data = spark.read.csv(path=’data.csv’, header=True, sep=’,’) ««
Далее, сгруппируем данные и «сместим» в колонке на одну позицию. В Pandas это делается так:
«` data[group1] = pandas_df.groupby(group2)[group3].shift(-1) «` В Spark «` w = Window().partitionBy(«group2»).orderBy(«group3») data = data.withColumn(«group2», lag(«group2», -1, 0).over(w)) «`
Можно использовать оконную функцию, где partitionBy отвечает за группировку данных, а orderBy сортировка. Функция lag принимает 3 параметра: это колонка, шаг смещения и значения, которые будет на месте шага.
МИТАП «Apache Spark за 2 часа — для нетерпеливых»_20 апреля 2022г
Или для группировки можно использовать обычную функцию groupBy, которая тоже есть в Spark. Разница в том, что с окном каждая строка будет связана с результатом агрегирования, вычисленным для всего окна. Однако при группировке каждая группа будет связана с результатом агрегации в этой группе (группа строк становится только одной строкой).
«` dataframe = spark.range(6).withColumn(«key», ‘id % 2) dataframe.show
windowing = Window.partitionBy(«key») dataframe.withColumn(«sum», sum(col(«id»)).over(windowing).show
dataframe.groupBy(«key»).agg(sum(‘id)).show
К сожалению, некоторых функций может не быть в Spark (например, factorize).
«` labels_start, uniques = pd.factorize(anomaly_time[‘activity_start’]) anomaly_time[‘activity_start_code’] = labels_start «` Spark «« win_func = Window().partitionBy().orderBy(lit(‘ ‘)) data = data.select(‘name_column’).distinct().withColumn(‘name_column’, row_number().over(win_func) — 1) ««
Функция factorize закодирует объект как перечислимый тип или категориальную переменную, или присвоит объекту идентификатор.
«` codes, uniques = pd.factorize([‘b’, ‘b’, ‘a’, ‘c’, ‘b’]) codes array([0, 0, 1, 2, 0]. ) «`
Для выполнения подобного функционала в Spark, берется колонка select (‘name_column’) и выбираются все уникальные значения, с помощью функции distinct. Далее с помощью функции withColumn создается колонка и присваивается номер строки (чтобы начиналось с 0 — я отнимаю 1).
Обработка больших данных при помощи Apache Spark, часть 1
Вывод
Apache Spark это огромная система, с множеством инструментов для разных типов задач от SQL до машинного обучения. В этой статье был показан лишь маленький кусочек от всего Spark, но даже этого хватит, чтобы начать обрабатывать данные.
Источник: newtechaudit.ru
Введение в Apache Spark с примерами и примерами использования
Впервые я услышал о Spark в конце 2013 года, когда заинтересовался Scala – языком, на котором написан Spark. Некоторое время спустя я сделал забавный проект по науке о данных, пытаясь предсказать выживание на Титанике . Это оказалось отличным способом познакомиться с концепциями и программированием Spark. Я настоятельно рекомендую его всем начинающим разработчикам Spark, которые ищут место для начала.
Сегодня Spark внедряют такие крупные игроки, как Amazon, eBay и Yahoo! Многие организации запускают Spark на кластерах с тысячами узлов. Согласно FAQ Spark, самый большой известный кластер имеет более 8000 узлов. Действительно, Spark – это технология, которую стоит отметить и изучить.
Эта статья содержит введение в Spark, включая примеры использования и примеры. Он содержит информацию с веб-сайта Apache Spark, а также книгу Learning Spark – молниеносный анализ больших данных.
Что такое Apache Spark? Введение
Spark – это проект Apache, рекламируемый как «молниеносные кластерные вычисления». Он имеет процветающее сообщество open-source и является самым активным проектом Apache на данный момент.
Spark обеспечивает более быструю и общую платформу обработки данных. Spark позволяет запускать программы в 100 раз быстрее в памяти или в 10 раз быстрее на диске, чем Hadoop. В прошлом году Spark взял на себя Hadoop, завершив конкурс Daytona GraySort объемом 100 ТБ в 3 раза быстрее на одну десятую числа машин, и он также стал самым быстрым движком с открытым исходным кодом для сортировки петабайта.
Spark также позволяет быстрее писать код, поскольку в вашем распоряжении более 80 высокоуровневых операторов. Чтобы продемонстрировать это, давайте посмотрим на «Hello World!» BigData: пример подсчета слов. Написанный на Java для MapReduce, он содержит около 50 строк кода, тогда как в Spark (и Scala) вы можете сделать это так просто:
sparkContext.textFile( «hdfs://. » )
.flatMap(line => line.split( » » ))
.map(word => (word, 1 )).reduceByKey(_ + _)
.saveAsTextFile( «hdfs://. » )
Другим важным аспектом при изучении использования Apache Spark является интерактивная оболочка (REPL), которую она предоставляет «из коробки». Используя REPL, можно проверить результат каждой строки кода без необходимости сначала кодировать и выполнять всю работу. Таким образом, путь к рабочему коду намного короче, и возможен специальный анализ данных.
Дополнительные ключевые функции Spark включают в себя:
- В настоящее время предоставляет API в Scala, Java и Python с поддержкой других языков (например, R)
- Хорошо интегрируется с экосистемой Hadoop и источниками данных (HDFS, Amazon S3, Hive, HBase, Cassandra и т. Д.)
- Может работать на кластерах, управляемых Hadoop YARN или Apache Mesos, а также может работать автономно
Ядро Spark дополняется набором мощных высокоуровневых библиотек, которые можно беспрепятственно использовать в одном приложении. Эти библиотеки в настоящее время включают SparkSQL, Spark Streaming, MLlib (для машинного обучения) и GraphX, каждая из которых более подробно описана в этой статье. Дополнительные библиотеки Spark и расширения в настоящее время находятся в стадии разработки.
Spark Core
Spark Core – базовый движок для крупномасштабной параллельной и распределенной обработки данных. Он отвечает за:
- управление памятью и устранение неисправностей
- планирование, распределение и мониторинг заданий в кластере
- взаимодействуя с системами хранения
Spark представляет концепцию RDD (Resilient Distributed Dataset) , неизменяемого отказоустойчивого распределенного набора объектов, с которым можно работать параллельно. СДР может содержать любой тип объекта и создается путем загрузки внешнего набора данных или распространения коллекции из программы драйвера.
СДР поддерживают два типа операций:
Преобразования в Spark являются «ленивыми», то есть они не сразу вычисляют свои результаты. Вместо этого они просто «запоминают» выполняемую операцию и набор данных (например, файл), для которого должна быть выполнена операция. Преобразования фактически вычисляются только тогда, когда вызывается действие и результат возвращается в программу драйвера. Такая конструкция позволяет Spark работать более эффективно. Например, если большой файл был преобразован различными способами и передан первому действию, Spark обрабатывает и возвращает результат только для первой строки, а не выполняет работу для всего файла.
По умолчанию каждый преобразованный СДР может пересчитываться каждый раз, когда вы выполняете над ним действие. Однако вы также можете сохранить RDD в памяти, используя метод persist или cache, и в этом случае Spark сохранит элементы в кластере для более быстрого доступа при следующем запросе.
SparkSQL
SparkSQL – это компонент Spark, который поддерживает запросы данных через SQL или с помощью языка запросов Hive . Он возник в виде порта Apache Hive для запуска поверх Spark (вместо MapReduce) и теперь интегрирован со стеком Spark. В дополнение к обеспечению поддержки различных источников данных, он позволяет создавать запросы SQL с преобразованиями кода, что приводит к очень мощному инструменту. Ниже приведен пример запроса, совместимого с Hive:
// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
sqlContext.sql( «CREATE TABLE IF NOT EXISTS src (key INT, value STRING)» )
sqlContext.sql( «LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt’ INTO TABLE src» )
// Queries are expressed in HiveQL
sqlContext.sql( «FROM src SELECT key, value» ).collect().foreach(println)
Spark Streaming
Spark Streaming поддерживает обработку потоковых данных практически в реальном времени, таких как файлы журналов рабочего веб-сервера (например, Apache Flume и HDFS / S3), социальные сети, такие как Twitter, и различные очереди сообщений, такие как Kafka. Под капотом Spark Streaming получает потоки входных данных и делит данные на пакеты. Затем они обрабатываются механизмом Spark и генерируют окончательный поток результатов в пакетном режиме, как показано ниже.
Spark Streaming API близко соответствует Spark Core, что облегчает программистам работу в мире как пакетных, так и потоковых данных.
MLlib
MLlib – это библиотека машинного обучения, которая предоставляет различные алгоритмы, предназначенные для масштабирования в кластере, для классификации, регрессии, кластеризации, совместной фильтрации и т. Д. (Дополнительную информацию по этой теме можно найти в статье Toptal об машинном обучении ). Некоторые из этих алгоритмов также работают с потоковыми данными, такими как линейная регрессия с использованием обычных наименьших квадратов или кластеризация с помощью k-средних (и многое другое). Apache Mahout (библиотека машинного обучения для Hadoop) уже отвернулась от MapReduce и объединила усилия на Spark MLlib.
Graphx
GraphX – это библиотека для управления графами и выполнения параллельных графов операций. Он предоставляет единый инструмент для ETL, исследовательского анализа и итерационных вычислений графа. Помимо встроенных операций для манипулирования графами, он предоставляет библиотеку общих алгоритмов графов, таких как PageRank.
Как использовать Apache Spark: сценарий использования обнаружения событий
Теперь, когда мы ответили на вопрос «Что такое Apache Spark?», Давайте подумаем о том, какие проблемы или проблемы могут быть использованы наиболее эффективно.
Недавно я наткнулся на статью об эксперименте по обнаружению землетрясения путем анализа потока в Твиттере . Интересно, что было показано, что этот метод, скорее всего, будет информировать вас о землетрясении в Японии быстрее, чем Японское метеорологическое агентство. Несмотря на то, что они использовали разные технологии в своей статье, я думаю, что это отличный пример того, как мы могли бы использовать Spark с упрощенными фрагментами кода и без связующего кода.
Во-первых, нам нужно будет отфильтровать твиты, которые кажутся актуальными, например, «землетрясение» или «сотрясение». Мы могли бы легко использовать Spark Streaming для этой цели следующим образом:
TwitterUtils.createStream(. )
.filter(_.getText.contains( «earthquake» ) || _.getText.contains( «shaking» ))
Затем нам нужно будет выполнить некоторый семантический анализ твитов, чтобы определить, ссылаются ли они на текущее землетрясение. Твиты типа «Землетрясение!» или «Сейчас трясется», например, можно было бы рассмотреть положительные совпадения, тогда как твиты, такие как «Участие в конференции по землетрясению» или «Вчерашнее землетрясение было страшным», не будут. Для этой цели авторы статьи использовали машину опорных векторов (SVM). Мы сделаем то же самое здесь, но также можем попробовать потоковую версию . Результирующий пример кода из MLlib будет выглядеть следующим образом:
// We would prepare some earthquake tweet data and load it in LIBSVM format.
val data = MLUtils.loadLibSVMFile(sc, «sample_earthquate_tweets.txt» )
// Split data into training (60%) and test (40%).
val splits = data.randomSplit(Array( 0.6 , 0.4 ), seed = 11L)
val training = splits( 0 ).cache()
val test = splits( 1 )
// Run training algorithm to build the model
val numIterations = 100
val model = SVMWithSGD.train(training, numIterations)
// Clear the default threshold.
model.clearThreshold()
// Compute raw scores on the test set.
val scoreAndLabels = test.map < point =>
val score = model.predict(point.features)
(score, point.label)
// Get evaluation metrics.
val metrics = new BinaryClassificationMetrics(scoreAndLabels)
val auROC = metrics.areaUnderROC()
println( «Area under ROC java plain»>+ auROC)
Если мы довольны скоростью предсказания модели, мы можем перейти к следующему этапу и реагировать всякий раз, когда обнаруживаем землетрясение. Чтобы обнаружить его, нам нужно определенное количество (то есть, плотность) положительных твитов в определенном временном окне (как описано в статье). Обратите внимание, что для твитов с включенными службами определения местоположения в Твиттере мы также извлекли бы местоположение землетрясения. Вооружившись этими знаниями, мы могли бы использовать SparkSQL и запросить существующую таблицу Hive (хранящую пользователей, заинтересованных в получении уведомлений о землетрясениях), чтобы получить их адреса электронной почты и отправить им персональное предупреждение по электронной почте следующим образом:
// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
// sendEmail is a custom function
sqlContext.sql( «FROM earthquake_warning_users SELECT firstName, lastName, city, email» )
.collect().foreach(sendEmail)
Другие случаи использования Apache Spark
Потенциальные варианты использования Spark, конечно, выходят далеко за рамки обнаружения землетрясений.
Вот быстрая (но, конечно, далеко не исчерпывающая!) Выборка из других вариантов использования, которые требуют учета скорости, разнообразия и объема больших данных, для которых Spark так хорошо подходит:
В игровой индустрии обработка и обнаружение паттернов из потенциального источника событий в игре в реальном времени и возможность мгновенно реагировать на них – это возможность, которая может принести прибыльный бизнес в таких целях, как удержание игрока, целевая реклама, авто -регулировка уровня сложности и так далее.
В индустрии электронной коммерции информация о транзакциях в реальном времени может передаваться алгоритму потоковой кластеризации, например, k-means или совместной фильтрации, такой как ALS . Затем результаты можно даже объединить с другими неструктурированными источниками данных, такими как комментарии клиентов или обзоры продуктов, и использовать их для постоянного улучшения и адаптации рекомендаций с течением времени с учетом новых тенденций.
В сфере финансов или безопасности стек Spark может применяться к системе обнаружения мошенничества или вторжений или аутентификации на основе рисков. Он может достичь первоклассных результатов, собирая огромное количество архивных журналов, комбинируя его с внешними источниками данных, такими как информация о взломах данных и взломанных учетных записях (см., Например, https://haveibeenpwned.com/ ) и информация из соединения / запрос, такой как IP-геолокация или время.
Вывод
Подводя итог, можно сказать, что Spark помогает упростить сложную и требующую большого объема вычислений задачу обработки больших объемов данных в реальном времени или архивированных данных, как структурированных, так и неструктурированных, бесшовно интегрируя соответствующие сложные возможности, такие как машинное обучение и алгоритмы графов. Spark приносит обработку больших данных в массы. Проверьте это!
- Преобразования – это операции (такие как отображение, фильтрация, объединение, объединение и т. Д.), Которые выполняются на СДР и дают новый СДР, содержащий результат.
- Действия – это операции (такие как уменьшение, подсчет, сначала и т. Д.), Которые возвращают значение после выполнения вычисления в СДР.
| Ссылка: | Введение в Apache Spark с примерами и примерами использования от нашего партнера JCG Радека Островского в блоге Mapr . |
Источник: coderlessons.com
Apache Spark установка, настройка, запуск скриптов
2. Установка и настройка Apache Spark.
3. Запуск распределенных вычислений в Apache Spark (на примере скрипта python).
Что такое Apache Spark.
Apache Spark – это универсальная и высокопроизводительная кластерная вычислительная платформа.
Фреймворк создан для того, чтобы охватить более широкий диапазон рабочих нагрузок, которые прежде требовали создания отдельных распределенных систем, включая приложения пакетной обработки, циклические алгоритмы, интерактивные запросы и потоковую обработку. Поддерживая все эти виды задач с помощью единого механизма, Spark упрощает и удешевляет объединение разных видов обработки, которые часто необходимо выполнять в едином конвейере обработки данных. Кроме того, он уменьшает бремя обслуживания, поддерживая отдельные инструменты.
Spark предоставляет простой API на языках Python, Java, Scala и SQL и богатую коллекцию встроенных библиотек. Он также легко объединяется с другими инструментами обработки больших данных. В частности, Spark может выполняться под управлением кластеров Hadoop и использовать любые источники данных Hadoop, включая Cassandra.
Внутренняя реализация Spark обеспечивает эффективное масштабирование от одного до многих тысяч вычислительных узлов. Для достижения такой гибкости Spark поддерживает большое многообразие диспетчеров кластеров (cluster managers), включая Hadoop YARN, Apache Mesos, а также простой диспетчер кластера, входящий в состав Spark, который называется Standalone Scheduler. При установке Spark на чистое множество машин на начальном этапе с успехом можно использовать Standalone Scheduler . При установке Spark на уже имеющийся кластер Hadoop YARN или Mesos можно пользоваться встроенными диспетчерами этих кластеров.
Проще говоря, Spark — это отдельный программный продукт, который может эффективно взаимодействовать с такими компонентами Hadoop как: YARN, HDFS, Hive, HBase.
Установка и настройка Apache Spark.
Скачаем архив с официального сайта . Тут необходимо выбрать версию Spark, я скачал версию spark-2.4.4.
После скачивания архива его нужно разархивировать и перенести извлеченный каталог в /usr/local с помощью команды sudo mv /откуда /куда.
Теперь необходимо настроить переменные окружения для этого запустим редактор (у меня gedit, можно vim или nano):
sudo gedit .bashrc
В конец добавить следующие строки с переменными:
export SPARK_HOME=/usr/local/spark-2.4.4
export PATH=$PATH:$SPARK_HOME/bin
export PYSPARK_PYTHON=/usr/bin/python3
export PYSPARK_DRIVER_PYTHON=python3
Выйдем из редактора и применим настройки:
source .bashrc
Теперь необходимо настроить файл spark-env.sh
Перейти в каталог настроек:
cd $SPARK_HOME/conf
Создать файл spark-env.sh из шаблона (template):
sudo cp spark-env.sh.template spark-env.sh
Открыть spark-env.sh в редакторе:
sudo gedit spark-env.sh
И добавить в конец строки:
SPARK_MASTER_HOST=master
export PYSPARK_PYTHON=/usr/bin/python3
export PYSPARK_DRIVER_PYTHON=python3
Сохранить изменения и выйти.
Создать файл slaves из шаблона (template):
sudo cp slaves.template slaves
Открыть slaves в редакторе:
sudo gedit slaves
Прописать здесь IP-адреса (или имена как у меня) всех подчиненных машин кластера(worker в spark) мастер также может выступать в роли worker-а (как в моем примере).
Сохранить изменения и выйти.
Все практически настроено, осталось только скопировать наши настройки (весь каталог spark) на подчиненные машины.
Сначала на всех подчиненных узлах выпонить:
— создать каталог Spark:
sudo mkdir /usr/local/spark-2.4.4
Назначить владельцем каталога пользователя (в нашем примере hduser):
sudo chown hduser:hadoop -R /usr/local/spark-2.4.4
Теперь можно синхронизировать каталоги, я пользуюсь утилитой rsync но можно и scp:
и на вторую машину:
Все готово к запуску, перейдем в каталог:
cd $SPARK_HOME/sbin
запустим spark на всех подчиненных узлах с помощью скрипта:
При удачном запуске увидим следующее:
На мастере, порт 8080, доступен веб-интерфейс:
Также доступна командная строка pyspark. Она запускается из каталога $SPARK_HOME/bin:
При удачном запуске увидим следующее:
Все готово к распределенной обработке, приступим!
Запуск распределенных вычислений в Apache Spark (на примере скрипта python).
В этом разделе запустим скрипт python для подсчета слов в произведениях нашего любимого Уильяма Шекспира (в моем примере файл william.txt).
Для этого нам понадобится любой текстовый файл, который нужно загрузить в HDFS из локальной директории (в примере /home/hduser/william.txt):
Создадим входной каталог HDFS: hadoop fs -mkdir /input
Загружаем исследуемый файл в HDFS: hadoop fs -put /home/hduser/william.txt /input
Проверим, загрузился ли файл командой: hadoop fs -ls /input
Далее нам нужен сам скрипт wcount.py
Аккуратнее с форматированием файла! В синтаксисе python отступы важны!
Обратить внимание на следующие строки:
f = sc.textFile(«hdfs://Master:9000/input/*.txt») — здесь указан входной каталог, где хранится наш текстовый файл.
counts.saveAsTextFile(«/home/hduser/res/») — выходной каталог результата в локальной директории, каталог res не должен существовать, его создает сама программа, в противном случае ошибка.
Сам скрипт я создал в корневом каталоге spark, создав в нем файл и скопировав из браузера код:
sudo gedit $SPARK_HOME/wcount.py
Запуск скриптов осуществляется с помощью утилиты spark-submit в каталоге $SPARK_HOME/bin
spark-submit —master spark://Master:7077 /usr/local/spark-2.4.4/wcount.py
Вместо имени хоста-мастера может быть его IP-адрес.
Все готово! Запускаем!
Процесс выполнения подробно логгируется, что наводит на мысли о серьезности производимых вычислений!)))
Итог при успешном завершении должен быть такой:
Задача выполнена, результат в каталоге /home/hduser/res/ файл part-00000! В нем приведены встретившиеся слова и напротив — их частота встречаемости.
Логи выполненной задачи также доступны в веб-интерфейсе:
Вот как-то так, вопросы и критика принимаются в группе в контакте, или прямо здесь, в комментариях, доступных после регистрации.
Комментарии
Комментировать могуть только зарегистрированные пользователи
Источник: xn—57-qdd4aqo.xn--p1ai