performance - Funciones Spark vs rendimiento UDF?

apache-spark pyspark

Question

Spark ahora ofrece funciones predefinidas que se pueden usar en marcos de datos, y parece que están altamente optimizadas. Mi pregunta original iba a ser cuál es más rápida, pero hice algunas pruebas y descubrí que las funciones de chispa son aproximadamente 10 veces más rápidas al menos en una instancia. ¿Alguien sabe por qué esto es así y cuándo un udf sería más rápido (solo en los casos en que exista una función de chispa idéntica)?

Aquí está mi código de prueba (ejecutado en Databricks community ed):

# UDF vs Spark function
from faker import Factory
from pyspark.sql.functions import lit, concat
fake = Factory.create()
fake.seed(4321)

# Each entry consists of last_name, first_name, ssn, job, and age (at least 1)
from pyspark.sql import Row
def fake_entry():
  name = fake.name().split()
  return (name[1], name[0], fake.ssn(), fake.job(), abs(2016 - fake.date_time().year) + 1)

# Create a helper function to call a function repeatedly
def repeat(times, func, *args, **kwargs):
    for _ in xrange(times):
        yield func(*args, **kwargs)
data = list(repeat(500000, fake_entry))
print len(data)
data[0]

dataDF = sqlContext.createDataFrame(data, (''last_name'', ''first_name'', ''ssn'', ''occupation'', ''age''))
dataDF.cache()

Función UDF:

concat_s = udf(lambda s: s+ ''s'')
udfData = dataDF.select(concat_s(dataDF.first_name).alias(''name''))
udfData.count()

Función de chispa:

spfData = dataDF.select(concat(dataDF.first_name, lit(''s'')).alias(''name''))
spfData.count()

Funcionó ambas veces varias veces, el udf usualmente tomó alrededor de 1.1 - 1.4 s, y la función Spark concat siempre tomó menos de 0.15 s.

Answer 1

¿Cuándo sería más rápido un udf?

Si preguntas sobre Python UDF, la respuesta probablemente nunca sea *. Dado que las funciones de SQL son relativamente simples y no están diseñadas para tareas complejas, es prácticamente imposible compensar el costo de la serialización, deserialización y movimiento de datos repetidos entre el intérprete de Python y JVM.

¿Alguien sabe por qué esto es así?

Las razones principales ya se enumeraron anteriormente y pueden reducirse a un simple hecho de que Spark DataFrame es nativamente una estructura JVM y los métodos de acceso estándar se implementan mediante llamadas simples a la API de Java. UDF, por otro lado, se implementan en Python y requieren mover datos de un lado a otro.

Si bien PySpark en general requiere movimientos de datos entre JVM y Python, en el caso de API RDD de bajo nivel, generalmente no requiere una actividad de serde costosa. Spark SQL agrega un costo adicional de serialización y serialización, así como el costo de mover datos desde y hacia una representación insegura en JVM. El último es específico para todos los UDF (Python, Scala y Java), pero el primero es específico para lenguajes no nativos.

A diferencia de los UDF, las funciones de Spark SQL funcionan directamente en JVM y, por lo general, están bien integradas con Catalyst y Tungsten. Significa que estos pueden optimizarse en el plan de ejecución y la mayoría de las veces pueden beneficiarse de las optimizaciones de codgen y otras optimizaciones de tungsteno. Además, estos pueden operar en datos en su representación "nativa".

Entonces, en cierto sentido, el problema aquí es que Python UDF tiene que traer datos al código mientras que las expresiones SQL van al revés.

* Según estimaciones aproximadas, la ventana UDF de PySpark puede superar la función de ventana Scala.

Answer 2

Desde el 30 de octubre de 2017, Spark acaba de presentar udfs vectorizados para pyspark.

https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html

La razón por la que Python UDF es lento es probablemente que PySpark UDF no se implementa de la manera más optimizada:

Según el párrafo del enlace.

Spark agregó una API de Python en la versión 0.7, con soporte para funciones definidas por el usuario. Estas funciones definidas por el usuario operan una fila a la vez y, por lo tanto, sufren una alta sobrecarga de serialización e invocación.

Sin embargo, los udfs recientemente vectorizados parecen estar mejorando mucho el rendimiento:

desde 3x hasta más de 100x.