Primjer Hadoop MapReduce: Prvi Java Program sa Code

U ovom vodiču naučit ćete koristiti Hadoop s MapReduce primjerima. Korišteni ulazni podaci su ProdajaSiječanj2009.csvSadrži podatke vezane uz prodaju kao što su naziv proizvoda, cijena, način plaćanja, grad i država klijenta. Cilj je saznati broj prodanih proizvoda u svakoj državi.

Prvi Hadoop MapReduce program

Sada u ovome Vodič za MapReduce, stvorit ćemo svoj prvi Java Program MapReduce:

Snimka zaslona u nastavku prikazuje neobrađene podatke o prodaji za siječanj 2009., gdje svaki redak predstavlja jednu transakciju, a država se nalazi u osmom stupcu odvojenom zarezom.

Provjerite imate li instaliran Hadoop. Prije nego što započnete sa stvarnim postupkom, promijenite korisnika u 'hduser' (id koji se koristio prilikom konfiguriranja Hadoop-a - možete se prebaciti na korisnički ID koji se koristio tijekom vlastite konfiguracije Hadoop-a).

su - hduser_

Uputa se mijenja u hduser račun, kao što je prikazano dolje.

Korak 1) Izradite direktorij projekta i izvorne datoteke

Izradite novi direktorij s nazivom MapReduceTutorial kao što je prikazano u donjem primjeru MapReduce.

sudo mkdir MapReduceTutorial

Dajte dozvole

sudo chmod -R 777 MapReduceTutorial

