Batch processing

Təsəvvür edin ki, e-ticarət saytınızın web server-ləri gündə bir neçə yüz gigabayt access log yazır. Marketinq komandası soruşur: "Keçən ay ən çox baxılan beş məhsul səhifəsi hansıdır?" Bu sualın cavabı heç bir istifadəçinin sorğusu deyil, dərhal lazım deyil və tək bir request-in nəticəsi də deyil. O, böyük həcmdə mövcud data-nın üzərindən keçib yeni bir data parçası çıxarmaqdır. Kitabın əvvəlki fəsillərində sistemlər əsasən online rejimdə, yəni sorğu gəlir, cavab gedir məntiqi ilə baxılırdı. Bu fəsil fərqli bir rejimə, batch processing-ə keçir.

Sistemləri üç kateqoriyaya bölmək faydalıdır. Service-lər request gözləyir və mümkün qədər tez cavab verir; əsas metrik response time-dır. Batch processing sistemləri böyük input-u götürür, onu emal edib output yaradır; əsas metrik throughput-dur, job saatlarla çəkə bilər. Stream processing sistemləri ikisinin arasındadır: event-lər gələn kimi emal olunur, lakin nəticə request-ə cavab deyil, derived data-dır. Bu fəsil batch, Fəsil 11 isə stream üzərində dayanır.

Unix alətləri ilə batch processing

Yuxarıdakı sualı Unix alətləri ilə həll edək. Log faylında hər sətir bir request-dir və yeddinci sütun URL-dir:

cat access.log |
  awk '{print $7}' |
  sort |
  uniq -c |
  sort -r -n |
  head -n 5

Eyni şeyi hər hansı proqramlaşdırma dilində in-memory hash table ilə də yazmaq olar: hər URL üçün sayğac saxlayırsınız, sonda sıralayırsınız. Hansı yaxşıdır? Fərqli URL-lərin sayı yaddaşa sığırsa, hash table daha sadədir. Sığmırsa, sort yanaşması qalib gəlir: sort yaddaşa sığmayan data-nı avtomatik olaraq diskə spill edir, parçaları ayrı-ayrı sıralayır və sonra merge edir. Bu, Fəsil 3-dəki SSTable-ların merge məntiqinin eynisidir. Beş sətirlik pipeline gigabaytlarla log-u bir laptopda bir neçə dəqiqəyə emal edir.

Unix fəlsəfəsi

Bu pipeline-ı mümkün edən dörd prinsip var və onlar bu fəslin qalanında dəfələrlə qayıdacaq:

• Uniform interface. Hər proqramın input-u da, output-u da eyni tipli şeydir: byte ardıcıllığı, adətən yeni sətirlə ayrılmış ASCII mətn. awk sort-un nə olduğunu bilmir, amma onunla işləyə bilir.
• Məntiqlə wiring-in ayrılması. Proqram stdin-dən oxuyur, stdout-a yazır; onun harada işlədiyi, input-un hardan gəldiyi shell-in işidir. Bu, loose coupling-in ən sadə formasıdır.
• Şəffaflıq və eksperiment. Input fayl immutable-dır, ona görə pipeline-ı istədiyiniz qədər təkrar işlədə bilərsiniz. Hər mərhələdən sonra less ilə nəticəyə baxmaq, pipeline-ı ortadan kəsib aralıq nəticəni fayla yazmaq olur.
• Kiçik, bir işi yaxşı görən alətlər. Hər alət sadədir, güc onların kompozisiyasındadır.

Unix alətlərinin ən böyük məhdudiyyəti isə tək maşında işləməsidir. Data terabaytlara çatanda kompozisiya ideyasını saxlayıb onu yüzlərlə maşına yaymaq lazımdır. MapReduce məhz bu idi.

MapReduce və distributed filesystem

MapReduce Unix pipeline-ının distributed versiyasına bənzəyir: input immutable-dır, output başqa job-un input-u ola bilər, proqramlar side effect yaratmır. Fərq ondadır ki, stdin və stdout əvəzinə distributed filesystem, məsələn HDFS, işlədilir.

