Тайны таинственного сортировки в Hadoop

Введение

В загадочном ночном рынке зачаровательный образ, украшенный роскошной маской, движется по-прежнему среди шумного толпы. Этот загадочный танцор с маской, кажется, обладает тайной силой, легко упорядочивая беспорядочные стойки при каждом вращении и качании. Ваша задача - разгадать тайну замечательного таланта, освоив искусство Hadoop Shuffle Comparable.

Skills Graph

Реализация маппера

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

Сначала смените пользователя на hadoop, а затем перейдите в домашнюю директорию пользователя hadoop:

su - hadoop

Затем создайте Java-файл для класса маппера:

touch /home/hadoop/WordLengthMapper.java

Добавьте следующий код в файл WordLengthMapper.java:

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordLengthMapper extends Mapper<LongWritable, Text, CompositeKey, Text> {

    private CompositeKey compositeKey = new CompositeKey();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split("\\s+");

        for (String word : words) {
            compositeKey.setFirstChar(word.charAt(0));
            compositeKey.setLength(word.length());
            context.write(compositeKey, new Text(word));
        }
    }
}

В вышеприведенном коде мы создаем класс WordLengthMapper, который наследует класс Mapper из Hadoop MapReduce API. Метод map принимает ключ LongWritable (представляющий смещение байтов входной строки) и значение Text (сама входная строка). Затем он разделяет входную строку на отдельные слова, создает объект CompositeKey для каждого слова (содержащий первый символ и длину слова) и передает CompositeKey в качестве ключа и слово в качестве значения.

Реализация составного ключа

В этом шаге мы создадим пользовательский класс CompositeKey, который реализует интерфейс WritableComparable из Hadoop MapReduce API. Этот класс будет использоваться в качестве ключа в нашей MapReduce задаче, позволяя нам сортировать и группировать данные на основе первого символа и длины каждого слова.

Сначала создайте Java-файл для класса CompositeKey:

touch /home/hadoop/CompositeKey.java

Затем добавьте следующий код в файл CompositeKey.java:

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class CompositeKey implements WritableComparable<CompositeKey> {

    private char firstChar;
    private int length;

    public CompositeKey() {
    }

    public void setFirstChar(char firstChar) {
        this.firstChar = firstChar;
    }

    public char getFirstChar() {
        return firstChar;
    }

    public void setLength(int length) {
        this.length = length;
    }

    public int getLength() {
        return length;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeChar(firstChar);
        out.writeInt(length);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        firstChar = in.readChar();
        length = in.readInt();
    }

    @Override
    public int compareTo(CompositeKey other) {
        int cmp = Character.compare(firstChar, other.firstChar);
        if (cmp!= 0) {
            return cmp;
        }
        return Integer.compare(length, other.length);
    }

    @Override
    public int hashCode() {
        return firstChar + length;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof CompositeKey) {
            CompositeKey other = (CompositeKey) obj;
            return firstChar == other.firstChar && length == other.length;
        }
        return false;
    }

    @Override
    public String toString() {
        return firstChar + ":" + length;
    }
}

В вышеприведенном коде мы создаем класс CompositeKey, который реализует интерфейс WritableComparable. Он имеет два поля: firstChar (первый символ слова) и length (длина слова). Класс предоставляет методы доступа и установки для этих полей, а также реализации методов write, readFields, compareTo, hashCode, equals и toString, требуемых интерфейсом WritableComparable.

Метод compareTo особенно важен, так как он определяет, как ключи будут сортироваться в MapReduce задаче. В нашей реализации мы сначала сравниваем поля firstChar двух ключей. Если они разные, мы возвращаем результат этой сравнения. Если поля firstChar одинаковы, мы сравниваем поля length.

Реализация редьюсера

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

Сначала создайте Java-файл для класса редьюсера:

touch /home/hadoop/WordLengthReducer.java

