Hadoop & Przykłady Mapreduce: Utwórz pierwszy program w Javie

Spisie treści:

Anonim

W tym samouczku nauczysz się używać Hadoop z przykładami MapReduce. Użyte dane wejściowe to SalesJan2009.csv. Zawiera informacje związane ze sprzedażą, takie jak nazwa produktu, cena, tryb płatności, miasto, kraj klienta itp. Celem jest sprawdzenie liczby produktów sprzedanych w każdym kraju.

W tym samouczku nauczysz się:

  • Pierwszy program Hadoop MapReduce
  • Wyjaśnienie klasy SalesMapper
  • Wyjaśnienie klasy SalesCountryReducer
  • Wyjaśnienie klasy SalesCountryDriver

Pierwszy program Hadoop MapReduce

Teraz w tym samouczku MapReduce utworzymy nasz pierwszy program MapReduce w języku Java:

Dane sprzedaży: styczeń 2009 r

Upewnij się, że masz zainstalowany Hadoop. Przed rozpoczęciem właściwego procesu zmień użytkownika na „hduser” (identyfikator używany podczas konfiguracji Hadoop, możesz przełączyć się na identyfikator użytkownika używany podczas konfiguracji programowania Hadoop).

su - hduser_

Krok 1)

Utwórz nowy katalog o nazwie MapReduceTutorial, jak pokazano w poniższym przykładzie MapReduce

sudo mkdir MapReduceTutorial

Przyznaj uprawnienia

sudo chmod -R 777 MapReduceTutorial

SalesMapper.java 

package SalesCountry;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.mapred.*;public class SalesMapper extends MapReduceBase implements Mapper  {private final static IntWritable one = new IntWritable(1);public void map(LongWritable key, Text value, OutputCollector  output, Reporter reporter) throws IOException {String valueString = value.toString();String[] SingleCountryData = valueString.split(",");output.collect(new Text(SingleCountryData[7]), one);}}

SalesCountryReducer.java 

