apache spark - org - ¿Cuál debería ser el valor óptimo para spark.sql.shuffle.partitions o cómo aumentamos las particiones cuando utilizamos Spark SQL?

org apache spark examples

Question

Hola, estoy usando Spark SQL en realidad hiveContext.sql() que usa group por consultas y me estoy encontrando con problemas OOM . Así que está pensando en aumentar el valor de spark.sql.shuffle.partitions de 200 por defecto a 1000 pero no está ayudando. Por favor, corríjanme si me equivoco, estas particiones compartirán la carga aleatoria de los datos para que más particiones menos datos contengan. Por favor, guía Soy nuevo en Spark. Estoy usando Spark 1.4.0 y tengo alrededor de 1TB de datos sin comprimir para procesar usando el grupo hiveContext.sql() por consultas.

Answer 1

De acuerdo, entonces creo que su problema es más general. No es específico de Spark SQL, es un problema general con Spark, donde ignora la cantidad de particiones que le indica cuando los archivos son pocos. Spark parece tener el mismo número de particiones que la cantidad de archivos en HDFS, a menos que llame a repartition . Así que llamar a la repartition debería funcionar, pero tiene la advertencia de causar una confusión algo innecesario.

Planteé esta pregunta hace un tiempo y todavía tengo que obtener una buena respuesta :(

Spark: aumentar el número de particiones sin causar un shuffle?

Answer 2

En realidad, depende de tus datos y tu consulta, si Spark debe cargar 1 TB, hay algo mal en tu diseño.

Utilice la interfaz de usuario web superbe para ver el DAG, es decir, cómo Spark está traduciendo su consulta SQL a tareas / etapas y tareas.

Las métricas útiles son "Input" y "Shuffle".

• Particiona tus datos (diseño de Hive / directory como / year = X / month = X)
• Utilice la función spark CLUSTER BY para trabajar por partición de datos
• Utilice el formato de archivo ORC / Parquet porque proporcionan "filtro de empuje hacia abajo", los datos inútiles no se cargan en Spark
• Analiza Spark History para ver cómo Spark está leyendo datos

Además, OOM podría ocurrir en su controlador?

-> este es otro problema, el controlador recogerá al final los datos que desee. Si solicita demasiados datos, el controlador OOM, intente limitar su consulta o escriba otra tabla (sintaxis Spark CREATE TABLE ...AS ).

Answer 3

Me encontré con esta publicación de Cloudera sobre Hive Partitioning. Consulte la sección "Sugerencias" que habla sobre el número de particiones y el número de archivos en cada partición que da como resultado la sobrecarga del nodo de nombre, lo que podría causar OOM.

Answer 4

Si se está quedando sin memoria al mezclar, intente configurar spark.sql.shuffle.partitions en 2001.

Spark utiliza una estructura de datos diferente para la contabilidad aleatoria cuando el número de particiones es mayor que 2000:

private[spark] object MapStatus {

  def apply(loc: BlockManagerId, uncompressedSizes: Array[Long]): MapStatus = {
    if (uncompressedSizes.length > 2000) {
      HighlyCompressedMapStatus(loc, uncompressedSizes)
    } else {
      new CompressedMapStatus(loc, uncompressedSizes)
    }
  }
...

Realmente me gustaría que te dejaran configurar esto de forma independiente.

Por cierto, encontré esta información en un mazo de diapositivas de Cloudera .