BigData: MapReduce-ilovalarni ishlab chiqish usullari va strategiyalari

5.1 Map only job

MapReduce’ni samarali ishlatish uchun turli usullarni tushuntirish va Hadoop’ning ba’zi xususiyatlarini koʻrsatish vaqti keldi, bu esa klasterda MapReduce-vazifasini ishlab chiqishni osonlashtiradi yoki tezligini oshiradi.

Esimizda bo‘lsa, MapReduce Map, Shuffle va Reduce bosqichlaridan iborat. Odatda, amaliy vazifalarda eng qiyin bosqich Shuffle hisoblanadi, chunki aynan shu bosqichda maʼlumotlar tartiblanadi. Aslida, faqat Map bosqichi bilan hal qilinadigan qator vazifalar mavjud. Mana shunday vazifalarga misollar:

• Maʼlumotlarni filtrlash (masalan, web-server loglarida "123.123.123.123 IP-manzilidan barcha yozuvlarni topish");
• Maʼlumotlarni oʻzgartirish ("csv-loglarda ustunni oʻchirib tashlash");
• Tashqi manbalardan maʼlumotlarni yuklash va yuklash ("Logdagi barcha yozuvlarni maʼlumotlar bazasiga qoʻshish").

Bunday vazifalar Map-Only yordamida yechiladi. Hadoopda Map-Only vazifasini yaratishda noldan kamaytiruvchilar sonini koʻrsatish kerak:

Hadoopda map-only vazifasining konfiguratsiya namunasi:

job.setNumReduceTasks(0); 

hadoop jar hadoop-streaming.jar \ 
 -D mapred.reduce.tasks=0\ 
-input input_dir\ 
-output output_dir\ 
-mapper "python mapper.py"\ 
-file "mapper.py"

Map-Only jobs aslida juda foydali boʻlishi mumkin. Misol uchun, Facetz.DCA platformasida foydalanuvchilarning xatti-harakatlariga qarab ularning xususiyatlarini aniqlash uchun aynan bir katta map-only ishlatiladi, har bir mapper foydalanuvchini qabul qilib, kirish uchun xarakteristikasini beradi.

5.2 Combine

Odatda, Map-Reduce vazifasini bajarishda eng qiyin bosqich Shuffle bosqichi ekanligini yozdim. Buning sababi, oraliq natijalar (mapper chiqishi) diskka yozilib, tartiblanadi va tarmoq orqali uzatiladi. Biroq, ushbu xatti-harakat mantiqsiz koʻringan vazifalar mavjud. Masalan, hujjatlarda so'zlarni sanash vazifasida bir necha mapper’larning natijalarini oldindan agregatlashimiz va kamaytiruvchi (reducer) ga har bir mashina bo'yicha allaqachon yig’ilgan qiymatlarni uzatishimiz mumkin.

Hadoopda buni amalga oshirish uchun kombinatsiya qilish funksiyasini aniqlash mumkin, bu mapper’lar chiqishini qayta ishlaydi. Kombinatsiya qilish funksiyasi reduce-ga juda oʻxshaydi - u mapper'lar chiqishini qabul qiladilar va bu mapper’lar uchun agregat natija chiqaradi, shuning uchun ko'pincha kamaytiruvchi (reducer) ni kombinator sifatida ishlatishadi.

Muqarrarlik, kombinatsiya qilish funksiyasi barcha qiymatlar uchun bajarilishini Hadoop kafolatlamaydi. Masalan, kalit bo'yicha median qiymatini qidirishda u har doim qo'llanilishi mumkin emas. Lekin qo‘llaniladigan vazifalarda uni ishlatish MapReduce-vazifasini bajarishini sezilarli darajada tezlashtirishi mumkin.

Hadoopda Combiner’ni qo'llash:

job.setMapperClass(TokenizerMapper.class); 
job.setCombinerClass(IntSumReducer.class); 
job.setReducerClass(IntSumReducer.class); 

hadoop jar hadoop-streaming.jar \ 
-input input_dir\ 
-output output_dir\ 
-mapper "python mapper.py"\ 
-reducer "python reducer.py"\ 
-combiner "python reducer.py"\ 
-file "mapper.py"\ 
-file "reducer.py"\

5.3 Цепочки MapReduce-задач

Ba'zida bir MapReduce yordamida vazifani hal qilishning iloji bo'lmaydi. Masalan, biroz o'zgartirilgan WordCount vazifasini koʻrib chiqaylik: bir qator matnli hujjatlar mavjud, to'plamda 1 tadan 1000 gacha so'zlar necha marta uchraganligini, 1001 dan 2000 gacha, 2001 dan 3000 gacha va hokazo so'zlar sanalishi kerak. Ushbu masalani hal qilish uchun 2 MapReduce job’lari kerak:

