BigData: Hadoop

4.1 Hadoop haqida umumiy ma'lumot

MapReduce paradigmasini Google dasturi 2004 yilda o'z maqolasida taklif qildi MapReduce: Simplified Data Processing on Large Clusters. Ushbu maqola paradigma tavsifini o'z ichiga olgani sababli, lekin amalga oshirilmagan – Yahoo kompaniyasidan bir necha dasturchilar nutch veb-kravler ustida ishlayotgan paytda o'z versiyalarini taklif qilishgan. Hadoop tarixini batafsilroq maqolada o'qishingiz mumkin Hadoop | History or Evolution.

Dastlab Hadoop, asosan, ma'lumotlarni saqlash va MapReduce vazifalarini ishga tushirish uchun asbob edi, hozirda esa Hadoop katta ma'lumotlarni ishlash bilan bog'liq texnologiyalar yig'indisi bo'lib, faqat MapReduce orqali emas.

Hadoop'ning asosiy (core) komponentlari:

• Hadoop Distributed File System (HDFS) – deyarli cheklanmagan hajmdagi ma'lumotlarni saqlash imkonini beruvchi tarqatilgan fayllar tizimi.
• Hadoop YARN – klaster resurslarinı boshqarish va vazifalarni boshqarish uchun freymvork, shuningdek, MapReduce freymvorkini o'z ichiga oladi.
• Hadoop common

Hadoop bilan bevosita bog'liq bo'lgan, lekin Hadoop core'ga kirmaydigan ko'plab loyihalar ham mavjud:

