Cómo hacer buenos ejemplos de marcos de datos Apache Spark reproducibles

Ajuste de rendimiento

Si la pregunta está relacionada con el ajuste de rendimiento, incluya la siguiente información.

Plan de ejecución

Es mejor incluir plan de ejecución extendido. En Python:

df.explain(True) 

En Scala:

df.explain(true)

O plan de ejecución ampliado con estadísticas. En Python:

print(df._jdf.queryExecution().stringWithStats())

En Scala:

df.queryExecution.stringWithStats

Modo e información del clúster

• mode - local, client, `Cluster.
• Administrador de clústeres (si corresponde) - ninguno (modo local), independiente, YARN, Mesos, Kubernetes.
• Información básica de configuración (número de núcleos, memoria del ejecutor).

Información de tiempo

slow es relativo, especialmente cuando portas una aplicación no distribuida o esperas una latencia baja. Los tiempos exactos para diferentes tareas y etapas se pueden recuperar desde la interfaz de usuario de Spark (sc.uiWebUrl) jobs o Spark REST UI.

Usar nombres estandarizados para contextos

El uso de nombres establecidos para cada contexto nos permite reproducir rápidamente el problema.

• sc para SparkContext.
• sqlContext para SQLContext.
• spark para SparkSession.

Proporcionar información de tipo (Scala )

La poderosa inferencia de tipos es una de las características más útiles de Scala, pero hace difícil analizar código sacado de contexto. Incluso si el tipo es obvio desde el contexto, es mejor anotar el variable. Prefer

val lines: RDD[String] = sc.textFile("path")
val words: RDD[String] = lines.flatMap(_.split(" "))

Sobre

val lines = sc.textFile("path")
val words = lines.flatMap(_.split(" "))

Las herramientas de uso común pueden ayudarlo:

• spark-shell / Scala shell

  Use :t

  scala> val rdd = sc.textFile("README.md")
  rdd: org.apache.spark.rdd.RDD[String] = README.md MapPartitionsRDD[1] at textFile at <console>:24
  
  scala> :t rdd
  org.apache.spark.rdd.RDD[String]
  

• InteliJ Idea

  Use Alt + =

Buena pregunta y respuesta; algunas sugerencias adicionales:

Incluye tu versión de Spark

La chispa todavía está evolucionando, aunque no tan rápidamente como en los días de 1.x. Siempre es (pero especialmente si está utilizando una versión algo más antigua) una buena idea incluir su versión de trabajo. Personalmente, siempre comienzo mis respuestas con:

spark.version
# u'2.2.0'

O

sc.version
# u'2.2.0'

Incluir tu versión de Python, también, nunca es una mala idea.

Incluya todos sus importaciones

Si su pregunta no es estrictamente sobre Spark SQL & dataframes, por ejemplo, si tiene la intención de usar su dataframe en alguna operación de aprendizaje automático, sea explícito sobre sus importaciones - vea esta pregunta, donde las importaciones se agregaron en el OP solo después de un extenso intercambio en los comentarios (ahora eliminados) (y resultó que estas importaciones incorrectas fueron la causa raíz del problema).

¿Por qué es necesario esto? Porque, por ejemplo, este LDA

from pyspark.mllib.clustering import LDA

Es diferente de este LDA:

from pyspark.ml.clustering import LDA

La primera proviene de la antigua API basada en RDD (anteriormente Spark MLlib), mientras que la segunda proviene de la nueva API basada en dataframe (Spark ML).

Incluir resaltado de código

Bien, confieso que esto es subjetivo: creo que las preguntas de PySpark no deben etiquetarse como python por defecto ; la cosa es, python etiqueta da automáticamente el código de resaltado (y creo que esta es una razón principal para aquellos que lo utilizan para PySpark preguntas). De todos modos, si estás de acuerdo, y todavía te gustaría un código bien resaltado, simplemente incluye la directiva markdown relevante:

<!-- language-all: lang-python -->

En algún lugar de tu post, antes de tu primer fragmento de código.

[ACTUALIZACIÓN: He solicitado resaltado de sintaxis automático para pyspark y sparkr etiquetas-votos positivos más bienvenidos]

Esta pequeña función auxiliar podría ayudar a analizar los archivos de salida de Spark en DataFrame:

PySpark:

def read_spark_output(file_path):
    t = spark.read \
             .option("header","true") \
             .option("inferSchema","true") \
             .option("delimiter","|") \
             .option("parserLib","UNIVOCITY") \
             .option("ignoreLeadingWhiteSpace","true") \
             .option("ignoreTrailingWhiteSpace","true") \
             .option("comment","+") \
             .csv("file:///tmp/spark.out")
    # select not-null columns
    return t.select([c for c in t.columns if not c.startswith("_")])

Scala:

// read Spark Output Fixed width table:
def readSparkOutput(filePath: String) : org.apache.spark.sql.DataFrame = {
    val t = spark.read
                 .option("header","true")
                 .option("inferSchema","true")
                 .option("delimiter","|")
                 .option("parserLib","UNIVOCITY")
                 .option("ignoreLeadingWhiteSpace","true")
                 .option("ignoreTrailingWhiteSpace","true")
                 .option("comment","+")
                 .csv(filePath)
    t.select(t.columns.filterNot(_.startsWith("_c")).map(t(_)):_*)
}

Uso:

df = read_spark_output("file:///tmp/spark.out")