• Har bir so‘zni qaysi intervalga tushganini hisoblaydigan o'zgartirilgan wordcount;
• Birinchi MapReduce chiqishida har bir intervallar sonini hisoblaydigan MapReduce.

Yecha tanqidiy kod:

#map1 
def map(doc): 
for word in doc: 
yield word, 1

#reduce1 
def reduce(word, values): 
yield int(sum(values)/1000), 1 

#map2 
def map(doc): 
interval, cnt = doc.split() 
yield interval, cnt 

#reduce2 
def reduce(interval, values): 
yield interval*1000, sum(values) 

Hadoopda ketma-ket MapReduce-vazifalarini bajarish uchun birinchi vazifa uchun chiqish deb berilgan papkani ikkinchisiga kirish ma'lumotlari sifatida koʻrsatish va ularni ketma-ket ishga tushirish kifoya qiladi.

Amaliyotda, MapReduce zanjirlari koʻp hollarda murakkab ketma-ketliklarni ifodalaydi, unda MapReduce-vazifalari bir-biriga ketma-ket yoki parallel tarzda ulanishi mumkin. Ushbu vazifalarni boshqarishni soddalashtirish uchun oozie va luigi kabi alohida vositalar mavjud.

5.4 Distributed cache

Hadoopdagi muhim mexanizmlardan biri Distributed Cache hisoblanadi. Distributed Cache, MapReduce-vazifasining bajarilish muhiti bilan fayllar qo'shishga imkon beradi (masalan, matnli fayllar, arxivlar, jar-fayllar).

HDFSda saqlanayotgan fayllarni, mahalliy fayllarni (vazifani ishga tushirish joyidan mahalliy) qo'shish mumkin. Men allaqachon Distributed Cache-ni hadoop streaming bilan qanday ishlatishni ko'rsatuvchi -file opsiyasi orqali mapper.py va reducer.py fayllarini qo'shish orqali ko'rsatib o'tdim. Aslida mapper.py va reducer.py dan boshqa har qanday fayllarni qo'shishingiz mumkin va keyin ularni mahalliy papkada mavjuddek foydalanishingiz mumkin.

Distributed Cache-dan foydalanish:

//Job’ning konfiguratsiyasi 
JobConf job = new JobConf(); 
DistributedCache.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"),  job); 
DistributedCache.addCacheArchive(new URI("/myapp/map.zip", job); 
DistributedCache.addFileToClassPath(new Path("/myapp/mylib.jar"), job); 
DistributedCache.addCacheArchive(new URI("/myapp/mytar.tar", job); 
DistributedCache.addCacheArchive(new URI("/myapp/mytgz.tgz", job); 
 
//mapper-da foydalanish namunasi: 
public static class MapClass extends MapReduceBase   
implements Mapper<K, V, K, V> { 
 
 private Path[] localArchives; 
 private Path[] localFiles; 
  
 public void configure(JobConf job) { 
   // arxivlardan kechilgan ma'lumotlarni olish 
   File f = new File("./map.zip/some/file/in/zip.txt"); 
 } 
  
 public void map(K key, V value,  
             	OutputCollector<K, V> output, Reporter reporter)  
 throws IOException { 
   // shu yerda ma'lumotlar ishlatiladi 
   // ... 
   // ... 
   output.collect(k, v); 
 } 
}

yarn  hadoop-streaming.jar\ 
-files mapper.py,reducer.py,some_cached_data.txt\ 
-input '/some/input/path' \ 
-output '/some/output/path' \  
-mapper 'python mapper.py' \ 
-reducer 'python reducer.py' \

import sys 
#faqat faylni mahalliy papkadan oʻqish 
data = open('some_cached_data.txt').read() 
 
for line in sys.stdin() 
#inputni qayta ishlash 
#shu yerda ma'lumotlarni ishlatish

5.5 Reduce Join

Relatsion ma'lumotlar bazalari bilan ishlashga odatlanganlar ba'zida ma'lum jadvalni bir xil kalit bo‘yicha birgalikda qayta ishlash imkonini beruvchi juda qulay Join operatsiyasidan foydalanishadi. Katta ma'lumotlar bilan ishlashda ham shunday masalalar yuzaga keladi. Quyidagi misolni ko‘rib chiqamiz:

Ikkita web-server loglari mavjud bo‘lib, har bir log quyidagi ko‘rinishda:

t\t

Logning bir parchasi:

1446792139	
178.78.82.1	
/sphingosine/unhurrying.css 
1446792139	
126.31.163.222	
/accentually.js 
1446792139	
154.164.149.83	
/pyroacid/unkemptly.jpg 
1446792139	
202.27.13.181	
/Chawia.js 
1446792139	
67.123.248.174	
/morphographical/dismain.css 
1446792139	
226.74.123.135	
/phanerite.php 
1446792139	
157.109.106.104	
/bisonant.css

Har bir IP-manzil uchun bu manzil qaysi serverga ko'proq tashrif buyurganini hisoblash kerak. Natija quyidagi tarzda taqdim etilishi kerak:

\t

Natijaning bir qismi:

178.78.82.1	
first 
126.31.163.222	
second 
154.164.149.83	
second 
226.74.123.135	
first

Afsuski, umuman olganda, ma'lumotlar bazalarida ikki logni kalit bo'yicha qo'shib birlashtirish (bizning holatda - IP-manzil bo'yicha) juda murakkab bir operatsiya hisoblanadi va 3 MapReduce va Reduce Join namunasi yordamida hal qilinadi:

ReduceJoin quyidagicha ishlaydi:

1) Har bir kiruvchi log uchun alohida MapReduce (Map only) ishga tushiriladi, kiruvchi ma'lumotlarni quyidagi ko'rinishga o'zgartiradi:

key -> (type, value)

Bu yerda key - jadvalni birlashtirish kerak boʻlgan kalit, Type - jadval turi (bizning holatda birinchi yoki ikkinchi), Value - kalitga bog‘liq boʻlgan har qanday qo‘shimcha ma‘lumotlar.

2) Ikkala MapReduce chiqishlari 3-MapReduce-ga kirish sifatida uzatiladi, bu esa birlashmani bajaradi. Ushbu MapReduce kiruvchi ma’lumotlarni nusxa ko‘chiruvchi bo‘sh Mapper’ni o'z ichiga oladi. Keyin shuffle ma'lumotlarni kalitlarga qarab joylashtiradi va ularni quyidagi koʻrinishda reducer’ga beradi:

