Глобальные переменные в hadoop

Question

Моя программа соответствует итеративному методу map/reduce. И он должен остановиться, если будут выполнены определенные условия. В любом случае я могу установить глобальную переменную, которая может быть распределена по всем задачам map/reduce и проверить, достигает ли глобальная переменная условие завершения.

Нечто подобное.

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

} 

Где условие - глобальная переменная, которая изменяется во время/после каждой карты/уменьшает выполнение.

Answer 1

Каждый раз, когда вы запускаете работу по сокращению карты, вы можете проверить состояние вывода, значения, содержащиеся в счетчиках и т. Д., И принять решение на узле, который контролирует итерацию, о том, хотите ли вы еще одного итерации или нет. Наверное, я не понимаю, откуда возникает потребность в глобальном состоянии в вашем сценарии.

В более общем плане - между двумя исполняемыми узлами разделяются два основных способа (хотя следует отметить, что состояние общего доступа лучше всего избегать, поскольку оно ограничивает масштабируемость).

1. Записать файл в HDFS, который могут считывать другие узлы (убедитесь, что файл очищается при завершении задания и что спекулятивное выполнение не приведет к возникновению странных сбоев).
2. Используйте ZooKeeper для хранения некоторых данных в выделенных узлах дерева ZK.

Answer 2

Вы можете использовать Cascading, чтобы организовать несколько работ Hadoop. Укажите путь HDFS, в котором вы хотите сохранить глобальную переменную состояния и инициализировать фиктивным содержимым. На каждой итерации прочитайте текущее содержимое этого пути HDFS, удалите это содержимое, выполните любое количество шагов карты/уменьшения и, наконец, выполните глобальное сокращение, которое обновляет глобальную переменную состояния. В зависимости от характера вашей задачи вам может потребоваться отключить спекулятивное выполнение и разрешить много попыток.

Answer 3

Вы можете использовать Configuration.set (имя String, String) значение установить значение, которое вы сможете получить доступ в ваших Картостроителях/Переходники/и т.д.:

В драйвере:

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

И, например, в вашем картографа:

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

Но это не будет, вероятно, поможет вам выглядеть на выходе предыдущих рабочих мест, чтобы решить, следует ли начинать еще одну итерацию. То есть это значение не будет отодвинуто после выполнения задания.

Вы также можете использовать Hadoop's DistributedCache для хранения файлов, которые будут распределены между всеми узлами. Это немного лучше, чем просто хранить что-то на HDFS, если значение, которое вы собираетесь пройти таким образом, является чем-то маленьким.

Конечно, counters также может быть использован для этой цели. Но они не выглядят слишком надежными для принятия решений в алгоритме. Похоже, в некоторых случаях их можно увеличить в два раза (если какая-то задача была выполнена более одного раза, например, в случае неудачи или спекулятивного исполнения) - я не уверен.

Answer 4

Как это работает в Hadoop 2.0

В драйвере:

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

И в вашем 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 
    }