package SalesCountry;import java.io.IOException;import java.util.*;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapred.*;public class SalesCountryReducer extends MapReduceBase implements Reducer {public void reduce(Text t_key, Iterator values, OutputCollector output, Reporter reporter) throws IOException {Text key = t_key;int frequencyForCountry = 0;while (values.hasNext()) {// replace type of value with the actual type of our valueIntWritable value = (IntWritable) values.next();frequencyForCountry += value.get();}output.collect(key, new IntWritable(frequencyForCountry));}}

SalesCountryDriver.java 

package SalesCountry;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.*;import org.apache.hadoop.mapred.*;public class SalesCountryDriver {public static void main(String[] args) {JobClient my_client = new JobClient();// Create a configuration object for the jobJobConf job_conf = new JobConf(SalesCountryDriver.class);// Set a name of the Jobjob_conf.setJobName("SalePerCountry");// Specify data type of output key and valuejob_conf.setOutputKeyClass(Text.class);job_conf.setOutputValueClass(IntWritable.class);// Specify names of Mapper and Reducer Classjob_conf.setMapperClass(SalesCountry.SalesMapper.class);job_conf.setReducerClass(SalesCountry.SalesCountryReducer.class);// Specify formats of the data type of Input and outputjob_conf.setInputFormat(TextInputFormat.class);job_conf.setOutputFormat(TextOutputFormat.class);// Set input and output directories using command line arguments,//arg[0] = name of input directory on HDFS, and arg[1] = name of output directory to be created to store the output file.FileInputFormat.setInputPaths(job_conf, new Path(args[0]));FileOutputFormat.setOutputPath(job_conf, new Path(args[1]));my_client.setConf(job_conf);try {// Run the jobJobClient.runJob(job_conf);} catch (Exception e) {e.printStackTrace();}}}

Pobierz pliki tutaj

Sprawdź uprawnienia do wszystkich tych plików

a jeśli brakuje uprawnień do odczytu, przyznaj te same-

Krok 2)

Eksportuj ścieżkę klasy, jak pokazano w poniższym przykładzie Hadoop

export CLASSPATH="$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.2.0.jar:$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-common-2.2.0.jar:$HADOOP_HOME/share/hadoop/common/hadoop-common-2.2.0.jar:~/MapReduceTutorial/SalesCountry/*:$HADOOP_HOME/lib/*"

Krok 3)

Skompiluj pliki Java (te pliki znajdują się w katalogu Final-MapReduceHandsOn ). Pliki klas zostaną umieszczone w katalogu pakietu

javac -d . SalesMapper.java SalesCountryReducer.java SalesCountryDriver.java

To ostrzeżenie można bezpiecznie zignorować.

Ta kompilacja utworzy katalog w bieżącym katalogu o nazwie z nazwą pakietu określoną w pliku źródłowym java (tj. SalesCountry w naszym przypadku) i umieści w nim wszystkie skompilowane pliki klas.

Krok 4)

Utwórz nowy plik Manifest.txt

sudo gedit Manifest.txt

dodaj do niego następujące wiersze,

Main-Class: SalesCountry.SalesCountryDriver

SalesCountry.SalesCountryDriver to nazwa klasy głównej. Pamiętaj, że musisz nacisnąć klawisz Enter na końcu tej linii.

Krok 5)

Utwórz plik Jar

jar cfm ProductSalePerCountry.jar Manifest.txt SalesCountry/*.class

Sprawdź, czy plik jar został utworzony

Krok 6)

Uruchom Hadoop

$HADOOP_HOME/sbin/start-dfs.sh
$HADOOP_HOME/sbin/start-yarn.sh

Krok 7)

Skopiuj plik SalesJan2009.csv do ~ / inputMapReduce

Teraz użyj poniższego polecenia, aby skopiować ~ / inputMapReduce do HDFS.

$HADOOP_HOME/bin/hdfs dfs -copyFromLocal ~/inputMapReduce /

Możemy spokojnie zignorować to ostrzeżenie.

Sprawdź, czy plik jest faktycznie kopiowany, czy nie.

$HADOOP_HOME/bin/hdfs dfs -ls /inputMapReduce

Krok 8)

Uruchom zadanie MapReduce

$HADOOP_HOME/bin/hadoop jar ProductSalePerCountry.jar /inputMapReduce /mapreduce_output_sales

Spowoduje to utworzenie katalogu wyjściowego o nazwie mapreduce_output_sales na HDFS. Zawartość tego katalogu będzie plikiem zawierającym sprzedaż produktów w poszczególnych krajach.

Krok 9)

Wynik można zobaczyć poprzez interfejs poleceń jako,

$HADOOP_HOME/bin/hdfs dfs -cat /mapreduce_output_sales/part-00000

Wyniki można również wyświetlić za pomocą interfejsu internetowego jako

Otwórz r w przeglądarce internetowej.

Teraz wybierz „Przeglądaj system plików” i przejdź do / mapreduce_output_sales

Otwórz część r-00000

Wyjaśnienie klasy SalesMapper

W tej sekcji zrozumiemy implementację klasy SalesMapper .

1. Zaczynamy od określenia nazwy pakietu dla naszej klasy. SalesCountry to nazwa naszego pakietu. Należy pamiętać, że wynik kompilacji SalesMapper.class zostanie przeniesiony do katalogu o nazwie tej pakietu: SalesCountry .

Następnie importujemy pakiety bibliotek.

Poniższy zrzut ekranu przedstawia implementację klasy SalesMapper-

Przykładowe objaśnienie kodu:

1. Definicja klasy SalesMapper-

public class SalesMapper rozszerza MapReduceBase implementuje Mapper {

Każda klasa mappera musi być rozszerzona z klasy MapReduceBase i musi implementować interfejs Mappera .

2. Definiowanie funkcji „mapy” -

public void map(LongWritable key,Text value,OutputCollector output,Reporter reporter) throws IOException

Główną częścią klasy Mapper jest metoda „map ()” , która przyjmuje cztery argumenty.

Przy każdym wywołaniu metody „map ()” przekazywana jest para klucz-wartość ( „klucz” i „wartość” w tym kodzie).

Metoda 'map ()' rozpoczyna się od podzielenia tekstu wejściowego, który jest odbierany jako argument. Używa tokenizera do dzielenia tych wierszy na słowa.

String valueString = value.toString();String[] SingleCountryData = valueString.split(",");

W tym przypadku „,” jest używany jako separator.

Następnie tworzona jest para przy użyciu rekordu pod siódmym indeksem tablicy „SingleCountryData” i wartości „1” .

output.collect (new Text (SingleCountryData [7]), one);

Wybieramy rekord na siódmym indeksie, ponieważ potrzebujemy danych o kraju i znajduje się on na siódmym indeksie w tablicy „SingleCountryData” .

Proszę pamiętać, że nasze dane wejściowe są w poniższym formacie (gdzie kraj jest 7 th indeksu, z 0 jako indeks początkowy) -

Data_transakcji, produkt, cena, typ_płaty, nazwa, miasto, stan, kraj , utworzono_konto, ostatnie_logowanie, szerokość geograficzna, długość geograficzna

Dane wyjściowe programu odwzorowującego to ponownie para klucz-wartość, która jest wyprowadzana za pomocą metody „collect ()” elementu „OutputCollector” .

Wyjaśnienie klasy SalesCountryReducer

W tej sekcji zrozumiemy implementację klasy SalesCountryReducer .

1. Zaczynamy od określenia nazwy pakietu dla naszej klasy. SalesCountry to nazwa pakietu out. Należy pamiętać, że wynik kompilacji SalesCountryReducer.class zostanie przeniesiony do katalogu o nazwie tej pakietu: SalesCountry .

Następnie importujemy pakiety bibliotek.

Poniższy zrzut ekranu przedstawia implementację klasy SalesCountryReducer-

Objaśnienie kodu:

1. SalesCountryReducer Class Definicja-

public class SalesCountryReducer rozszerza MapReduceBase implementuje Reducer {

Tutaj pierwsze dwa typy danych, „Text” i „IntWritable”, są typami danych wejściowych klucz-wartość do reduktora.

Dane wyjściowe programu mapującego są w postaci , . To wyjście programu odwzorowującego staje się wejściem do reduktora. Tak więc, aby wyrównać z typem danych, jako typ danych używane są tutaj Text i IntWritable .

Ostatnie dwa typy danych, „Tekst” i „IntWritable”, to typ danych wyjściowych generowanych przez reduktor w postaci pary klucz-wartość.

Każda klasa reduktora musi być rozszerzona z klasy MapReduceBase i musi implementować interfejs Reducer .

2. Definiowanie funkcji „zmniejsz” -

public void reduce( Text t_key,Iterator values,OutputCollector output,Reporter reporter) throws IOException {

Dane wejściowe do metody redukuj () to klucz z listą wielu wartości.

Na przykład w naszym przypadku będzie to-

, , , , , .

Jest to podane dla reduktora jako

Tak więc, aby zaakceptować argumenty tego formularza, używane są pierwsze dwa typy danych, a mianowicie Text i Iterator . Tekst jest typem danych klucza, a Iterator jest typem danych dla listy wartości dla tego klucza.

Następny argument jest typu OutputCollector , który zbiera dane wyjściowe fazy reduktora.

Metodaredred () rozpoczyna się od skopiowania wartości klucza i zainicjowania licznika częstotliwości do 0.

Klucz tekstowy = t_key; int frequencyForCountry = 0;

Następnie, używając pętli „ while ” , iterujemy listę wartości powiązanych z kluczem i obliczamy końcową częstotliwość, sumując wszystkie wartości.

 while (values.hasNext()) {// replace type of value with the actual type of our valueIntWritable value = (IntWritable) values.next();frequencyForCountry += value.get();}

Teraz wypychamy wynik do kolektora wyjściowego w postaci klucza i uzyskanej liczby częstotliwości .

Poniższy kod robi to-

output.collect(key, new IntWritable(frequencyForCountry));

Wyjaśnienie klasy SalesCountryDriver

W tej sekcji zrozumiemy implementację klasy SalesCountryDriver

1. Zaczynamy od określenia nazwy pakietu dla naszej klasy. SalesCountry to nazwa pakietu out. Należy pamiętać, że wynik kompilacji SalesCountryDriver.class zostanie przeniesiony do katalogu o nazwie tej pakietu: SalesCountry .

Oto wiersz określający nazwę pakietu, po którym następuje kod importu pakietów bibliotek.

2. Zdefiniuj klasę sterownika, która utworzy nowe zadanie klienta, obiekt konfiguracyjny i rozgłosi klasy Mapper i Reducer.

Klasa sterownika jest odpowiedzialna za ustawienie naszego zadania MapReduce do uruchamiania w Hadoop. W tej klasie określamy nazwę zadania, typ danych wejścia / wyjścia oraz nazwy klas mapowania i reduktora .

3. W poniższym fragmencie kodu ustawiamy katalogi wejściowe i wyjściowe, które są używane odpowiednio do konsumowania zestawu danych wejściowych i generowania danych wyjściowych.

arg [0] i arg [1] to argumenty wiersza poleceń przekazane za pomocą polecenia podanego w programie praktycznym MapReduce, tj.

$ HADOOP_HOME / bin / hadoop jar ProductSalePerCountry.jar / inputMapReduce / mapreduce_output_sales

4. Uruchom naszą pracę

Poniższy kod rozpoczyna wykonywanie zadania MapReduce-

try {// Run the jobJobClient.runJob(job_conf);} catch (Exception e) {e.printStackTrace();}