• Hive – katta ma'lumotlar ustida SQL-like so'rovlar uchun asbob (SQL so'rovlarini MapReduce vazifalariga aylantiradi);
• Pig – yuqori darajadagi ma'lumotlarni tahlil qilish tili. Ushbu tilning bir qatorli kodi MapReduce vazifalar ketma-ketligiga aylanishi mumkin;
• Hbase – BigTable paradigmasini amalga oshiruvchi ustunli baza;
• Cassandra – yuqori samarali tarqatilgan key-value baza;
• ZooKeeper – konfiguratsiyalarni tarqatilgan saqlash va o'zgartirishlarni sinxronizatsiya qilish servisi;
• Mahout – katta ma'lumotlar ustida mashina o'rganish kutubxonasi va motor.

Alohida o'ziga e'tibor qaratish kerak bo'lgan loyiha Apache Spark, bu tarqatilgan ma'lumotlarni ishlash uchun motor hisoblanadi. Apache Spark odatda Hadoop komponentlari, masalan, HDFS va YARN'ni ishlatib o'z ishini amalga oshiradi, ammo o'zi o'zi hozirgi paytda Hadoop'dan mashhurlashib boryapti:

Ushbu komponentlarning bir qismi ushbu materiallar tsikli maqolalari orqali batafsil yoritiladi, hozircha Hadoop bilan qanday ishlashni boshlash va uni amalda qo'llashni o'rganamiz.

4.2 Hadoop'da MapReduce dasturlarini ishga tushirish

Endi, Hadoop'da MapReduce vazifasini qanday ishga tushirishni ko'ramiz. Vazifa sifatida klassik WordCount misolidan foydalanamiz, bu o'tgan ma'ruzada ko'rib chiqilgan edi.

Vazifaning formulasi: hujjatlar to'plami mavjud. Hujjatlar to'plamida uchraydigan har bir so'z uchun, so'zning necha marta to'plamda uchraganini sanash kerak.

Yechim:

Map hujjatni so'zlarga bo'ladi va (word, 1) juftligi to'plamini qaytaradi.

Reduce har bir so'zning kirinishlarini yig'adi:

def map(doc):  
for word in doc.split():  
	yield word, 1 

def reduce(word, values):  
	yield word, sum(values)

Endi bu yechimni kod sifatida programa shaklida yozish kerak, uni Hadoop'da ishga tushirish mumkin bo'ladi.

4.3 Usul №1. Hadoop Streaming

Hadoop'da MapReduce-programmani ishga tushirishning eng oson yo'li - Hadoop streaming-interfeysidan foydalanishdir. Streaming-interfeys, map va reduce dasturlar ko'rinishida amalga oshirilishini, ular stdin'dan ma'lumotlarni qabul qilib, stdout'ga natija chiqarishini talab qiladi.

Map funktsiyani bajaruvchi dastur mapper deb ataladi. Reduce funktsiyani bajarayotgan dastur, mos ravishda, reducer deb ataladi.

Streaming interfeysi odatda bir kiruvchi qatorni mapper yoki reducer'da map uchun bir kiruvchi yozuv sifatida qabul qiladi.

Mapper'ning chiqishi reducer'ning kirishi sifatida (kalit, qiymat) juftliklar shaklida keladi, bu juftliklarning barchasi bir kalit uchun:

• Bitta reducer'ning ishga tushirilishda qayta ishlanishi kafolatlanadi;
• Kiruvchi ketma-ket (ya'ni, agar bir reducer turli kalitlarni qayta ishlasa – kirish kaliti bo'yicha guruhlangan bo'ladi).

Endi mapper va reducer'ni python tilida amalga oshiramiz:

#mapper.py  
import sys  
  
def do_map(doc):  
for word in doc.split():  
	yield word.lower(), 1  
  
for line in sys.stdin:  
	for key, value in do_map(line):  
    	print(key + "\t" + str(value))  
 
#reducer.py  
import sys  
  
def do_reduce(word, values):  
	return word, sum(values)  
  
prev_key = None  
values = []  
  
for line in sys.stdin:  
	key, value = line.split("\t")  
	if key != prev_key and prev_key is not None:  
    	result_key, result_value = do_reduce(prev_key, values)  
    	print(result_key + "\t" + str(result_value))  
    	values = []  
	prev_key = key  
	values.append(int(value))  
  
if prev_key is not None:  
	result_key, result_value = do_reduce(prev_key, values)  
	print(result_key + "\t" + str(result_value))

Hadoop tomonidan qayta ishlanishi kerak bo'lgan ma'lumotlar HDFS'da saqlanishi kerak. Bizning maqolalarni yuklab olish va HDFS'da saqlash uchun hadoop fs komandasidan foydalanish kerak:

wget https://www.dropbox.com/s/opp5psid1x3jt41/lenta_articles.tar.gz  
tar xzvf lenta_articles.tar.gz  
hadoop fs -put lenta_articles 

Hadoop fs utilitasi fayl tizimi bilan manipulyatsiya qilish uchun ko'plab usullarni qo'llab-quvvatlaydi, ko'plari standart linux utilitalariga mos keladi.

Endi streaming-vazifani ishga tushiramiz:

yarn jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar\  
 -input lenta_articles\  
 -output lenta_wordcount\  
 -file mapper.py\  
 -file reducer.py\  
 -mapper "python mapper.py"\  
 -reducer "python reducer.py" 

Yarn utilitasi turli xil dasturlarni (shu jumladan map-reduce asoslangan) klasterda ishga tushirish va boshqarish uchun mo'ljallangan. Hadoop-streaming.jar – bu aynan shunday yarn-dasturining misolidir.

Keyin ishga tushirish parametrlari keladi:

• input – hdfs'dagi dastlabki ma'lumotlar papkasi;
• output – hdfs'da natijani qo'yish kerak bo'lgan papka;
• file – map-reduce vazifa ishlash jarayonida kerak bo'lgan fayllar;
• mapper – map-bosqichi uchun ishlatiladigan konsol komanda;
• reduce – reduce-bosqichi uchun ishlatiladigan konsol komanda.

Ishga tushirgandan so'ng konsolda vazifaning bajarilish jarayoni va vazifa haqida batafsil ma'lumot ko'rish uchun URL mavjud bo'ladi.

Ushbu URL orqali mavjud interfeysda vazifaning batafsil holatini bilib olish, har bir mapper va reducer'ning loglarini ko'rish mumkin (bu vazifalar tushib ketgan bo'lsa juda foydali).

Yaxshi bajarilganidan so'ng, natija HDFS'da output maydonida ko'rsatilgan papkaga qo'yiladi. Uning mazmunini "hadoop fs -ls lenta_wordcount" komandasi yordamida ko'rish mumkin.

O‘zingiz natijani quyidagi tarzda olishingiz mumkin:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5  
с	
41  
что	
43  
на	
82  
и	
111  
в	
194 

Hadoop fs -text komanda papkaning mazmunini matn ko'rinishida beradi. Men natijani so'zlar kirishlari soniga ko'ra tartibladim. Kutilmagandek, eng ko'p uchraydigan so'zlar - predloglar.

4.4 Usul №2: Java ishlatamiz

Hadoop o'zi java'da yozilgan va native interfeysi ham java-based. Java uchun native wordcount dasturini qanday ko'rinishini ko'rsatamiz:

import java.io.IOException;  
import java.util.StringTokenizer;  
  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.mapreduce.Mapper;  
import org.apache.hadoop.mapreduce.Reducer;  
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
  
public class WordCount {  
  
	public static class TokenizerMapper  
        	extends Mapper<Object, Text, Text, IntWritable>{  
  
    	private final static IntWritable one = new IntWritable(1);  
    	private Text word = new Text();  
  
    	public void map(Object key, Text value, Context context  
    	) throws IOException, InterruptedException {  
        	StringTokenizer itr = new StringTokenizer(value.toString());  
        	while (itr.hasMoreTokens()) {  
            	word.set(itr.nextToken());  
            	context.write(word, one);  
        	}  
    	}  
	}  
  
	public static class IntSumReducer  
        	extends Reducer<Text,IntWritable,Text,IntWritable> {  
    	private IntWritable result = new IntWritable();  
  
    	public void reduce(Text key, Iterable
      
  
      
  
    values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_articles")); FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_wordcount")); System.exit(job.waitForCompletion(true) ? 0 : 1); } } 
  

Ushbu klass aynan Python misolida ko'rgazilgan kodni bajaradi. Biz TokenizerMapper va IntSumReducer klasslarini yaratamiz, ularni Mapper va Reducer klasslaridan meros qilib olamiz. Shablon parametrlarida kiritilgan klasslar kirish va chiqarish qiymatlarining turlarini belgilaydi. Native API funktsiyasi, map kirishi sifatida kalit-qiymat juftligi olinadi. Bizning holatda kalit bo'sh – shuning uchun kalit turini shunchaki Object sifatida belgiladik.

Main metodida mapreduce vazifasini yaratamiz va uning parametrlarini belgilaymiz – nomi, mapper va reducer, HDFS'dagi kirish ma'lumotlar joyi va natijani qo'yadigan joy. Kompilyatsiya qilish uchun bizga hadoop kutubxonalari kerak bo'ladi. Men build uchun Maven ishlataman, bu uchun cloudera maven uchun repozitoriy bor. Uni sozlash bo'yicha ko'rsatmalarni link orqali topishingiz mumkin. Natijada, men uchun pom.xmp fayl (maven tomonidan loyiha build'ni tavsiflash uchun ishlatiladigan fayl) quyidagi kabi bo'ldi):

<?xml version="1.0" encoding="UTF-8"?>  
<project xmlns="http://maven.apache.org/POM/4.0.0"  
     	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
     	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
	<modelVersion>4.0.0</modelVersion>  
  
	<repositories>  
    	<repository>  
        	<id>cloudera</id>  
        	<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>  
    	</repository>  
	</repositories>  
  
	<dependencies>  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-common</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-auth</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-hdfs</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-mapreduce-client-app</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
	</dependencies>  
  
	<groupId>org.dca.examples</groupId>  
	<artifactId>wordcount</artifactId>  
	<version>1.0-SNAPSHOT</version>  
 
</project>

Loyiha jar-paketini to'plang:

mvn clean package

Jar-faylda loyiha yig'ilgandan so'ng ishga tushirish streaming interfeysi holatidagi kabi amalga oshiriladi:

yarn jar wordcount-1.0-SNAPSHOT.jar  WordCount 

Bajarishni kutib turing va natijani tekshiring:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5  
с	
41  
что	
43  
на	
82  
и	
111  
в	
194 

Xulosa qilib aytganda, bizning native dasturimizning natijasi avvalgi usulda streaming dasturini ishga tushirganimizdagi natija bilan teng.