Многократные итерации sha2 в столбце искровых данных

Question

У меня есть случай, когда мне нужно вычислить 5000 раундов sha512 в столбце строки. До сих пор я пытался использовать для этого функцию pyspark sha2, python «старый» UDF и python pandas udf. Я ищу способ ускорить вычисления.

Для pyspark Мне не удалось определить столбец, использующий sha2 5000 раз (переполнение стека даже при отображении определения столбца) - я определил его используя l oop:

for _ in range(5000):
     column = sha2(column,512)

Для python я определил аналогичную функцию, используя hashlib:

def sha(text):
    for _ in range(5000):
        text = hashlib.sha512(text.encode('utf-8')).hexdigest()

    return text

Но это приводит к издержкам сериализации / десериализации и передача данных.

Я экспериментировал с переписыванием этой функции для pandas_udf , но, к сожалению, узлы в моем кластере не содержат pyarrow, и он не изменится, пока мне не понадобится рабочий прототип.

Так что я ищу способ ускорить процесс.

• Я не знаю scala или java, но я открыт, чтобы попытаться использовать scala / java udf, чтобы ускорить процесс - я прав, предполагая, что переход на scala / java udf должен ускорить работу в этом сценарии?
• Есть ли способ, с помощью которого pyspark не позволяет определить такую ​​функцию?

Редактировать: я работаю с Spark 2,3 в python 3,7. Поэтому у меня нет доступа к функциям более высокого порядка, представленным в 2.4

Answer 1

Конечно, вы можете сделать это только с функциями SQL. Имея

df = spark.createDataFrame(["Hello World"], "string")

In Spark 2.4 или более поздней (ранее при использовании некоторых проприетарных платформ), вы можете

df.selectExpr("""aggregate(
    sequence(1, 5000),            -- Dummy sequence
    value,                        -- Init
    (acc, x) -> sha2(acc, 512)    -- Aggregation function
) AS hash""")

In Spark 3.1 или позже вы можете

from pyspark.sql.functions import aggregate, col, lit, sequence

df.select(aggregate(
    sequence(lit(1), lit(5000)),     # Dummy sequence
    col("value"),                    # Init
    lambda acc, _: sha2(acc, 512)    # Aggregation function
).alias("hash"))

Редактировать (если вы не можете обновить):

На практике 5000 раундов хеширования, вероятно, более чем компенсируют стоимость перемещения данных, поэтому у вас должно быть все в порядке с простым udf, особенно для прототипирования.