Как решить проблему перекоса данных MapReduce?

Hadoop

Это седьмой день моего участия в августовском испытании обновлений, подробности о мероприятии:Испытание августовского обновления

текст

Что такое искажение данных?

Перекос данных является распространенным явлением в данных.

Выбросы неизбежно появятся в данных и приведут к искажению данных.

Эти выбросы могут значительно замедлить выполнение MapReduce.

Общие перекосы данных следующие:

  1. Частота данных искажена — количество данных в одной области намного больше, чем в других областях.
  2. Неравномерность размера данных — некоторые записи намного больше среднего.

Возможные причины перекоса данных MapReduce

Перекос данных может возникать как на стороне карты, так и на стороне уменьшения.

Перекос данных на стороне карты снижает эффективность обработки различных наборов данных.

Перекос данных на стороне сокращения часто возникает из-за разделителя MapReduce по умолчанию.

Возможные последствия перекоса данных MapReduce

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

Как диагностировать искажение данных

Добавить возможность записывать детали выходных ключей карты в методе сокращения.

После обнаружения искаженных данных необходимо диагностировать те ключи, которые вызывают искаженные данные.

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

package com.shockang.study.bigdata.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

public class WordCountReducerWithDataSkew extends Reducer<Text, IntWritable, Text, IntWritable> {

    public static final String MAX_VALUES = "skew.maxvalues";
    private int maxValueThreshold;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        Configuration conf = context.getConfiguration();
        maxValueThreshold = Integer.parseInt(conf.get(MAX_VALUES));
    }

    /*
            key: hello
            value: List(1, 1, ...)
        */
    protected void reduce(Text key, Iterable<IntWritable> values,
                          Context context) throws IOException, InterruptedException {
        int i = 0;
        for (IntWritable value : values) {
            System.out.println(value);
            i++;
        }

        if (++i > maxValueThreshold) {
            System.out.println("Received " + i + " values for key " + key);
        }
    }
}

После запуска задания по логу можно определить, какие ключи перекошены и насколько они перекошены.

Отслеживание искаженных данных — важный шаг в понимании данных и важная основа для разработки заданий MapReduce.

Обычные средства для уменьшения перекоса данных на стороне сокращения

  1. Выборка и разделение диапазона

Разделитель Hadoop по умолчанию основан на хеш-значении выходного ключа карты. Это хорошо только тогда, когда данные распределены достаточно равномерно. Это очень проблематично, когда есть перекос данных. Использование разделителя требует сначала понимания характеристик данных. Граничное значение раздела может быть задано набором результатов, полученным путем выборки исходных данных. Разделители диапазона могут разделять по заданным значениям границ раздела. Таким образом, его также можно очень хорошо использовать для исправления искаженных данных для некоторых ключей в данных.

  1. пользовательский раздел

Другой альтернативой выборке и разбиению по диапазонам является создание пользовательского разбиения на основе базовых знаний о выходных ключах. Например, если слово выходного ключа карты взято из книги. Большинство из них должны быть стоп-словами. Затем пользовательский раздел может отправить эту часть многоточия в фиксированную часть редукции экземпляров. А все остальное отправить на оставшиеся редюкции.

  1. Combine

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

  1. Соединения на стороне карты и полусоединения

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

  1. Пользовательская стратегия для перекоса размера данных
  • Установите mapred.linerecordreader.maxlength, чтобы ограничить максимальную длину чтения RecordReader. RecordReader используется в классах TextInputFormat и KeyValueTextInputFormat. Длина по умолчанию не имеет верхнего предела.
  • Установите максимальное количество записей для кэшированного набора данных через org.apache.hadoop.contrib.utils.join. Верхний предел записей кэша по умолчанию при уменьшении составляет 100.
  • Рассмотрите возможность сжатия данных с использованием структур данных с потерями, таких как фильтры Блума.
Категории