HDFS shared-nothing prinsipi üzərində qurulub: hər maşının öz diski var, mərkəzi storage cihazı yoxdur. Hər node-da daemon fayl bloklarını saxlayır, mərkəzi metadata server isə hansı blokun hansı maşında olduğunu bilir. Bloklar bir neçə maşına replicate olunur, yəni Fəsil 5-dəki replication burada fayl səviyyəsindədir. Nəticədə on minlərlə maşının diski bir böyük filesystem kimi görünür, qiyməti isə adi commodity hardware-in qiymətidir.

Job-un icrası

MapReduce job-u iki funksiya ilə təyin olunur:

• Mapper hər input record üçün bir dəfə çağırılır və sıfır, bir və ya çox key-value cütü çıxarır. Yuxarıdakı nümunədə mapper log sətirindən URL-i çıxarıb (url, 1) qaytarır.
• Reducer eyni key-ə aid bütün value-ları iterator kimi alır və nəticə yaradır: (url, count).

İkisinin arasında framework-ün etdiyi iş shuffle adlanır: mapper-in output-u key-in hash-inə görə partition-lara bölünür, hər partition sıralanır, reducer-lər öz partition-larını mapper maşınlarından çəkib merge edir. Yəni Unix pipeline-ındakı sort burada görünməz şəkildə, lakin yüzlərlə maşında paralel baş verir.

Vacib bir dizayn qərarı: hesablamanı data-nın yanına aparmaq. Scheduler mapper task-ı mümkün qədər input blokunun replica-sının olduğu maşında işlədir. Kodu şəbəkə ilə göndərmək bir neçə meqabaytdır, data-nı göndərmək isə terabaytlar.

Tək job-un bacardığı məhduddur, ona görə real analizlər 50-100 job-dan ibarət workflow-lardır: bir job-un output qovluğu növbətinin input-udur. Bu zəncirləri idarə etmək üçün ayrıca scheduler alətləri yaranıb.

Reduce tərəfində join və grouping

Analitik sorğuların çoxu join tələb edir. Məsələn, user activity event-lərini user profile cədvəli ilə birləşdirmək lazımdır ki, "hansı yaş qrupu hansı səhifələrə baxır" sualı cavablansın. Hər event üçün user database-inə sorğu göndərmək fəlakətdir: throughput database-in limitinə düşür, nəticə isə job işlədiyi müddətdə dəyişən data-dan asılı olur. Düzgün yol database-in snapshot-ını da distributed filesystem-ə kopyalayıb join-i batch-in içində etməkdir.

Sort-merge join belə işləyir: bir mapper event-ləri oxuyub (user_id, event), digəri profile-ları oxuyub (user_id, profile) çıxarır. Shuffle hər user_id-ni eyni reducer-ə gətirir. Secondary sort hiyləsi ilə reducer əvvəl profile record-u, sonra event-ləri görür: profile-ı lokal dəyişəndə saxlayır, hər event üçün output yaradır. Join bir keçiddə, əlavə şəbəkə sorğusu olmadan tamamlanır.

Eyni mexanizm group by üçün də işləyir: mapper qruplaşdırma açarını key kimi qoyur, reducer isə sayır, cəmləyir və ya session-ları bərpa edir. Bunun dəyəri reducer-ə düşən yükün key paylanmasından asılı olmasıdır. Sosial şəbəkədə bir məşhurun milyonlarla follower-i varsa, həmin key bir reducer-də hot spot yaradır və bütün job onu gözləyir. Həll üsulları: hot key-ləri əvvəlcədən tapıb bir neçə reducer-ə random səpələmək, kiçik tərəfi replicate etmək, və ya iki mərhələli aggregation.

Map tərəfində join

Reduce-side join ümumi, lakin bahalıdır: bütün data shuffle-dan keçir. Input haqqında əlavə fərziyyə edə bilirsinizsə, join-i mapper-də bitirmək olur:

