Цепочка пользовательских преобразований 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 приложений»