Цепочка пользовательских преобразований DataFrame в Spark

Перевод материала подготовлен в рамках набора студентов на онлайн-курс «Экосистема Hadoop, Spark, Hive».

Всех желающих приглашаем на открытый вебинар «Тестирование Spark приложений». На этом открытом уроке рассмотрим проблемы в тестировании Spark приложений: стат данные, частичную проверку и запуск/остановку тяжелых систем. Изучим библиотеки для решения и напишем тесты. Присоединяйтесь!

Для цепочки преобразований DataFrame в Spark можно использовать implicit classes или метод Dataset#transform. В этой статье блога будет продемонстрировано, как выстраивать цепочки преобразований DataFrame, и объяснено, почему метод Dataset#transform предпочтительнее, чем implicit classes.

Структурирование кода Spark в виде преобразований DataFrame отличает сильных программистов Spark от «спагетти-хакеров», как подробно описано в статье «Написание идеального кода Spark (Writing Beautiful Spark Code)». После публикации в блоге, ваш код Spark будет намного проще тестировать и повторно использовать.

Если вы используете PySpark, смотрите эту статью о цепочке пользовательских преобразований PySpark DataFrame.

Метод transform (преобразования) набора данных

Метод transform (преобразования) набора данных предоставляет «краткий синтаксис для цепочки пользовательских преобразований».

Предположим, у нас есть метод withGreeting(), который добавляет столбец приветствия к DataFrame, и метод withFarewell(), который добавляет столбец прощания к DataFrame.

def withGreeting(df: DataFrame): DataFrame = {   df.withColumn("greeting", lit("hello world")) }  def withFarewell(df: DataFrame): DataFrame = {   df.withColumn("farewell", lit("goodbye")) }

Мы можем использовать метод transform (преобразования) для запуска методов withGreeting() и withFarewell().

val df = Seq(   "funny",   "person" ).toDF("something")  val weirdDf = df   .transform(withGreeting)   .transform(withFarewell)

weirdDf.show()  +---------+-----------+--------+ |something|   greeting|farewell| +---------+-----------+--------+ |    funny|hello world| goodbye| |   person|hello world| goodbye| +---------+-----------+--------+

Метод transform (преобразования) можно легко объединить со встроенными методами Spark DataFrame, такими как select.

df   .select("something")   .transform(withGreeting)   .transform(withFarewell)

Если метод transform (преобразования) не используется, то нам придется вложить вызовы методов, и код станет менее читабельным.

withFarewell(withGreeting(df))  // even worse withFarewell(withGreeting(df)).select("something")

Метод transform (преобразования) c аргументами

Пользовательские преобразования DataFrame, использующие аргументы, также могут использовать метод transform (преобразования), используя карринг / списки с несколькими параметрами в Scala.

Давайте воспользуемся тем же методом withGreeting(), что и ранее, и добавим метод withCat(), который принимает в качестве аргумента строку.

def withGreeting(df: DataFrame): DataFrame = {   df.withColumn("greeting", lit("hello world")) }  def withCat(name: String)(df: DataFrame): DataFrame = {   df.withColumn("cats", lit(s"$name meow")) }

Мы можем использовать метод transform (преобразования) для запуска методов withGreeting() и withCat().

val df = Seq(   "funny",   "person" ).toDF("something")  val niceDf = df   .transform(withGreeting)   .transform(withCat("puffy"))

niceDf.show()  +---------+-----------+----------+ |something|   greeting|      cats| +---------+-----------+----------+ |    funny|hello world|puffy meow| |   person|hello world|puffy meow| +---------+-----------+----------+

Метод transform (преобразования) можно использовать для пользовательских преобразований DataFrame, которые также могут использовать аргументы!

Манкипатчинг с помощью неявных классов (Implicit Classes)

Неявные классы можно использовать для добавления методов в существующие классы. Следующий код добавляет те же методы withGreeting() и withFarewell() к самому классу DataFrame.

object BadImplicit {    implicit class DataFrameTransforms(df: DataFrame) {      def withGreeting(): DataFrame = {       df.withColumn("greeting", lit("hello world"))     }      def withFarewell(): DataFrame = {       df.withColumn("farewell", lit("goodbye"))     }    }  }

Методы withGreeting() и withFarewell() можно объединить в цепочку и выполнить следующим образом.

import BadImplicit._  val df = Seq(   "funny",   "person" ).toDF("something")  val hiDf = df.withGreeting().withFarewell()

Расширение основных классов работает, но это плохая программистская практика, которой следует избегать.

Избегание неявных классов

Изменение базовых классов известно как манкипатчинг и является восхитительной особенностью Ruby, но может быть рискованным в неопытных руках.
— Санди Метц

Комментарий Санди был адресован языку программирования Ruby, но тот же принцип применим и к неявным классам Scala.

Манкипатчинг обычно не приветствуется в сообществе Ruby, и его следует избегать в Scala.

Spark был достаточно любезен, чтобы предоставить метод transform (преобразования),  и вам не потребуется манкипатчинг для класса DataFrame. С помощью некоторых приемов программирования на Scala мы даже можем заставить метод transform работать с пользовательскими преобразованиями, которые могут использовать аргументы. Это делает метод transform явным победителем!

Подробнее о курсе: «Экосистема Hadoop, Spark, Hive»

Смотреть демо-урок: «Тестирование Spark приложений»