Bu üsullar ucuzdur, amma input-un strukturuna dair şərtlər qoyur. Bu şərtlər də çox vaxt əvvəlki job-un output-u ilə təmin olunur: bir job data-nı user_id-yə görə partition edib sıraladısa, sonrakı job-lar map-side join-dən istifadə edə bilər. Workflow dizaynı ona görə vacibdir.

Batch workflow-ların output-u

Batch job-un məqsədi nədir? OLAP hesabatları bir cavabdır, amma ən maraqlı output-lar başqa sistemlərin input-u olan derived data-dır:

• Search index. Bir neçə job sənədləri partition edir, hər partition üçün term dictionary və posting list-lər qurur. Index faylları immutable-dır; sənədlər dəyişəndə ya hamısı yenidən qurulur, ya da Fəsil 3-dəki kimi segment-lər merge edilir.
• Serving üçün key-value store. Məsələn, recommendation model-inin hər user üçün nəticələri. Reducer-dən birbaşa database-ə yazmaq səhvdir: hər record üçün şəbəkə sorğusu yavaşdır, job yarıda qalsa database yarımçıq vəziyyətdə qalır. Düzgün yol job-un içində Fəsil 3-dəki SSTable formatında immutable fayllar yaratmaq, sonra onları serving database-inə bulk load etməkdir.

Bu yanaşmanın fəlsəfəsi Unix-dən gəlir və çox praktik faydalar verir:

• Input immutable olduğu üçün səhv kod işlətsəniz, düzəldib yenidən işlədirsiniz; köhnə output silinir. Bu, human fault tolerance-dır və deploy-ları cəsarətli edir.
• Task uğursuz olanda framework onu yenidən işlədir, çünki mapper və reducer deterministic və side effect-siz olmalıdır. Bu, Fəsil 1-dəki reliability-nin batch versiyasıdır.
• Eyni input faylı müxtəlif komandalar müxtəlif məqsədlə oxuya bilir; wiring məntiqdən ayrılıb.

Hadoop və distributed database-lər

MPP, yəni massively parallel processing, database-ləri də cluster üzərində paralel sorğu icra edir. Niyə o zaman MapReduce ekosistemi ayrıca yarandı? Üç fərq var.

Storage müxtəlifliyi. Database data-nın model-ə uyğun strukturlaşmasını tələb edir. Distributed filesystem isə hər şeyi qəbul edir: log, şəkil, video, parse olunmamış JSON. Data əvvəl yığılır, sonra kimsə onu şərh etməyi tapır. Bu, Fəsil 2-dəki schema-on-read-in ekstremal formasıdır.

Processing model müxtəlifliyi. SQL güclü, lakin bütün hesablamalar üçün uyğun deyil: machine learning model-i train etmək, şəkil emal etmək, graph üzərində iterasiya. MapReduce ixtiyari kod işlədir, SQL isə onun üstündə bir layer kimi qurulur.

Tez-tez fault-lara dizayn. MPP database sorğu zamanı node düşəndə bütün sorğunu yenidən işlədir, çünki sorğular saniyələr çəkir. Saatlarla çəkən job-da bu qəbuledilməzdir, ona görə MapReduce hər task-ın nəticəsini diskə yazır və yalnız uğursuz task-ı təkrarlayır. Bunun kökündə Google-un öz cluster-lərində aşağı prioritetli batch task-ların istənilən vaxt preempt edilməsi, yəni daha vacib iş üçün öldürülməsi dayanırdı.

MapReduce-dan sonrası

MapReduce sadə və güclüdür, amma workflow-ların çoxu üçün səmərəsizdir. Əsas problem aralıq state-in materialisation-udur: hər job öz output-unu tam olaraq distributed filesystem-ə yazır, növbəti job onu oxuyur. Bunun qiyməti:

• növbəti job əvvəlkinin bütün task-ları bitənə qədər başlaya bilmir, bir yavaş task hər şeyi saxlayır;
• yalnız bir job-a lazım olan aralıq data üç dəfə replicate edilir;
• mapper çox vaxt sadəcə əvvəlki reducer-in output-unu oxuyub partition edir, yəni artıq işdir.

