Перейти на PRO

Что такое UDF в Spark?

Тип вопроса: Apache Spark Вероятность: 33% Категория: Data Инженер

Ответ

UDF (User-Defined Function) в Apache Spark — это пользовательская функция, которую можно зарегистрировать и использовать в SQL-запросах или DataFrame API для выполнения операций, не предусмотренных встроенными функциями Spark.

Зачем нужны UDF? Для реализации сложной, специфичной для предметной области логики, которую нельзя выразить стандартными функциями Spark (например, парсинг специализированного формата строки, сложные математические расчеты).

Типы UDF в Spark:

  1. Обычные UDF (Scalar UDF): Принимают одну строку входных данных и возвращают одно значение.

    from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Определяем функцию на Python
def reverse_string(s):
    return s[::-1]

# Создаем и регистрируем UDF
reverse_udf = udf(reverse_string, StringType())

# Использование в DataFrame API
df.withColumn('reversed_name', reverse_udf(df['name'])).show()

# Регистрация для использования в Spark SQL
spark.udf.register("sql_reverse", reverse_string, StringType())
spark.sql("SELECT sql_reverse(name) FROM users").show()

  2. Pandas UDF (Vectorized UDF): Используют Apache Arrow для пакетной передачи данных между JVM и Python, что значительно повышает производительность по сравнению с обычными UDF за счет векторных операций.

    import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

@pandas_udf(DoubleType())
def multiply_by_two(series: pd.Series) -> pd.Series:
    # Операция над целым столбцом (Series) Pandas
    return series * 2

df.withColumn('doubled_value', multiply_by_two(df['value'])).show()

Важные ограничения:

  • Производительность: Обычные (невекторизованные) Python UDF выполняются медленнее, чем встроенные функции Spark (написанные на Scala/JVM), так как данные сериализуются, передаются в Python-процесс и десериализуются.
  • Оптимизация: Catalyst Optimizer не может "заглянуть" внутрь UDF, поэтому некоторые оптимизации (предикат pushdown) могут не работать.

Рекомендация: Всегда сначала ищите решение через встроенные функции Spark. Используйте Pandas UDF вместо обычных, где это возможно. Для сложной логики, критичной к производительности, рассмотрите написание собственных агрегаторов (Aggregator) или использование Scala.

Ответ 18+ 🔞

А, ну слушай, про UDF в Spark. Это, блядь, такая штука, когда встроенных функций тебе уже не хватает, и ты такой: «ёпта, а вот эту мою хитрожопую логику, которую я на коленке написал, как впихнуть в запрос?». Вот для этого они и есть.

Зачем вообще это нужно, эта UDF? Представь, что встроенные функции — это стандартный набор инструментов: молоток, отвёртка, гаечный ключ. А UDF — это когда тебе нужно, блядь, выточить какую-то ебанутую детальку под старинный унитаз, и ты берёшь напильник и делаешь свой, ебать, кастомный инструмент. Например, у тебя строки в каком-то пиздецовом формате, который только твой бизнес-аналитик придумал после трёх бутылок портвейна. Стандартными средствами не распарсишь — вот тут и пишешь свою функцию.

Какие они бывают, эти UDF?

  1. Обычные, скалярные (Scalar UDF): Самые простые, но, блядь, самые медленные. Взял одну строчку — вернул одну строчку. Как будто почтальон Печкин: принёс одну посылку, получил одну расписку.

    from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Пишем свою функцию на Python. Допустим, переворачиваем строку.
def reverse_string(s):
    return s[::-1]  # Типа, хитрая магия

# Оборачиваем это дело в UDF
reverse_udf = udf(reverse_string, StringType())

# И тыкаем в DataFrame
df.withColumn('reversed_name', reverse_udf(df['name'])).show()

# А если хочешь в SQL-запросе юзать — надо зарегистрировать.
spark.udf.register("sql_reverse", reverse_string, StringType())
spark.sql("SELECT sql_reverse(name) FROM users").show()

    Но, чувак, предупреждение сразу: производительность у таких UDF — просто пиздец. Это как ехать на Запорожце через всю страну: доедешь, но овердохуища времени уйдёт. Данные туда-сюда между JVM и Python гоняются, сериализуются — кошмар.

  2. Pandas UDF (Векторизованные): Вот это уже ближе к теме, ядрёна вошь! Тут уже не по одной строчке, а целыми пачками, через Apache Arrow. Быстрее овермного, потому что работа идёт с целыми столбцами как с Pandas Series.

    import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

# Декоратором помечаем
@pandas_udf(DoubleType())
def multiply_by_two(series: pd.Series) -> pd.Series:
    # А тут уже работаем не со строкой, а с целой пачкой данных разом!
    return series * 2

df.withColumn('doubled_value', multiply_by_two(df['value'])).show()

    Вот это уже серьёзно. Если уж писать на Python — то только так.

Главные подводные камни, на которые можно наступить:

  • Скорость: Я уже сказал, но повторюсь: обычные Python UDF — это пиздопроебибна для производительности. Spark начинает кряхтеть как старый дед на лавочке.
  • Оптимизатор: Catalyst, этот умный оптимизатор Spark, в твою UDF заглянуть не может. Для него это чёрный ящик. Поэтому он может не сделать каких-то своих фишек, типа переноса фильтров (predicate pushdown), и запрос будет работать медленнее, чем мог бы.

Так что, главный совет: Сначала, блядь, разберись со встроенными функциями досконально. Очень часто кажется, что своё надо писать, а потом находишь, что всё уже есть. Если своё писать всё же надо — сразу смотри в сторону Pandas UDF. Ну а если логика вообще жёсткая и на скорость, то тут, чувак, надо думать про кастомные агрегаторы (Aggregator) или, совсем красивый вариант, на Scala писать. Но это уже другая история, с блэкджеком и шлюхами.

© ХакСобесов, 2026
support@hacksobesov.com