cluster computing - software - ¿Cuál es el significado de "Locality Level" en Spark cluster

spark software

Question

¿Cuál es el significado del título "Nivel local" y el estado 5 Datos locales -> proceso local -> nodo local -> rack local -> Cualquiera?

Answer 1

El nivel de localidad, hasta donde yo sé, indica qué tipo de acceso a los datos se ha realizado. Cuando un nodo finaliza todo su trabajo y su CPU queda inactiva, Spark puede decidir iniciar otra tarea pendiente que requiera obtener datos de otros lugares. Por lo tanto, idealmente, todas sus tareas deben procesarse localmente ya que están asociadas a una latencia de acceso a datos más baja.

Puede configurar el tiempo de espera antes de moverse a otros niveles de localidad usando:

spark.locality.wait

Se puede encontrar más información sobre los parámetros en los documentos de configuración de Spark

Con respecto a los diferentes niveles PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, o CUALQUIERA, creo que los métodos findTask y findSpeculativeTask en org.apache.spark.scheduler.TaskSetManager ilustran cómo Spark elige las tareas según su nivel de localidad. Primero comprobará las tareas PROCESS_LOCAL que se lanzarán en el mismo proceso de ejecutor. De lo contrario, comprobará las tareas NODE_LOCAL que puedan estar en otros ejecutores en el mismo nodo o que necesiten recuperarse de sistemas como HDFS, en caché, etc. RACK_LOCAL significa que los datos están en otro nodo y, por lo tanto, deben transferirse antes ejecución. Y finalmente, CUALQUIER es solo para tomar cualquier tarea pendiente que pueda ejecutarse en el nodo actual.

  /**
   * Dequeue a pending task for a given node and return its index and locality level.
   * Only search for tasks matching the given locality constraint.
   */
  private def findTask(execId: String, host: String, locality: TaskLocality.Value)
    : Option[(Int, TaskLocality.Value)] =
  {
    for (index <- findTaskFromList(execId, getPendingTasksForExecutor(execId))) {
      return Some((index, TaskLocality.PROCESS_LOCAL))
    }

    if (TaskLocality.isAllowed(locality, TaskLocality.NODE_LOCAL)) {
      for (index <- findTaskFromList(execId, getPendingTasksForHost(host))) {
        return Some((index, TaskLocality.NODE_LOCAL))
      }
    }

    if (TaskLocality.isAllowed(locality, TaskLocality.RACK_LOCAL)) {
      for {
        rack <- sched.getRackForHost(host)
        index <- findTaskFromList(execId, getPendingTasksForRack(rack))
      } {
        return Some((index, TaskLocality.RACK_LOCAL))
      }
    }

    // Look for no-pref tasks after rack-local tasks since they can run anywhere.
    for (index <- findTaskFromList(execId, pendingTasksWithNoPrefs)) {
      return Some((index, TaskLocality.PROCESS_LOCAL))
    }

    if (TaskLocality.isAllowed(locality, TaskLocality.ANY)) {
      for (index <- findTaskFromList(execId, allPendingTasks)) {
        return Some((index, TaskLocality.ANY))
      }
    }

    // Finally, if all else has failed, find a speculative task
    findSpeculativeTask(execId, host, locality)
  }