Variables globales en hadoop

Question

Mi programa sigue un enfoque iterativo de mapa/reducción. Y debe detenerse si se cumplen ciertas condiciones. De todos modos, puedo establecer una variable global que pueda distribuirse en todas las tareas de asignación/reducción y verificar si la variable global alcanza la condición para completarse.

Algo como esto.

While(Condition != true){ 

      Configuration conf = getConf(); 
      Job job = new Job(conf, "Dijkstra Graph Search"); 

      job.setJarByClass(GraphSearch.class); 
      job.setMapperClass(DijkstraMap.class); 
      job.setReducerClass(DijkstraReduce.class); 

      job.setOutputKeyClass(IntWritable.class); 
      job.setOutputValueClass(Text.class); 

} 

Donde la condición es una variable global que se modifica durante/después de cada mapa/reduce la ejecución.

Answer 1

Cada vez que ejecuta un trabajo de reducción de mapa, puede examinar el estado de la salida, los valores contenidos en los contadores, etc., y tomar una decisión en el nodo que controla la iteración sobre si desea una más iteración o no. Supongo que no entiendo de dónde viene la necesidad de un estado global en su escenario.

Más en general: hay dos formas principales en que el estado se comparte entre los nodos de ejecución (aunque debe tenerse en cuenta que el estado compartido es que mejor se evita ya que limita la escalabilidad).

1. Escriba un archivo en HDFS que otros nodos puedan leer (asegúrese de que el archivo se limpia cuando el trabajo finaliza y que la ejecución especulativa no ocasionará fallas extrañas).
2. Utilice ZooKeeper para almacenar algunos datos en nodos de árbol ZK dedicados.

Answer 2

Puede usar Cascading para organizar múltiples trabajos de Hadoop. Especifique una ruta HDFS donde desee mantener la variable de estado global e inicialice con contenido ficticio. En cada iteración, lea los contenidos actuales de esta ruta de HDFS, elimine esos contenidos, realice cualquier cantidad de pasos de asignación/reducción y, finalmente, realice una reducción global que actualice la variable de estado global. Dependiendo de la naturaleza de su tarea, es posible que deba deshabilitar la ejecución especulativa y permitir muchos intentos.

Answer 3

Puede utilizar Configuration.set (String nombre, String value) para establecer un valor que será capaz de acceder a sus aplicadores/Reductores/etc:

En su conductor:

conf.set("my.dijkstra.parameter", "value"); 

Y por ejemplo en su mapeador:

public void configure(JobConf job) { 
     myParam = job.get("my.dijkstra.parameter"); 
    } 

Pero esto no es probable que ayuda a mirar en la salida de las tareas previas para decidir si desea iniciar una iteración más. Es decir. este valor no se restituirá después de la ejecución del trabajo.

También puede usar Hadoop's DistributedCache para almacenar archivos que se distribuirán entre todos los nodos. Esto es un poco mejor que simplemente almacenar algo en HDFS si un valor que va a pasar de esta manera es algo pequeño.

Por supuesto counters también se puede utilizar para este fin. Pero no parecen demasiado confiables para los propósitos de tomar decisiones en el algoritmo. Parece que en algunos casos se pueden incrementar dos veces (si alguna tarea se ejecutó más de una vez, por ejemplo, en caso de error o ejecución especulativa) - No estoy seguro.

Answer 4

Así es como funciona en Hadoop 2.0

En su conductor:

conf.set("my.dijkstra.parameter", "value"); 

Y en su Mapper:

protected void setup(Context context) throws IOException, 
      InterruptedException { 
     Configuration conf = context.getConfiguration(); 

     strProp = conf.get("my.dijkstra.parameter"); 
     // and then you can use it 
    }