Stvorite tri Java izvorne datoteke u nastavku unutar MapReduceTutoriala. Imajte na umu da sva tri koriste stariju org.apache.hadoop.mapred API, koji se još uvijek isporučuje s Hadoopom 3.x.

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 <LongWritable, Text, Text, IntWritable> {
	private final static IntWritable one = new IntWritable(1);

	public void map(LongWritable key, Text value, OutputCollector <Text, IntWritable> 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<Text, IntWritable, Text, IntWritable> {

	public void reduce(Text t_key, Iterator<IntWritable> values, OutputCollector<Text,IntWritable> 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 value
			IntWritable 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 job
        JobConf job_conf = new JobConf(SalesCountryDriver.class);

        // Set a name of the Job
        job_conf.setJobName("SalePerCountry");

        // Specify data type of output key and value
        job_conf.setOutputKeyClass(Text.class);
        job_conf.setOutputValueClass(IntWritable.class);

        // Specify names of Mapper and Reducer Class
        job_conf.setMapperClass(SalesCountry.SalesMapper.class);
        job_conf.setReducerClass(SalesCountry.SalesCountryReducer.class);

        // Specify formats of the data type of Input and output
        job_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 job 
            JobClient.runJob(job_conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Preuzmite datoteke ovdje

Arhiva se proširuje u iste tri izvorne datoteke, kao što je ovdje prikazano.

Provjerite dopuštenja za sve te datoteke

Ako nedostaju dozvole za 'čitanje', dodijelite ih:

Korak 2) Izvoz Hadoop classpath-a

Izvezite putanju klase kao što je prikazano u donjem Hadoop primjeru.

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/*"

Izvezena putanja klase se vraća u promptu.

Korak 3) Sastavite Java slika

Sastavite Java datoteke (ove se datoteke nalaze u direktoriju Final-MapReduceHandsOn). Njihove datoteke klasa bit će smještene u direktorij paketa.

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

Ovo upozorenje se može sigurno zanemariti — ono samo izvještava da je mapirani API zastario.

Ova kompilacija će stvoriti direktorij u trenutnom direktoriju nazvan s nazivom paketa navedenim u Java izvornu datoteku (tj. SalesCountry u našem slučaju) i u nju staviti sve kompilirane datoteke klase.

Korak 4) Izradite datoteku manifesta

Stvorite novu datoteku Manifest.txt

sudo gedit Manifest.txt

Dodajte mu sljedeći redak:

Main-Class: SalesCountry.SalesCountryDriver

SalesCountry.SalesCountryDriver je naziv glavne klase. Imajte na umu da morate pritisnuti tipku Enter na kraju ovog retka.

Korak 5) Zapakirajte klase u staklenku

Stvorite Jar datoteku

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

Provjerite je li jar datoteka stvorena

Korak 6) Pokrenite Hadoop

Pokrenite Hadoop

$HADOOP_HOME/sbin/start-dfs.sh

$HADOOP_HOME/sbin/start-yarn.sh

Korak 7) Kopirajte ulaznu datoteku u HDFS

Kopirajte datoteku SalesJan2009.csv u ~/inputMapReduce

Sada upotrijebite naredbu u nastavku za kopiranje ~/inputMapReduce u HDFS.

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

Možemo slobodno zanemariti ovo upozorenje.

Provjerite je li datoteka stvarno kopirana ili ne.

$HADOOP_HOME/bin/hdfs dfs -ls /inputMapReduce

Korak 8) Pokrenite zadatak MapReduce

Pokrenite zadatak MapReduce

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

Ovo će stvoriti izlazni direktorij pod nazivom mapreduce_output_sales on HDFS. Sadržaj ovog imenika bit će datoteka koja će sadržavati prodaju proizvoda po zemlji.

Korak 9) Pročitajte rezultate

Rezultat se može vidjeti kroz sučelje naredbi kao,

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

Rezultati se također mogu vidjeti putem web sučelja kao-

Otvoren http://localhost:50070/ u web pregledniku. Na Hadoopu 3.x, web sučelje NameNodea premješteno je na port 9870, pa koristite http://localhost:9870/ tamo umjesto toga.

Sada odaberite 'Pregledaj datotečni sustav' i idite na /mapreduce_output_sales

Otvori dio-r-00000

Objašnjenje klase SalesMapper

S obzirom na to da se posao izvodi od početka do kraja, sljedeća tri odjeljka opisuju što svaka klasa zapravo radi.

U ovom odjeljku ćemo razumjeti implementaciju klase SalesMapper.

1. Počinjemo određivanjem naziva paketa za našu klasu. SalesCountry je naziv našeg paketa. Imajte na umu da će izlaz kompilacije, SalesMapper.class, ići u direktorij nazvan ovim nazivom paketa: SalesCountry.

Nakon toga uvozimo knjižnične pakete.

Donja snimka prikazuje implementaciju klase SalesMapper-

Uzorak Code Objašnjenje:

1. Definicija klase SalesMapper-

public class SalesMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {

Svaka klasa mappera mora biti proširena iz klase MapReduceBase i mora implementirati Mapper sučelje.

2. Definiranje funkcije 'map'-

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

Glavni dio klase Mapper je metoda 'map()' koja prihvaća četiri argumenta.

Pri svakom pozivu metode 'map()', prosljeđuje se par ključ-vrijednost ('ključ' i 'vrijednost' u ovom kodu).

Metoda 'map()' započinje dijeljenjem ulaznog teksta koji se prima kao argument. Svaki redak dijeli na polja.

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

Ovdje se ',' koristi kao razdjelnik.

Nakon toga, par se formira korištenjem zapisa na 7. indeksu polja 'SingleCountryData' i vrijednosti '1'.

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

Zapis na 7. indeksu odabiremo jer nam trebaju podaci o državi, a on se nalazi na 7. indeksu u nizu 'SingleCountryData'.

Imajte na umu da su naši ulazni podaci u sljedećem formatu (gdje je Država na 7. indeksu, s 0 kao početnim indeksom) -

Transaction_date,Product,Price,Payment_Type,Name,City,State,Country,Account_Created,Last_Login,Latitude,Longitude

Izlaz mappera je opet par ključ-vrijednost koji se emitira pomoću metode 'collect()' klase 'OutputCollector'.

Objašnjenje klase SalesCountryReducer

U ovom odjeljku ćemo razumjeti implementaciju klase SalesCountryReducer.

1. Počinjemo određivanjem naziva paketa za našu klasu. SalesCountry je naziv našeg paketa. Imajte na umu da će izlaz kompilacije, SalesCountryReducer.class, ići u direktorij nazvan ovim nazivom paketa: SalesCountry.

Nakon toga uvozimo knjižnične pakete.

Donja snimka prikazuje implementaciju klase SalesCountryReducer.

Code Objašnjenje:

1. Definicija klase SalesCountryReducer-

public class SalesCountryReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {

Ovdje su prva dva tipa podataka, 'Text' i 'IntWritable', tip podataka ulaznog ključa-vrijednosti za reducer.

Izlaz mappera je u obliku , Ovaj izlaz mappera postaje ulaz za reducer. Dakle, radi usklađivanja s njegovim tipom podataka, ovdje se kao tip podataka koriste Text i IntWritable.

Posljednja dva tipa podataka, 'Text' i 'IntWritable', su tipovi podataka izlaza koje generira reducer u obliku para ključ-vrijednost.

Svaka klasa reducera mora biti proširena iz klase MapReduceBase i mora implementirati Reducer sučelje.

2. Definiranje funkcije reduciranja-

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

Ulaz u metodu reduce() je ključ s popisom više vrijednosti.

Na primjer, u našem slučaju to će biti-

, , , , , .

To se daje reduktoru kao

Dakle, za prihvaćanje argumenata ovog oblika koriste se prva dva tipa podataka, i to tekst i iterator. Tekst je tip podataka ključa i iteratora. je tip podataka za popis vrijednosti za taj ključ.

Sljedeći argument je tipa OutputCollector koji prikuplja izlaz faze redukcije.

Metoda reduce() započinje kopiranjem vrijednosti ključa i inicijalizacijom brojača frekvencija na 0.

Text key = t_key;
int frequencyForCountry = 0;

Zatim, koristeći petlju 'while', iteriramo kroz popis vrijednosti povezanih s ključem i izračunavamo konačnu frekvenciju zbrajanjem svih vrijednosti.

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

Sada rezultat šaljemo na izlazni kolektor u obliku ključa i dobivenog brojača frekvencija.

Donji kod radi ovo-

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

Objašnjenje klase SalesCountryDriver

U ovom odjeljku ćemo razumjeti implementaciju klase SalesCountryDriver.

1. Počinjemo određivanjem naziva paketa za našu klasu. SalesCountry je naziv našeg paketa. Imajte na umu da će izlaz kompilacije, SalesCountryDriver.class, ići u direktorij nazvan ovim nazivom paketa: SalesCountry.

Ovdje je redak koji navodi naziv paketa nakon kojeg slijedi kod za uvoz paketa knjižnice.

2. Definirajte klasu pokretača koja će kreirati novi posao klijenta, konfiguracijski objekt i oglašavati klase Mapper i Reducer.

Klasa upravljačkog programa odgovorna je za postavljanje našeg MapReduce posla za izvođenje HadoopU ovoj klasi određujemo naziv posla, tip podataka ulaza/izlaza i nazive klasa mappera i reducera.

3. U donjem isječku koda postavljamo ulazne i izlazne direktorije koji se koriste za potrošnju ulaznog skupa podataka i proizvodnju izlaza.

arg[0] i arg[1] su argumenti naredbenog retka koji se prosljeđuju s naredbom zadanom u praktičnom MapReduceu, tj.

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

4. Pokrenite naš posao

Donji kod pokreće izvršavanje MapReduce posla -

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

Pitanja i odgovori