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

Question

Моя программа следует итеративному подходу «карта / уменьшение».И это нужно прекратить, если соблюдены определенные условия.Есть ли в любом случае, я могу установить глобальную переменную, которая может быть распределена по всем задачам карты / сокращения и проверить, достигает ли глобальная переменная условия для завершения.

Примерно так.

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

Вы можете использовать Configuration.set (String name, String value) , чтобы установить значение, к которому вы сможете получить доступ в ваших Mappers / Reducers / etc:

В вашем драйвере:

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

И, например, в вашем маппере:

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

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

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

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

Answer 2

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

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

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

Answer 3

Вот как это работает в 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
    }

Answer 4

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