ตัวอย่าง Hadoop MapReduce: ขั้นแรก Java โปรแกรมด้วย Code

ในบทช่วยสอนนี้ คุณจะได้เรียนรู้การใช้ Hadoop กับตัวอย่าง MapReduce ข้อมูลอินพุตที่ใช้คือ ยอดขายมกราคม 2009.csvไฟล์นี้ประกอบด้วยข้อมูลที่เกี่ยวข้องกับการขาย เช่น ชื่อสินค้า ราคา วิธีการชำระเงิน เมือง และประเทศของลูกค้า โดยมีเป้าหมายเพื่อหาจำนวนสินค้าที่ขายได้ในแต่ละประเทศ

โปรแกรม Hadoop MapReduce ตัวแรก

ตอนนี้อยู่ในนี้ บทช่วยสอน MapReduceเราจะสร้างของเราก่อน Java โปรแกรม MapReduce:

ภาพหน้าจอข้างล่างแสดงข้อมูลดิบของยอดขายเดือนมกราคม 2009 โดยแต่ละบรรทัดคือหนึ่งรายการธุรกรรม และประเทศจะอยู่ในคอลัมน์ที่แปดซึ่งคั่นด้วยเครื่องหมายจุลภาค

ตรวจสอบให้แน่ใจว่าคุณได้ติดตั้ง Hadoop แล้ว ก่อนเริ่มกระบวนการจริง ให้เปลี่ยนผู้ใช้เป็น 'hduser' (รหัสผู้ใช้ที่ใช้ในการกำหนดค่า Hadoop — คุณสามารถเปลี่ยนไปใช้รหัสผู้ใช้ที่ใช้ในระหว่างการกำหนดค่า Hadoop ของคุณเองได้)

su - hduser_

ข้อความแจ้งเตือนจะเปลี่ยนไปใช้บัญชี hduser ดังแสดงด้านล่าง

ขั้นตอนที่ 1) สร้างไดเร็กทอรีโครงการและไฟล์ต้นฉบับ

สร้างไดเร็กทอรีใหม่ชื่อ MapReduceTutorial ดังตัวอย่าง MapReduce ด้านล่าง

sudo mkdir MapReduceTutorial

ให้สิทธิ์

sudo chmod -R 777 MapReduceTutorial

สร้างสาม Java ไฟล์ต้นฉบับด้านล่างอยู่ใน MapReduceTutorial โปรดทราบว่าทั้งสามไฟล์ใช้เวอร์ชันเก่ากว่า org.apache.hadoop.mapred API ซึ่งยังคงมาพร้อมกับ Hadoop 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);
	}
}

ฝ่ายขาย CountryReducer.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();
        }
    }
}

ดาวน์โหลดไฟล์ที่นี่

ไฟล์เก็บข้อมูลจะขยายออกเป็นไฟล์ต้นฉบับสามไฟล์เช่นเดียวกับที่แสดงไว้ในที่นี้

ตรวจสอบการอนุญาตไฟล์ของไฟล์เหล่านี้ทั้งหมด

หากไม่มีสิทธิ์ในการอ่าน โปรดให้สิทธิ์เหล่านั้น:

ขั้นตอนที่ 2) ส่งออก classpath ของ Hadoop

