как пропустить плохие записи в Hadoop Map-Reduce

Поскольку вы фильтруете записи на основе отсутствия полей, эта логика подходит для вашей реализации Mapper. Mapper API Java может выглядеть примерно так:

public class FilteringMapper extends Mapper<LongWritable, Text, LongWritable, Text>{

private static final Logger _logger = Logger.getLogger(FilteringMapper.class);

protected void map(LongWritable key, Text value, Context context) {

    if(recordIsBad(value))
        _logger.info(<log record data you care about>);
    else
        context.write(key, value);

}

private boolean recordIsBad(Text record){
    //return true if record is bad by your standards
}

}

Этот Mapper будет фильтровать только на основе ваших стандартов. Если вам нужны дальнейшие преобразования данных в Mapper, это легко добавить.

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

Таким образом, для задачи, которая постоянно терпит неудачу из-за плохой записи, средство отслеживания задач запускает следующие попытки задачи со следующими результатами:

1. Задача не выполняется.

2. Задача не выполняется.

3. Режим пропуска включен. Задача завершается неудачно, но запись об ошибке сохраняется в трекере задач.

4. Режим пропуска по-прежнему включен. Задача завершается успешно, пропуская плохую запись, которая не удалась в предыдущей попытке.

Режим пропуска по умолчанию отключен; вы включаете его независимо для карты и уменьшаете задачи, используя класс SkipBadRecords. Важно отметить, что режим пропуска может обнаруживать только одну плохую запись за попытку задачи, поэтому этот механизм подходит только для обнаружения случайных плохих записей (скажем, нескольких для каждой задачи). Возможно, вам придется увеличить максимальное количество попыток выполнения задачи (через mapred.map.max.attempts и mapred.reduce.max.attempts), чтобы дать режиму пропуска достаточно попыток для обнаружения и пропуска всех неверных записей во входном разделении. Плохие записи, обнаруженные Hadoop, сохраняются в виде файлов последовательности в выходном каталоге задания в подкаталоге _logs/skip. Их можно проверить в диагностических целях после завершения задания (например, с помощью hadoop fs -text).

Текст из книги Тома Уайта Hadoop: полное руководство.

Вы можете пропустить записи в задании MapReduce, установив для свойств mapreduce.map.skip.maxrecords и/или mapreduce.reduce.skip.maxgroups значение >0 (по умолчанию они равны 0).

Вот простая команда, которую вы можете запустить в оболочке с помощью потоковой передачи Hadoop. для запуска задания MapReduce, которое будет игнорировать 1000 отказов картографа:

mapred streaming \
  -D mapreduce.map.skip.maxrecords=1000 \
  -file map.sh \
  -file reduce.sh \
  -input myInputDir \
  -output myOutputDir \
  -mapper map.sh \
  -reducer reduce.sh

(где map.sh и reduce.sh — исполняемые bash-скрипты).

Из документация:

Hadoop предоставляет возможность пропустить определенный набор неправильных входных записей при обработке входных данных карты. Приложения могут управлять этой функцией через SkipBadRecords. класс.

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

Вы можете увидеть все свойства, связанные с пропуском записей, в mapred-default.xml.

Другое важное свойство — mapreduce.task.skip.start.attempts: количество неудачных попыток, прежде чем задача MR начнет пропускать записи (по умолчанию — 2).