key -> [(type, value)]

Muhim jihat, bu paytda reducer’ga ikkala logdan yozuvlar kelib tushadi va type maydoni ma’lum bir qiymatni qaysi logdan kelib tushganini aniqlash imkonini beradi. Demak, dastlabki vazifani hal qilish uchun kifoya qilinadigan ma’lumotlar mavjud. Bizning holatda reducer har bir kalit uchun qaysi type ko'proq uchraganini hisoblab, ushbu type’ni chiqarishi kerak.

5.6 MapJoin

ReduceJoin namunasi ikki logni kalit bo‘yicha umumiy birlashtirish holatini tasvirlaydi. Biroq, vazifani sezilarli darajada soddalashtirish va tezlashtirish mumkin bo'lgan maxsus holat mavjud. Bu holat, loglardan biri boshqa logga nisbatan ancha kichik o'lchamga ega bo'lgan holat. Quyidagi vazifani koʻrib chiqaylik:

Ikkita loglar mavjud. Birinchi log web-server logini oʻz ichiga oladi (avvalgi masalada boʻlganidek), ikkinchi fayl (100 kB) URL->tematika mosligini oʻz ichiga oladi. 2-faylning namunasi:

/toyota.php 	
auto 
/football/spartak.html 	
sport 
/cars 	
auto 
/finances/money 	
business

Har bir IP-manzil uchun ushbu IP-manzildan qaysi kategoriyadan sahifalar ko'proq yuklanganligini hisoblash kerak.

Ushbu holatda, biz URL bo'yicha 2 logni qo'shishimiz kerak. Biroq, bu holatda biz 3 MapReduce ishga tushirish shart emas, chunki ikkinchi log to'liq xotiraga sigʻishi mumkin. 1 MapReduce yordamida vazifani hal qilish uchun biz ikkinchi logni Distributed Cache-ga yuklab, mapper’ni ishga tushirishda uni xotiraga oqishi mumkinmiz, bu bilan ularni key -> topic’ga yunaltirish mumkin.

Keyin vazifa quyidagicha yechiladi:

Map:

# birinchi logidagi har bir sahifa tematikasini topish 
input_line -> [ip, topic]

Reduce:

Ip -> [topics] -> [ip, most_popular_topic]

Reduce ip manzilini va barcha tematikani ro‘yxatini qabul qiladi va eng ko‘p uchragan tematikani aniqlaydi. Shunday qilib, vazifa 1 MapReduce yordamida hal qilinadi va Join'ning o'zi asosan map ichida amalga oshiriladi (shuning uchun, agar qo'shimcha kalit bo'yicha agregatsiya kerak bo'lmasa - MapOnly job’ni ishga tushirish kifoya):