MapReduce
Ikhtisar: data besar adalah paradigma baru dan pengolahannya adalah daerah yang paling penting untuk berkonsentrasi. data besar mengacu pada volume besar data yang berasal dari sumber yang berbeda seperti media sosial, Data sensor, data seluler, internet data dan banyak lagi. Pada artikel ini kita akan berkonsentrasi pada bagian pengolahan menggunakan framework Hadoop dan Peta-Mengurangi pemrograman. Peta-Mengurangi dapat didefinisikan sebagai tipe khusus dari kerangka kerja pemrograman yang digunakan untuk memproses sejumlah besar data dalam kerangka yang disebut hardware komoditas didistribusikan. Jadi artikel ini akan menjelaskan Peta-Mengurangi konsep dan implementasi praktis dengan menggunakan kode Java.
Pengantar: Hadoop MapReduce dapat didefinisikan sebagai kerangka kerja pemrograman perangkat lunak yang digunakan untuk memproses volume yang besar dari data yang (di tingkat terabyte) dalam lingkungan paralel node berkerumun. cluster terdiri dari ribuan node perangkat keras komoditas. pengolahan didistribusikan, handal dan kesalahan toleran. Sebuah pekerjaan MapReduce khas dilakukan sesuai dengan langkah-langkah berikut
1) Membagi data ke dalam potongan independen berdasarkan pasangan kunci-nilai. Hal ini dilakukan dengan Peta tugas secara paralel.
2) Out put dari pekerjaan Map diurutkan berdasarkan nilai key.
3) output diurutkan adalah masukan untuk pekerjaan Mengurangi. Dan kemudian menghasilkan output akhir pengolahan dan mengembalikan hasilnya ke klien.
Note: hardware komoditas pada dasarnya sistem komputer biaya rendah
kerangka MapReduce: kerangka Apache Hadoop MapReduce ditulis di Jawa. Kerangka kerja ini terdiri dari konfigurasi master-budak. master dikenal sebagai JobTracker dan budak dikenal sebagai master TaskTrackers.The mengontrol tugas diproses pada budak (yang tidak lain adalah node dalam sebuah cluster). Perhitungan dilakukan pada budak. Jadi menghitung dan penyimpanan node yang sama di lingkungan cluster. Konsep ini ‘ pindah perhitungan untuk node di mana data disimpan ', dan itu membuat pengolahan lebih cepat.
MapReduce Pengolahan: Model kerangka MapReduce adalah berat sangat ringan. Sehingga biaya hardware rendah dibandingkan dengan kerangka kerja lainnya. Tetapi pada saat yang sama kita harus memahami bahwa model bekerja secara efisien hanya dalam lingkungan terdistribusi pengolahan dilakukan pada node di mana data berada. Fitur lain seperti skalabilitas, keandalan dan toleransi kesalahan juga bekerja dengan baik pada lingkungan terdistribusi.
pelaksanaan MapReduce: Sekarang kita akan membahas tentang penerapan model MapReduce menggunakan platform pemrograman Java. Berikut ini adalah komponen yang berbeda dari seluruh ujung ke ujung pelaksanaan.
- The program klien yang merupakan kelas driver dan memulai proses
- The fungsi peta yang melakukan split dengan menggunakan pasangan kunci-nilai.
- The mengurangi fungsi yang menggabungkan data diolah dan mengirim output kembali ke klien.
kelas driver: Berikut ini adalah kelas driver yang mengikat Peta dan Mengurangi fungsi dan mulai pengolahan. Ini adalah program klien yang memulai proses.
Listing1: Program klien (kelas driver) memulai proses
[kode]
paket com.mapreduce.techalpine;
impor org.apache.commons.logging.Log;
impor org.apache.commons.logging.LogFactory;
impor org.apache.hadoop.conf.Configuration;
impor org.apache.hadoop.fs.Path;
impor org.apache.hadoop.io.Text;
impor org.apache.hadoop.mapreduce.Job;
impor org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
impor org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
impor org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
impor org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
/**
* @author kaushik pal
*
* Ini adalah kelas pendorong utama untuk memulai peta-mengurangi
* process. Ini menetapkan tanah kembali untuk seluruh proses dan
* Kemudian mulai itu.
*/
public class DevXDriver {
public static void main(Senar[] args) throws Exception {
// memulai konfigurasi
Konfigurasi configx = konfigurasi baru();
// Tambahkan file resource
configx.addResource(Jalur baru(“/user / Hadoop / inti-site.xml”));
configx.addResource(Jalur baru(“/user / Hadoop / HDFS-site.xml”));
// Buat MapReduce Pekerjaan
devxmapjob pekerjaan = new job(configx,”DevXDriver.class”);
devxmapjob.setJarByClass(DevXDriver.class);
devxmapjob.setJobName(“DevX MapReduce Pekerjaan”);
// Set output kay dan kelas nilai
devxmapjob.setOutputKeyClass(Text.class);
devxmapjob.setOutputValueClass(Text.class);
// Set kelas Peta
devxmapjob.setMapperClass(DevXMap.class);
// Set kelas Combiner
devxmapjob.setCombinerClass(DevXReducer.class);
// Set kelas Reducer
devxmapjob.setReducerClass(DevXReducer.class);
// Set Peta kunci output dan nilai kelas
devxmapjob.setMapOutputKeyClass(Text.class);
devxmapjob.setMapOutputValueClass(Text.class);
// Mengatur jumlah tugas peredam
devxmapjob.setNumReduceTasks(10);
// Mengatur input dan output kelas Format
devxmapjob.setInputFormatClass(TextInputFormat.class);
devxmapjob.setOutputFormatClass(TextOutputFormat.class);
// Set input dan jalur output
FileInputFormat.addInputPath(devxmapjob, Jalur baru(“/user / map_reduce / input /”));
FileOutputFormat.setOutputPath(devxmapjob,Jalur baru(“/user / map_reduce / output”));
// Mulai MapReduce Pekerjaan
devxmapjob.waitForCompletion(benar);
}
}
[/kode]
fungsi peta: Ini bertanggung jawab untuk memisahkan data berdasarkan pasangan kunci-nilai. Hal ini dikenal sebagai pemetaan data.
Listing2: Ini adalah fungsi Peta membagi data ke dalam potongan
[kode]
paket com.mapreduce.techalpine;
import java.io.BufferedReader;
impor java.io.InputStreamReader;
impor java.net.URI;
impor java.util.StringTokenizer;
impor org.apache.commons.logging.Log;
impor org.apache.commons.logging.LogFactory;
impor org.apache.hadoop.conf.Configuration;
impor org.apache.hadoop.fs.FileSystem;
impor org.apache.hadoop.fs.Path;
impor org.apache.hadoop.io.LongWritable;
impor org.apache.hadoop.io.Text;
impor org.apache.hadoop.mapreduce.Mapper;
/**
* @author kaushik pal
*
* Ini adalah proses peta. Ia melakukan pemetaan untuk pasangan kunci-nilai.
*/
public class DevXMap meluas Mapper<LongWritable, Teks, Teks,Teks> {
// Buat Jalur, BufferedReader dan Teks variabel
jalan file_path;
BufferedReader buffer_reader;
tweet_values teks = teks baru();
/**
* kunci @param
* nilai @param
* @param konteks
*/
peta public void(kunci LongWritable, nilai teks, konteks konteks) {
coba{
// Buat konfigurasi untuk Peta
Konfigurasi map_config = konfigurasi baru();
// Memuat file inti Hadoop dalam konfigurasi
map_config.addResource(Jalur baru(“/user / Hadoop / inti-site.xml”));
map_config.addResource(Jalur baru(“/user / Hadoop / HDFS-site.xml”));
// Buat variabel
String searchkeyword = “”;
// Membuka file dari path file
file_path = Jalur baru(“file / repositori / keys.txt”);
FileSystem file_system = FileSystem.get(URI.create(“file / repositori / keys.txt”),Konfigurasi baru());
// Beban penyangga reader
buffer_reader = BufferedReader baru(InputStreamReader baru(file_system.open(file_path)));
while(buffer_reader.ready())
{
searchkeyword = buffer_reader.readLine().memangkas();
}
// Dapatkan nilai kunci
akhir Text key_value = Teks baru(Cari Kata Kunci);
// Periksa nilai dan mengambil keputusan
jika(Nilai == null)
{
return;
}
lain{
StringTokenizer string_tokens = StringTokenizer baru(value.toString(),”,”);
count int = 0;
while(string_tokens.hasMoreTokens()) {
count ++;
jika(count <= 1)
terus;
String new_tweet_value = string_tokens.nextToken().toLowerCase().memangkas().menggantikan semua(“\\*”,””);
jika(new_tweet_value.contains(searchkeyword.toLowerCase().memangkas())) {
tweet_values.set(new_tweet_value);
context.write(key_value,tweet_values);
}
}
}
}
menangkap(Exception e){
e.printStackTrace();
}
}
}
[/kode]
mengurangi fungsi: Ini bertanggung jawab untuk menggabungkan data. agregasi tersebut dilakukan berdasarkan nilai-nilai kunci. Jadi setelah pengolahan dan pemilahan agregasi selesai dan mengirimkan hasilnya kembali ke program klien.
Listing3: Fungsi Mengurangi mengumpulkan data diolah
[kode]
paket com.mapreduce.techalpine;
import java.io.BufferedReader;
impor java.io.IOException;
impor java.io.InputStreamReader;
impor java.net.URI;
impor java.util.RandomAccess;
impor java.util.regex.Matcher;
impor java.util.regex.Pattern;
impor org.apache.commons.logging.Log;
impor org.apache.commons.logging.LogFactory;
impor org.apache.hadoop.conf.Configuration;
impor org.apache.hadoop.fs.FSDataOutputStream;
impor org.apache.hadoop.fs.FileSystem;
impor org.apache.hadoop.fs.Path;
impor org.apache.hadoop.io.Text;
impor org.apache.hadoop.mapreduce.Reducer;
/**
* @author kaushik pal
*
* Ini adalah fungsi peredam. Agregat output berdasarkan
* menyortir pasangan kunci-nilai.
*/
public class DevXReducer meluas Reducer<Teks ,Teks,Teks,Teks>
{
// Buat variabel untuk path file
jalan positive_file_path;
jalan negative_file_path;
jalan output_file_path;
jalan keyword_file_path;
// Buat variabel untuk penyangga
BufferedReader positive_buff_reader;
BufferedReader negative_buff_reader;
BufferedReader keyword_buff_reader;
// Membuat variabel untuk perhitungan
statis Dua total_record_count = Dua baru(“0”);
statis Dua count_neg = Dua baru(“0”);
statis count_pos Dua = Dua baru(“0”);
statis Dua count_neu = Dua baru(“0”);
statis Dua percent_neg = Dua baru(“0”);
statis percent_pos Dua = Dua baru(“0”);
statis Dua percent_neu = Dua baru(“0”);
pola pattrn_matcher;
pertandingan matcher_txt;
static int new_row = 0;
FSDataOutputStream out_1st,out_2nd;
/**
* kunci @param
* nilai @param
* @param konteks
* @throws IOException
* @throws InterruptedException
*/
public void mengurangi(kunci teks, iterable<Teks> values,konteks konteks) throws IOException, InterruptedException
{
// Buat konfigurasi untuk reducer
Konfigurasi reduce_config = konfigurasi baru();
// Beban Hadoop file konfigurasi
reduce_config.addResource(Jalur baru(“/user / Hadoop / inti-site.xml”));
reduce_config.addResource(Jalur baru(“/user / Hadoop / HDFS-site.xml”));
// Buat variabel
String key_word = “”;
String check_keyword = key_word;
keyword_file_path = Jalur baru(“file / repositori / keys.txt”);
FileSystem file_system_read = FileSystem.get(URI.create(“file / repositori / keys.txt”),Konfigurasi baru());
keyword_buff_reader = BufferedReader baru(InputStreamReader baru(file_system_read.open(keyword_file_path)));
FileSystem get_filesys = FileSystem.get(reduce_config);
FileSystem get_filesys_posneg = FileSystem.get(reduce_config);
Jalan path_output = jalan baru(“/user / sentiment_output_file.txt”);
Jalan path_output_posneg = jalan baru(“/user / posneg_output_file.txt”);
// Dapatkan kata kunci
while(keyword_buff_reader.ready())
{
key_word = keyword_buff_reader.readLine().memangkas();
}
// Periksa sistem file
jika (!get_filesys.exists(path_output)) {
out_1st = get_filesys.create(path_output);
out_2nd = get_filesys_posneg.create(path_output_posneg);
}
// Periksa pencocokan kata kunci menggunakan kamus positif dan negatif
jika(check_keyword.equals(key.toString().toLowerCase()))
{
untuk(new_tweets teks:values)
{
// Memuat kamus kata positif
positive_file_path = Jalur baru(“/user / map_reduce / pos_words.txt”);
FileSystem filesystem_one = FileSystem.get(URI.create(“file / pos_words.txt”),Konfigurasi baru());
positive_buff_reader = BufferedReader baru(InputStreamReader baru(filesystem_one.open(positive_file_path)));
// Memuat kata disctinary negatif
negative_file_path = Jalur baru(“/user / map_reduce / neg_words.txt”);
FileSystem filesystem_two = FileSystem.get(URI.create(“file / neg_words.txt”),Konfigurasi baru());
negative_buff_reader = BufferedReader baru(InputStreamReader baru(filesystem_two.open(negative_file_path)));
++total_record_count;
boolean first_flag = false;
boolean second_flag = false;
String all_tweets = new_tweets.toString();
String first_regex = “”;
String second_regex = “”;
while(positive_buff_reader.ready())
{
first_regex = positive_buff_reader.readLine().memangkas();
new_row ;
pattrn_matcher = Pattern.compile(first_regex, Pattern.CASE_INSENSITIVE);
matcher_txt = pattrn_matcher.matcher(all_tweets);
first_flag = matcher_txt.find();
jika(first_flag)
{
out_2nd.writeBytes(all_tweets);
context.write(Teks baru(first_regex),Teks baru(all_tweets));
istirahat;
}
}
while(negative_buff_reader.ready())
{
new_row ;
second_regex = negative_buff_reader.readLine().memangkas();
pattrn_matcher = Pattern.compile(second_regex, Pattern.CASE_INSENSITIVE);
matcher_txt = pattrn_matcher.matcher(all_tweets);
second_flag = matcher_txt.find();
jika(second_flag)
{
out_2nd.writeBytes(all_tweets);
context.write(Teks baru(second_regex),Teks baru(all_tweets));
istirahat;
}
}
jika(first_flag&second_flag)
{
++count_neu;
}
lain
{
jika(first_flag)
{
++count_pos;
}
jika(second_flag)
{
++count_neg;
}
jika(first_flag == false&second_flag == false)
{
++count_neu;
}
}
// Tutup buffer
negative_buff_reader.close();
positive_buff_reader.close();
}
// Menghitung nilai persen
percent_pos = count_pos / total_record_count * 100;
percent_neg = count_neg / total_record_count * 100;
percent_neu = count_neu / total_record_count * 100;
coba{
// Menulis ke file
out_1st.writeBytes(“\n”+key_word);
out_1st.writeBytes(“,”+total_record_count);
out_1st.writeBytes(“,”+percent_neg);
out_1st.writeBytes(“,”+percent_pos);
out_1st.writeBytes(“,”+percent_neu);
// Tutup file sistem
out_1st.close();
get_filesys.close();
}menangkap(Exception e){
e.printStackTrace();
}
}
}
}
[/kode]
Kesimpulan: Pada artikel ini saya telah membahas pengolahan MapReduce menggunakan lingkungan pemrograman Java. Komponen yang berbeda seperti Peta dan Mengurangi fungsi melakukan tugas utama dan mengembalikan output ke klien. pemrosesan melakukan efisien pada lingkungan terdistribusi hanya. Jadi kita harus menyiapkan kerangka Apache Hadoop pada lingkungan terdistribusi untuk mendapatkan hasil yang terbaik.
Harap Anda menikmati artikel dan Anda akan dapat menerapkannya dalam pemrograman praktis Anda. Keep reading.
