Что такое PySpark и зачем его использовать в Big Data

Apache Spark. PySpark может использоваться для распределенных вычислений на Python в рамках анализа и обработки больших данных (Big Data), а также машинного обучения (Machine Learning).

Apache Spark и PySpark

Apache Spark реализован на языке программирования Scala, который выполняется на JVM (Java Virtual Machine). Чтобы получить функциональность Spark в Python, используется PySpark. Поэтому те, кто не знаком со Scala, но знаком с Python, могут запросто использовать возможности фрейвморка Apache Spark.

PySpark взаимодействует с самим Spark через специальную библиотеку Py4J [1]. Она позволяет программам Python, которые выполняются интерпретатором, динамически обращаться к объектам Java в JVM, транслируя код Scala в JVM. Для большей совместимости PySpark поддерживает парадигму функционального программирования, поскольку:

Язык Scala — функциональный.
Функциональный код намного проще распараллелить.

Таким образом, PySpark позволяет проводить параллельную обработку без необходимости использования каких-либо модулей Python для потоковой или многопроцессорной обработки. Вся сложная коммуникация и синхронизация между потоками, процессами и даже разными CPU обрабатываются в Spark.

Точка входа через SparkContext и SparkConf или через SparkSession

Точкой входа в Spark-приложение для создания DataFrame является SparkSession, в котором определяются параметры конфигурации: название приложения, кластерный менеджер (т.е. каким образом подключиться — локально, к Kubernates или YARN и т.д.), количество выделяемых ядер и памяти. Пример инициализации может выглядеть так:

from pyspark.sql import SparkSession

spark = SparkSession.builder
    .master("local[*]") \
    .appName("Word Count") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

Другой способ инициализации осуществляется через SparkContext и SparkConf. Этот способ использовался до появления Spark 2.0, однако, некоторые разработчики применяют его до сих. Выглядеть это может следующим образом:

from pyspark import SparkConf
from pyspark import  SparkContext
conf = SparkConf().setAppName('appName').setMaster('local[*]')
sc = SparkContext(conf=conf)
spark = SparkSession(sc)

PySpark

PySpark включает модули: SQL, Streaming, ML, MLlib.

Модуль SQL

PySparkSQL служит для создания DataFrame и включает такие классы как:

SparkSession — точка входа для создания DataFrame и использования функций SQL.
DataFrame — распределенный набор данных, сгруппированных в именованные столбцы.
Column — столбец в DataFrame.
Row — строка в DataFrame.
GroupedData — агрегационные методы, возвращаемые DataFrame.groupBy().
DataFrameNaFunctions — методы обработки отсутствующих данных (Nan значения).
DataFrameStatFunctions — методы для статистической обработки данных.
functions — список встроенных функций, доступных для DataFrame.
types — список доступных типов данных.
Window для работы с оконными функциями.

Модуль Streaming

Модуль Streaming служит для доступа к функциональности потоковой передачи и является расширением основного API Spark, которое позволяет Data Scientist’ам обрабатывать данные в режиме реального времени из различных источников, включая (но не ограничиваясь) Kafka, Flume и Amazon Kinesis. Эти обработанные данные могут быть отправлены в файловые системы, базы данных или дэшборды.

В основе Streaming лежит DStream (Discretized Stream), который представляет поток данных, разделенный на небольшие пакеты RDD. Такие пакеты могут интегрироваться с любыми другими компонентами Spark, например, MLlib.

Модули ML и MLlib.

В PySpark есть два похожих модуля для машинного обучения (Machine Learning) — ML и MLlib. Они отличаются только типом построения данных: ML использует DataFrame, а MLlib — RDD. Поскольку DataFrame более удобен в работе, то разработчики Spark рекомендуют использовать именно модуль ML [2].

Модули машинного обучения богаты разными инструментами, а интерфейс схож с другой популярной Python-библиотекой для Machine Learning — Scikit-learn. Перечислим основные инструменты:

Конвейер (pipeline), который составляет стадии моделирования;
Извлечение данных с помощью Binarizer, MinMaxScaler, CoutVectorizer, Word2Vec и других классов, которых насчитывается 51;
Классификация, включая логистическую регрессию (logistic regression), деревья решения (decision trees), случайные леса (random forest) и др. Всего их 22.
Кластеризация из 13 алгоритмов, таких как k-средние (k-means), Латентное размещение Дирихле (LDA);
Регрессия – линейная (linear regression), деревья решения и еще 18 регрессионных алгоритмов.