Ninil 24 апр в 12:34

[Перевод] Почему стоит начать писать собственные Spark Native Functions?

5 мин

1.2K

Scala*Big Data*Hadoop*Data Engineering*

Туториал

Перевод

Комментарии 20

Geckelberryfinn 24 апр в 12:44

Ну, допустим, а можно теперь эту кастомную функцию как-то заппакеджить и дать дата-сайнтистам, которые на PySpark в основном? Особенно, если речь идёт о облачных средах, таких как Databricks или Glue?

Ninil 24 апр в 12:51

Интересный кейс. Спасибо! В рамках своих изысканий попробую.
Пока я хочу использовать подобноное для своих Spark-приложений.

mitgard 24 апр в 14:25

Вполне, вам нужно сделать джарник с таким кодом, а потом его добавить в зависимость spark-приложения, как условный драйвер для БД.

А в коде просто сделать импорт.

Но если мы говорим, про pySpark, скорее всего придется сделать питон враппер или вызывать через jvm

Ninil 24 апр в 14:30

Еще интересный юзкейс для исследования - как сделать доступной эту функцию из Датабриск ноутбука, в котором дата-сайентисты исползуют PySpark

Geckelberryfinn 24 апр в 17:40

Посмотрел, да это может сработать через Jar и врапперы. Как посоветовал mitgard, нужно упаковать Jar, затем его указывать в spark-submit --jars. Враппер будет иметь вид вроде такого

from pyspark.sql.column import Column
from pyspark.sql.functions import expr

def my_catalyst_expression(column):
    return Column(expr(f"MyCatalystExpression({column._jc.toString()})"))

...
df.select(my_catalyst_expression(df.field)).show()

В случае databricks тогда все даже чуть проще, созданный Jar можно загрузить в Libraries у кластера.

druzyk 24 апр в 12:55

Интересно!

sshikov 24 апр в 21:07

Реализация генератора UUID с использованием UDF проста.

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

Но тут еще могут добавляться вопросы с класслоадерами. Скажем, мы как-то попытались создать экземпляр UDF и зарегистрировать его в groovy скрипте, который динамически выполнялся из кода на спарке. Так вот, ничего не получилось, потому что класслоадер оказался другой, и наша функция имела сигнатуру, отличную от нужной. А как устроены класслоадеры в спарке, описано примерно так же, как описано создание catalyst выражений из этой статьи - т.е. примерно никак.

Ninil 24 апр в 21:38

Да, про сериализацию верно подмечено. Плюсую

csl 4 июн в 14:14

Offtop: Spark поддерживает Scala 3?

Part 2: https://habr.com/ru/articles/816997

Ninil 4 июн в 14:38

Нет, официально нет. Но есть способы заставить работать)

sshikov 4 июн в 19:45

Насколько я помню - нет. Но - спарк 3.4 привнес Spark Connect - это API, сделанный с целью поддержать другие языки. Сами авторы ничего кроме стандартных Java Scala и Python c R не поддержали, но написать реализацию вроде можно. Для JVM based Scala 3 - уж наверняка. Т.е. у вас будет Dataset, скажем просто.

Ninil 4 июн в 19:57

Недавно столкнулся и с удивлением узнал о возможности написания Spark-приложений на .NET: https://github.com/dotnet/spark

sshikov 4 июн в 20:13

мне кажется это устареет с широким внедрением Connect. По-моему они даже бросили его поддерживать уже. Да собственно, последний спарк там 3.2, он не то чтобы старый, мы на таком и работаем, но далеко не самый последний.

sshikov 24 апр в 21:10

eval(input: InternalRow)

Ну вот хорошо что у автора 0-арная функция. А если мы заходим использовать аргументы? Я пытался как-то разобраться с expressions, но так и уперся в отсутствие документации, скажем, непонятно где вот эти InternalRow взять, и что с ними можно делать.