Variabili globali in hadoop

Question

Il mio programma segue una mappa iterativa/approccio di riduzione. E ha bisogno di fermarsi se sono soddisfatte determinate condizioni. È comunque possibile impostare una variabile globale che può essere distribuita su tutte le mappe/ridurre le attività e verificare se la variabile globale raggiunge la condizione per il completamento.

Qualcosa di simile.

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); 

} 

Dove condizione è una variabile globale che viene modificato durante/dopo ogni mappa/ridurre esecuzione.

Answer 1

Ogni volta che si esegue un lavoro di riduzione della mappa, è possibile esaminare lo stato dell'output, i valori contenuti nei contatori, ecc. E prendere una decisione sul nodo che sta controllando l'iterazione se si desidera uno più iterazione o no. Immagino di non capire da dove viene la necessità di uno stato globale nel tuo scenario.

Più in generale - ci sono due modi principali in cui lo stato è condiviso tra l'esecuzione di nodi (sebbene si noti che lo stato di condivisione è meglio evitato poiché limita la scalabilità).

1. scrivere un file HDFS che altri nodi in grado di leggere (assicurarsi che il file viene ripulito quando le uscite di posti di lavoro, e che l'esecuzione speculativa non causerà guasti strani).
2. Utilizzare ZooKeeper per memorizzare alcuni dati nei nodi dell'albero ZK dedicati.

Answer 2

È possibile utilizzare Cascading per organizzare più lavori Hadoop. Specificare un percorso HDFS in cui si desidera mantenere la variabile di stato globale e inizializzarla con contenuti fittizi. In ogni iterazione, leggi i contenuti correnti di questo percorso HDFS, elimina quei contenuti, esegui qualsiasi numero di mappe/riduci i passaggi e infine esegui una riduzione globale che aggiorna la variabile di stato globale. A seconda della natura dell'attività, potrebbe essere necessario disabilitare l'esecuzione speculativa e consentire numerosi tentativi.

Answer 3

È possibile utilizzare Configuration.set (String name, String value) per impostare un valore si sarà in grado di accedere nel vostro Mappers/Riduttori/etc:

Nel driver:

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

E ad esempio nel mapper:

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

Ma questo non sarà probabilmente aiuterà a guardare sull'uscita di posti di lavoro precedenti per decidere se avviare un altro iterazione. Cioè questo valore non verrà respinto dopo l'esecuzione del lavoro.

È anche possibile utilizzare Hadoop DistributedCache per archiviare i file che verranno distribuiti tra tutti i nodi. Questo è un po 'meglio che archiviare qualcosa su HDFS se un valore che si sta per passare in questo modo è qualcosa di piccolo.

Ovviamente counters può essere utilizzato anche per questo scopo. Ma non sembrano troppo affidabili allo scopo di prendere decisioni nell'algoritmo. Sembra che in alcuni casi possano essere incrementati due volte (se alcune attività sono state eseguite più di una volta, ad esempio in caso di fallimento o esecuzione speculativa) - Non sono sicuro.

Answer 4

Ecco come funziona in Hadoop 2.0

Nel driver:

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

E nel tuo 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 
    }