Apache Spark, универсальная платформа для крупномасштабной обработки данных, в сочетании с Redis способна обеспечить ускоренные расчеты в реальном времени для таких задач, как анализ временных рядов, прогнозы и рекомендации на основе машинного обучения и т. д.

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

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

Redis Labs недавно опубликовала в общий доступ пакет “spark-redis”. Как следует из названия, это коннектор Redis для Apache Spark, который обеспечивает доступ для чтения и записи ко всем основным структурам данных Redis в виде RDD (Resilient Distributed Datasets, в терминологии Spark), что позволяет Spark использовать Redis в качестве одного из источников данных. Этот коннектор предоставляет Spark структуры данных Redis, тем самым обеспечивая значительный прирост производительности для всех типов расчетов. Он также позволяет нам организовать совместный доступ к DataSet/DataFrame/RDD Spark из сразу нескольких разных приложений.

Но прежде чем мы сможем использовать коннектор Spark-Redis, нам сперва нужно позаботиться о наличии нескольких ключевых элементов, а именно: Apache Spark, Scala, Jedis и Redis.

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

spark-redis
jedis

Запись Spark-датасета из одного приложения в Redis с помощью Scala 

Следующий фрагмент кода демонстрирует, как выглядит кэширование Spark-датасета в Redis:

def main(args: Array[String]): Unit = {
val sparkSession = org.apache.spark.sql.SparkSession.builder
    .config(new SparkConf()
      .setMaster("local")
      .setAppName("myApp")
      // начальный хост Redis - может быть любым узлом в кластерном режиме
      .set("redis.host", "localhost")
      // начальный порт redis
      .set("redis.port", "6379")
      // опциональный AUTH-пароль redis
      //.set("redis.auth", "")
    ).getOrCreate;
val path = "/dev/test/test.csv"
  val base_df = sparkSession.read.option("header","true").csv(path)
  // сохраняем датасет в Redis
  base_df.write
    .format("org.apache.spark.sql.redis")
    .option("table", "temp")
    .mode(SaveMode.Overwrite)
    .save()
}

Чтение Spark-датасета из Redis с помощью Scala

Следующий фрагмент кода демонстрирует, как читать кэшированный Spark-датасет из Redis:

def main(args: Array[String]): Unit = {
  val sparkSession = org.apache.spark.sql.SparkSession.builder
    .config(new SparkConf()
      .setMaster("local")
      .setAppName("myApp")
      // начальный хост Redis — может быть любым узлом в кластерном режиме
      .set("redis.host", "localhost")
      // начальный порт redis
      .set("redis.port", "6379")
      // опциональный AUTH-пароль redis
      //.set("redis.auth", "")
    ).getOrCreate;
// считываем ранее сохраненный датасет из Redis
  val loadedDf = sparkSession.read
    .format("org.apache.spark.sql.redis")
    .option("table", "temp")
    .load()
  loadedDf.printSchema()
  loadedDf.show()  
}

Преимущества использования Redis со Spark

• Redis может повысить производительность Spark до 50 раз для некоторых вариантов использования, таких как, например, spark-timeseries. Коннектор Spark-Redis автоматизирует их, предоставляя Spark API и структуры данных Redis.

На приведенном ниже графике, взятом с сайта redislabs.com, показан прирост производительности при использовании Redis Cache вместе со Spark.

• Redis позволяет Spark совместно использовать датасет разными приложениями.

• Структуры данных Redis позволяют получить доступ к отдельным элементам данных, сводя к минимуму накладные расходы на сериализацию/десериализацию и избегая необходимости передавать туда-сюда большие фрагменты данных.

• Модули Redis, такие как Redis-ML, ускоряют библиотеки Spark ML, позволяя реализовать быстрейшую доставку моделей машинного обучения, которые изначально хранятся в Redis, могут повторно использоваться во многих приложениях и языках, легко развертываются в рабочей среде с уменьшенной в 5–10 раз задержкой выполнения.

Заключение

Redis “утоляет жажду” Spark к данным. Spark-Redis позволяет объединить RDD и основные структуры данных Redis с помощью всего одной строчки кода на Scala. Пакет Spark-Redis предоставляет прямой доступ для чтения/записи ко всем основным структурам данных с параллелизацией RDD и bpzoysq (т. е. на основе SCAN) способ получения имен ключей. Кроме того, коннектор заключает в себе значительный скрытый потенциал, поскольку он фактически поддерживает кластер (Redis) и сопоставляет партиции RDD с хэш-слотами, чтобы уменьшить перетасовку между ядрами (inter-engine shuffling).

Ссылки

Коннектор для Spark, который позволяет читать и записывать в/из кластера Redis — RedisLabs/spark-redis

Какие коннекторы используются Spark из коробки, какие коннекторы подключаются из библиотек, а какие и когда надо писать самому? Ответы на эти вопросы можно получить на открытом уроке «Использование коннекторов для Spark». Регистрация по ссылке.