Dataflow engine-lər, məsələn Spark, Flink və Tez, bütün workflow-u bir job kimi götürür və onu operator-ların graph-ı kimi icra edir. Operator-lar arasında data pipelined ötürülür: növbəti operator input-un bir hissəsi hazır olan kimi işə başlayır. Shuffle yalnız həqiqətən lazım olanda edilir; sort lazım deyilsə, edilmir. Aralıq data mümkün olduqca yaddaşda və ya lokal diskdə qalır.

Materialisation olmadan fault tolerance necə təmin olunur? Hesablamanı yenidən etməklə. Engine hər data parçasının hansı input-dan hansı operator-larla alındığını izləyir və itirilmiş parçanı yenidən hesablayır. Bunun şərti operator-ların deterministic olmasıdır: eyni input-a eyni output. Random nömrə, sistem saatı və ya hash table iteration sırası determinizmi pozur və yenidən hesablama fərqli nəticə verə bilər. Təcrübədə engine-lər belə operator-ların nəticəsini materialise etməyi seçir.

Graph-lar və iterativ emal

PageRank kimi alqoritmlər eyni hesablamanı yaxınlaşma alınana qədər təkrarlayır. MapReduce ilə hər iterasiya ayrıca job-dur, bu da hər dəfə bütün graph-ı yenidən oxumaq deməkdir. Bulk synchronous parallel model, Pregel kimi, fərqli işləyir: hər vertex öz state-ini saxlayır, iterasiyada qonşularına mesaj göndərir, növbəti iterasiyada mesajları alıb state-ini yeniləyir. Mesajlar batch halında toplanır, fault tolerance isə hər iterasiyadan sonra checkpoint ilə təmin edilir. Graph yaddaşa sığırsa, tək maşında işləmək çox vaxt cluster-dən sürətlidir; distributed graph alqoritmləri hələ də açıq sahədir.

Yüksək səviyyəli API-lər

Mapper və reducer yazmaq assembly-də proqramlaşdırmağa bənzəyir. Buna görə SQL-ə bənzər deklarativ layer-lər və dataframe tipli API-lər yarandı. Fəsil 2-də gördüyümüz üstünlük burada da işləyir: deklarativ join yazanda engine broadcast hash join və ya sort-merge join seçimini özü edir, Fəsil 3-dəki column-oriented storage-dan lazımsız sütunları oxumur, vectorised icra edir. Nəticə MPP database-lərə yaxınlaşır, lakin ixtiyari kod işlətmək imkanı qalır: statistik alqoritmlər, model train etmək, şəkil emalı eyni engine-in operator-larına çevrilir.

Bu fəslin əsas ideyası tək cümlədə belədir: immutable input, deterministic funksiya, yenidən işlədilə bilən output. Fəsil 11 eyni ideyanı sonsuz input-a tətbiq edir.

Tapşırıqlar

1. Bir ayın access log-larında hər user-in ilk və son request-i arasındakı müddəti hesablayan MapReduce job-u dizayn edin. Mapper nə çıxarır, reducer nə alır, secondary sort lazımdırmı? Eyni işi Unix pipeline-ı ilə yazın və hansı həcmdən sonra pipeline-ın bəs etmədiyini əsaslandırın.
2. 2 TB sifariş cədvəlini 200 MB məhsul kataloqu ilə join etmək lazımdır. Broadcast hash join, partitioned hash join və reduce-side join üçün hər birinin şəbəkə trafiki və yaddaş tələbini təxmini qiymətləndirin. Kataloq 50 GB olsaydı, cavab necə dəyişərdi?
3. Recommendation nəticələrini batch job-dan serving database-inə çatdırmaq üçün iki variant müqayisə edin: reducer-dən birbaşa yazma və immutable SSTable faylları yaradıb bulk load. Job yarıda qalanda, kod səhvi tapılanda və database-i 10 dəfə böyütmək lazım olanda hər variant necə davranır?