Затем добавьте следующий код в файл WordLengthReducer.java:

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordLengthReducer extends Reducer<CompositeKey, Text, CompositeKey, Text> {

    public void reduce(CompositeKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuilder sb = new StringBuilder();
        for (Text value : values) {
            sb.append(value.toString()).append(", ");
        }
        sb.setLength(sb.length() - 2);
        context.write(key, new Text(sb.toString()));
    }
}

В вышеприведенном коде мы создаем класс WordLengthReducer, который наследует класс Reducer из Hadoop MapReduce API. Метод reduce принимает ключ CompositeKey (содержащий первый символ и длину слова) и итерируемый объект Text значений (слова, соответствующие ключу).

Внутри метода reduce мы объединяем все слова, соответствующие ключу, в строку, разделенную запятыми. Мы используем StringBuilder для эффективного построения выходной строки, и удаляем конечную запятую и пробел перед записью пары ключ-значение в выход.

Реализация драйвера

В этом шаге мы создадим класс драйвера для настройки и запуска MapReduce задачи.

Сначала создайте Java-файл для класса драйвера:

touch /home/hadoop/WordLengthDriver.java

Затем добавьте следующий код в файл WordLengthDriver.java:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordLengthDriver {

    public static void main(String[] args) throws Exception {
        if (args.length!= 2) {
            System.err.println("Usage: WordLengthDriver <input> <output>");
            System.exit(1);
        }

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Word Length");

        job.setJarByClass(WordLengthDriver.class);
        job.setMapperClass(WordLengthMapper.class);
        job.setReducerClass(WordLengthReducer.class);
        job.setOutputKeyClass(CompositeKey.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true)? 0 : 1);
    }
}

В вышеприведенном коде мы создаем класс WordLengthDriver, который является точкой входа для нашей MapReduce задачи. Метод main принимает два аргумента командной строки: путь к входным данным и путь к выходным данным для задачи.

Внутри метода main мы создаем новый объект Configuration и новый объект Job. Мы настраиваем задачу, устанавливая классы маппера и редьюсера, классы выходных ключей и значений, а также пути к входным и выходным данным.

Наконец, мы отправляем задачу и ждем ее завершения. Если задача завершилась успешно, мы выходим с кодом статуса 0; в противном случае - с кодом статуса 1.

Для запуска задачи вы можете использовать следующую команду:

javac -source 8 -target 8 -classpath "/home/hadoop/:/home/hadoop/hadoop/share/hadoop/common/hadoop-common-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/common/lib/*" -d /home/hadoop /home/hadoop/WordLengthMapper.java /home/hadoop/CompositeKey.java /home/hadoop/WordLengthReducer.java /home/hadoop/WordLengthDriver.java
jar cvf word-length.jar *.class
hadoop jar word-length.jar WordLengthDriver /input /output

Наконец, мы можем проверить результаты, выполнив следующую команду:

hadoop fs -cat /output/*

Пример вывода:

A:3 Amr
A:6 AADzCv
A:10 AlGyQumgIl
...
h:7 hgQUIhA
h:8 hyrjMGbY, hSElGKux
h:10 hmfHJjCkwB
...
z:6 zkpRCN
z:8 zfMHRbtk
z:9 zXyUuLHma

Резюме

В этом лабе мы изучили концепцию Hadoop Shuffle Comparable, реализовав MapReduce задачу, которая группирует слова по первому символу и длине. Мы создали пользовательский маппер для передачи пар ключ-значение с составным ключом, пользовательский класс CompositeKey, который реализует интерфейс WritableComparable, редьюсер для объединения слов с одинаковым ключом и класс драйвера для настройки и запуска задачи.

Через этот лаб я получил более глубокое понимание Hadoop MapReduce фреймворка и важности пользовательских типов данных и сортировки в распределенных вычислениях. Осваивание Hadoop Shuffle Comparable позволяет нам проектировать эффективные алгоритмы для обработки и анализа данных, раскрывая силу больших данных, подобно тому, как загадочный танцор в маске упорядочивает хаотичные торговые посты на ночном рынке.