Agradeceríamos tu ayuda para difundir nuestros escritos sobre las ciencias informáticas.
Solución:
Aunque la respuesta a su pregunta se encuentra solo en uno de los siguientes problemas, permítame volver a escribir su ejemplo para explicar lo que está sucediendo.
Estableciendo su configuración
Primero, no necesita iniciar y detener un contexto para establecer su configuración. Desde chispa 2.0, puede crear la sesión de chispa y luego establecer las opciones de configuración.
from pyspark.sql import SparkSession
spark = (SparkSession.builder.appName("yourAwesomeApp").getOrCreate())
spark.conf.set("spark.executor.memory", "40g")
spark.conf.set("spark.executor.cores", "2")
Leyendo tus datos
Spark evaluará perezosamente el DAG. El tiempo que está midiendo en su recorte no es la carga de los datos en el marco de datos, sino solo la inferencia del esquema para el archivo JSON.. La inferencia de esquemas es costosa, debe intentar evitarla configurando el esquema de sus datos. Verá una gran diferencia en el rendimiento entre:
df = spark.read.json("../data/a_very_large_json.json.gz")
y
from pyspark.sql.types import (
StructType,
StringType,
StructField,
)
json_schema = schema = StructType([
StructField('data', StructType([
StructField("field1", StringType(), nullable=False),
StructField("field2", StringType(), nullable=False),
StructField("field3", StringType(), nullable=True),
StructField("field4", StringType(), nullable=True),
StructField("field5", LongType(), nullable=False),
])),
])
df = spark.read.json("../data/a_very_large_json.json.gz", schema=json_schema)
Si proporciona el esquema, esta instrucción debería ser casi instantánea. Como ya mencionó otro usuario, para ejecutar la tarea es necesario tener una actividad, como mostrar, encabezar, recopilar, persistir, etc.
df.show()
Puede establecer la cantidad de instancias de ejecutor y núcleos en la configuración, pero el uso real de esas instancias también depende de sus datos de entrada y las transformaciones/acciones que realice. Según su descripción, asumo que está trabajando en modo independiente, por lo que tener una instancia de ejecutor será el valor predeterminado (usando todos los núcleos), y debe configurar la memoria del ejecutor para usar la que tiene disponible. Por lo que recuerdo, cuando trabajas en un modo independiente, el spark.executor.instances se ignora y el número real de ejecutores se basa en el número de núcleos disponibles y el spark.executor.cores
Comparación con los pandas
Si está trabajando con un solo nodo, cargando los datos en un marco de datos, la comparación entre chispa y pandas es injusta. Spark siempre tendrá una sobrecarga más alta. Las chispas brillarán cuando tenga conjuntos de datos que no caben en la memoria de una máquina y tenga varios nodos para realizar el trabajo de cálculo. Si te sientes cómodo con los pandas, creo que te pueden interesar los koalas de Databricks.
Recomendación
Prefiero configurar los detalles de ejecución fuera de la aplicación (por ejemplo, usando los parámetros de envío de chispa). En raras ocasiones, para mejorar el rendimiento, deberá configurar algunos de ellos en el código, pero con cada nueva versión de Spark, esto es menos frecuente. Si puede lograr esto, su aplicación estará más preparada para el futuro y será más fácil de escalar.
spark.sparkContext.stop()
spark = SparkSession.builder.config(conf=conf).getOrCreate()
df = spark.read.json("../Data/inasnelylargefile.json.gz")
Agrega esto:
df.show()
##OR
df.persist()
La comparación que está haciendo no es manzanas con manzanas, Spark realiza una evaluación perezosa, lo que significa que si no llama a una acción sobre su operación, no hará nada más que compilar y mantener el DAG listo para usted.
En Spark, hay dos conceptos,
- Transformación: Evaluada perezosamente
- Acciones: (como recolectar(), tomar(), mostrar(), persistir()) evaluadas instantáneamente.
En su caso, read() es solo una transformación, agregar una acción debería desencadenar el cálculo.
Más sobre acciones vs transformación: https://training.databricks.com/visualapi.pdf
La razón por la que su lectura de Spark es más lenta que pandas es porque el archivo gz no se puede dividir, por lo tanto, Spark tiene que leer todo el archivo con una sola tarea. Sin embargo, al leer un archivo sin comprimir o un archivo comprimido con un formato de compresión divisible como bzip2, Spark implementará una cantidad x de tareas en paralelo (hasta la cantidad de núcleos disponibles en su clúster) para leer el archivo. Intente descomprimir el archivo antes de pasarlo a Spark.
Te mostramos las reseñas y valoraciones de los lectores
Tienes la opción de amparar nuestro trabajo escribiendo un comentario o dejando una puntuación te lo agradecemos.