ส่งออก classpath ดังแสดงในตัวอย่าง 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/*"

คลาสพาธที่ส่งออกจะแสดงกลับมาที่พรอมต์

ขั้นตอนที่ 3) รวบรวม Java ไฟล์

รวบรวม Java ไฟล์เหล่านี้ (อยู่ในไดเร็กทอรี Final-MapReduceHandsOn) ส่วนไฟล์คลาสจะถูกใส่ไว้ในไดเร็กทอรีแพ็กเกจ

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

สามารถเพิกเฉยต่อคำเตือนนี้ได้อย่างปลอดภัย เพราะมันเพียงแค่แจ้งว่า API ของ mapred นั้นเลิกใช้งานแล้ว

การคอมไพล์นี้จะสร้างไดเร็กทอรีในไดเร็กทอรีปัจจุบัน โดยตั้งชื่อตามชื่อแพ็กเกจที่ระบุไว้ในไฟล์ Java ไฟล์ต้นฉบับ (เช่น SalesCountry ในกรณีของเรา) และใส่ไฟล์คลาสที่คอมไพล์แล้วทั้งหมดไว้ในนั้น

ขั้นตอนที่ 4) สร้างไฟล์ manifest

สร้างไฟล์ใหม่ชื่อ Manifest.txt

sudo gedit Manifest.txt

เพิ่มบรรทัดต่อไปนี้ลงไป:

Main-Class: SalesCountry.SalesCountryDriver

SalesCountry.SalesCountryDriver คือชื่อของคลาสหลัก โปรดทราบว่าคุณต้องกดปุ่ม Enter ที่ท้ายบรรทัดนี้

ขั้นตอนที่ 5) บรรจุคลาสเรียนลงในขวดโหล

สร้างไฟล์ Jar

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

ตรวจสอบว่าไฟล์ jar ถูกสร้างขึ้น

ขั้นตอนที่ 6) เริ่มใช้งาน Hadoop

เริ่ม Hadoop

$HADOOP_HOME/sbin/start-dfs.sh

$HADOOP_HOME/sbin/start-yarn.sh

ขั้นตอนที่ 7) คัดลอกไฟล์อินพุตไปยัง HDFS

คัดลอกไฟล์ SalesJan2009.csv ไปยัง ~/inputMapReduce

ตอนนี้ให้ใช้คำสั่งด้านล่างเพื่อคัดลอก ~/inputMapReduce ไปยัง HDFS

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

เราสามารถเพิกเฉยต่อคำเตือนนี้ได้อย่างปลอดภัย

ตรวจสอบว่าไฟล์ถูกคัดลอกจริงหรือไม่

$HADOOP_HOME/bin/hdfs dfs -ls /inputMapReduce

ขั้นตอนที่ 8) รันงาน MapReduce

เรียกใช้งาน MapReduce

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

สิ่งนี้จะสร้างไดเร็กทอรีเอาต์พุตชื่อ mapreduce_output_sales on เอชดีเอฟเอส- เนื้อหาของไดเร็กทอรีนี้จะเป็นไฟล์ที่มียอดขายผลิตภัณฑ์ตามประเทศ

ขั้นตอนที่ 9) อ่านผลลัพธ์

สามารถดูผลลัพธ์ได้ผ่านทางอินเทอร์เฟซคำสั่งดังนี้

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

สามารถดูผลลัพธ์ได้ผ่านทางเว็บอินเตอร์เฟสเช่นกัน

จุดเปิด http://localhost:50070/ ในเว็บเบราว์เซอร์ บน Hadoop 3.x ส่วนติดต่อผู้ใช้บนเว็บของ NameNode ได้ย้ายไปยังพอร์ต 9870, ดังนั้นใช้ http://localhost:9870/ ที่นั่นแทน

จากนั้นเลือก 'เรียกดูระบบไฟล์' และไปยังไดเร็กทอรี /mapreduce_output_sales

เปิด part-r-00000

คำอธิบายของคลาส SalesMapper

เมื่อกระบวนการทำงานดำเนินไปตั้งแต่ต้นจนจบ ส่วนถัดไปสามส่วนจะอธิบายว่าแต่ละคลาสทำอะไรบ้าง

ในส่วนนี้ เราจะทำความเข้าใจเกี่ยวกับการใช้งานคลาส SalesMapper

1. เริ่มต้นด้วยการกำหนดชื่อแพ็กเกจสำหรับคลาสของเรา โดยชื่อแพ็กเกจคือ SalesCountry โปรดทราบว่าไฟล์ผลลัพธ์ของการคอมไพล์ คือ SalesMapper.class จะถูกจัดเก็บไว้ในไดเร็กทอรีที่มีชื่อเดียวกับชื่อแพ็กเกจนี้ คือ SalesCountry

ตามด้วยสิ่งนี้ เรานำเข้าแพ็คเกจไลบรารี

ภาพด้านล่างแสดงตัวอย่างการใช้งานคลาส SalesMapper

ตัวอย่าง Code คำอธิบาย:

1. คำจำกัดความของคลาส SalesMapper-

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

คลาส Mapper ทุกคลาสจะต้องสืบทอดมาจากคลาส MapReduceBase และต้องใช้งานอินเทอร์เฟซ Mapper ด้วย

2. การกำหนดฟังก์ชัน 'แผนที่'-

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

ส่วนหลักของคลาส Mapper คือเมธอด 'map()' ซึ่งรับอาร์กิวเมนต์สี่ตัว

ในการเรียกใช้เมธอด 'map()' แต่ละครั้ง จะมีการส่งคู่คีย์-ค่า ('key' และ 'value' ในโค้ดนี้) เข้าไป

เมธอด 'map()' เริ่มต้นด้วยการแยกข้อความที่รับเข้ามาเป็นอาร์กิวเมนต์ โดยจะแยกแต่ละบรรทัดออกเป็นฟิลด์ต่างๆ

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

ในที่นี้ เครื่องหมาย ',' ถูกใช้เป็นตัวคั่น

หลังจากนั้น จะมีการสร้างคู่โดยใช้เรคอร์ดที่ดัชนีลำดับที่ 7 ของอาร์เรย์ 'SingleCountryData' และค่า '1'

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

เราเลือกเรคอร์ดที่ดัชนีลำดับที่ 7 เพราะเราต้องการข้อมูลประเทศ และข้อมูลนั้นอยู่ที่ดัชนีลำดับที่ 7 ในอาร์เรย์ 'SingleCountryData'

โปรดทราบว่าข้อมูลที่เราป้อนนั้นอยู่ในรูปแบบดังต่อไปนี้ (โดยที่ประเทศอยู่ในลำดับที่ 7 และเริ่มต้นที่ 0)

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

ผลลัพธ์ของ mapper จะเป็นคู่คีย์-ค่า ซึ่งถูกส่งออกมาโดยใช้วิธี 'collect()' ของ 'OutputCollector'

คำอธิบายของคลาส SalesCountryReducer

ในส่วนนี้ เราจะทำความเข้าใจเกี่ยวกับการใช้งานคลาส SalesCountryReducer

1. เริ่มต้นด้วยการกำหนดชื่อแพ็กเกจสำหรับคลาสของเรา ในที่นี้ชื่อแพ็กเกจคือ SalesCountry โปรดทราบว่าไฟล์ผลลัพธ์ของการคอมไพล์ คือ SalesCountryReducer.class จะถูกจัดเก็บไว้ในไดเร็กทอรีที่มีชื่อเดียวกับชื่อแพ็กเกจนี้ คือ SalesCountry

ตามด้วยสิ่งนี้ เรานำเข้าแพ็คเกจไลบรารี

ภาพด้านล่างแสดงตัวอย่างการใช้งานคลาส SalesCountryReducer

Code คำอธิบาย:

1. คำจำกัดความของคลาส SalesCountryReducer-

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

ในที่นี้ ชนิดข้อมูลสองชนิดแรกคือ 'Text' และ 'IntWritable' เป็นชนิดข้อมูลของคู่คีย์-ค่าที่ป้อนเข้าสู่ตัวลด (reducer)

ผลลัพธ์ของ mapper อยู่ในรูปแบบดังนี้ , ผลลัพธ์จาก mapper จะกลายเป็นอินพุตให้กับ reducer ดังนั้น เพื่อให้สอดคล้องกับชนิดข้อมูล จึงใช้ Text และ IntWritable เป็นชนิดข้อมูลในที่นี้

ประเภทข้อมูลสองประเภทสุดท้ายคือ 'Text' และ 'IntWritable' เป็นประเภทข้อมูลเอาต์พุตที่สร้างโดยตัวลด (reducer) ในรูปแบบคู่คีย์-ค่า

คลาส Reducer ทุกคลาสจะต้องสืบทอดมาจากคลาส MapReduceBase และต้องใช้งานอินเทอร์เฟซ Reducer ด้วย

2. การกำหนดฟังก์ชัน 'ลด'-

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

อินพุตของเมธอด reduce() คือคีย์ที่มีลิสต์ของค่าหลายค่า

ตัวอย่างเช่น ในกรณีของเรา มันจะเป็น-

- - - - - -

สิ่งนี้ถูกส่งไปยังตัวลดขนาดดังนี้

ดังนั้น ในการรับอาร์กิวเมนต์ในรูปแบบนี้ จึงต้องใช้ชนิดข้อมูลสองชนิดแรก ได้แก่ Text และ Iterator ข้อความ (Text) เป็นชนิดข้อมูลของคีย์และตัววนซ้ำ (Iterator) เป็นชนิดข้อมูลสำหรับรายการค่าต่างๆ สำหรับคีย์นั้น

อาร์กิวเมนต์ถัดไปเป็นประเภท OutputCollector ซึ่งรวบรวมผลลัพธ์จากขั้นตอนการลดขนาด

เมธอด reduce() เริ่มต้นด้วยการคัดลอกค่าคีย์และกำหนดค่าความถี่เริ่มต้นเป็น 0

Text key = t_key;
int frequencyForCountry = 0;

จากนั้น ใช้ลูป 'while' เพื่อวนซ้ำผ่านรายการค่าที่เชื่อมโยงกับคีย์ และคำนวณความถี่สุดท้ายโดยการรวมค่าทั้งหมดเข้าด้วยกัน

 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

ในส่วนนี้ เราจะทำความเข้าใจเกี่ยวกับการใช้งานคลาส SalesCountryDriver

1. เริ่มต้นด้วยการกำหนดชื่อแพ็กเกจสำหรับคลาสของเรา โดยชื่อแพ็กเกจคือ SalesCountry โปรดทราบว่าไฟล์ผลลัพธ์ของการคอมไพล์ คือ SalesCountryDriver.class จะถูกจัดเก็บไว้ในไดเร็กทอรีที่มีชื่อเดียวกับชื่อแพ็กเกจนี้ คือ SalesCountry

นี่คือบรรทัดที่ระบุชื่อแพ็คเกจตามด้วยรหัสเพื่อนำเข้าแพ็คเกจไลบรารี

2. กำหนดคลาสไดรเวอร์ซึ่งจะสร้างงานไคลเอนต์ใหม่ ออบเจ็กต์การกำหนดค่า และโฆษณาคลาส Mapper และ ลด

คลาสไดรเวอร์มีหน้าที่ตั้งค่างาน MapReduce ของเราให้ทำงาน Hadoopในคลาสนี้ เราจะระบุชื่อของงาน ประเภทข้อมูลของอินพุต/เอาต์พุต และชื่อของคลาส mapper และ reducer

3. ในโค้ดด้านล่าง เราได้ตั้งค่าไดเร็กทอรีอินพุตและเอาต์พุตซึ่งใช้เพื่อใช้ชุดข้อมูลอินพุตและสร้างเอาต์พุตตามลำดับ

arg[0] และ arg[1] คืออาร์กิวเมนต์บรรทัดคำสั่งที่ส่งผ่านคำสั่งที่กำหนดในการฝึกปฏิบัติ MapReduce เช่น

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

4. กระตุ้นงานของเรา

โค้ดด้านล่างนี้จะเริ่มการทำงานของงาน MapReduce -

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

คำถามที่พบบ่อย