apache-spark - software - spark data science

¿Cómo funciona la partición Spark(ing) en archivos en HDFS?

Question

Estoy trabajando con Apache Spark en un clúster usando HDFS. Según tengo entendido, HDFS está distribuyendo archivos en nodos de datos. Entonces, si coloca un "archivo.txt" en el sistema de archivos, se dividirá en particiones. Ahora estoy llamando

rdd = SparkContext().textFile("hdfs://.../file.txt")

de Apache Spark. ¿Tiene rdd ahora automáticamente las mismas particiones que "file.txt" en el sistema de archivos? Que pasa cuando llamo

rdd.repartition(x)

donde x> entonces las particiones utilizadas por hdfs? ¿Spark reorganizará físicamente los datos en hdfs para que funcionen localmente?

Ejemplo: puse un archivo de texto de 30 GB en el sistema HDFS, que lo distribuye en 10 nodos. ¿Spark a) usará las mismas 10 particiones? y b) barajar 30 GB en el clúster cuando llamo repartition (1000)?

Answer 1

Aquí está la instantánea de " Cómo se cargan los bloques en HDFS en los trabajadores de Spark como particiones "

En estas imágenes se cargan 4 bloques HDFS como particiones Spark dentro de la memoria de 3 trabajadores

Ejemplo: puse un archivo de texto de 30 GB en el sistema HDFS, que lo distribuye en 10 nodos.

Will Spark

a) usa las mismas 10 particiones?

Spark carga los mismos 10 bloques HDFS en la memoria de los trabajadores como particiones. Supongo que el tamaño de bloque del archivo de 30 GB debe ser de 3 GB para obtener 10 particiones / bloques (con la configuración predeterminada)

b) barajar 30 GB en el clúster cuando llamo repartition (1000)?

Sí , Spark baraja los datos entre los nodos de trabajo para crear 1000 particiones en la memoria de los trabajadores.

Nota:

HDFS Block -> Spark partition : One block can represent as One partition (by default)
Spark partition -> Workers : Many/One partitions can present in One workers

Answer 2

Adición a @ 0x0FFF Si toma de HDFS como archivo de entrada, calculará como para este rdd = SparkContext().textFile("hdfs://.../file.txt") y cuando haga rdd.getNumPatitions resultará Max(2, Number of HDFS block) . Hice muchos experimentos y encontré esto como resultado. De nuevo, explícitamente, puede hacer rdd = SparkContext().textFile("hdfs://.../file.txt", 400) para obtener 400 como particiones o incluso puede hacer re-particiones por rdd.repartition o disminuir a 10 por rdd.coalesce(10)

Answer 3

Al leer archivos HDFS no agrupados (por ejemplo, parquet) con spark-sql, el número de particiones df.rdd.getNumPartitions depende de estos factores:

• spark.default.parallelism (se traduce aproximadamente a #cores disponibles para la aplicación)
• spark.sql.files.maxPartitionBytes (por defecto 128MB)
• spark.sql.files.openCostInBytes (predeterminado 4MB)

Una estimación aproximada del número de particiones es:

• Si tiene suficientes núcleos para leer todos sus datos en paralelo (es decir, al menos un núcleo por cada 128 MB de datos)

  AveragePartitionSize ≈ min(4MB, TotalDataSize/#cores) NumberOfPartitions ≈ TotalDataSize/AveragePartitionSize

• Si no tienes suficientes núcleos ,

  AveragePartitionSize ≈ 128MB NumberOfPartitions ≈ TotalDataSize/AveragePartitionSize

Los cálculos exactos son un poco complicados y se pueden encontrar en la base del código para FileSourceScanExec, consulte here .

Answer 4

Cuando Spark lee un archivo de HDFS, crea una única partición para una sola división de entrada. La división de entrada se establece mediante el InputFormat Hadoop utilizado para leer este archivo. Por ejemplo, si usa textFile() sería TextInputFormat en Hadoop, que le devolvería una única partición para un solo bloque de HDFS (pero la división entre particiones se haría en la división de línea, no la división exacta del bloque), a menos que Tienes un archivo de texto comprimido. En el caso de un archivo comprimido, obtendría una única partición para un solo archivo (ya que los archivos de texto comprimido no se pueden dividir).

Cuando llama a rdd.repartition(x) , realizaría una combinación aleatoria de los datos de N particiones que tiene en rdd a x particiones que desea tener, la partición se realizará en forma de turnos.

Si tiene un archivo de texto sin comprimir de 30GB almacenado en HDFS, entonces con la configuración predeterminada de tamaño de bloque HDFS (128MB) se almacenaría en 235 bloques, lo que significa que el RDD que leyó de este archivo tendría 235 particiones. Cuando llama a repartition(1000) su RDD se marcará como reparticionada , pero de hecho se barajará a 1000 particiones solo cuando ejecutará una acción sobre este RDD (concepto de ejecución diferida)