معرفی جامع و آشنایی کامل با آپاچی اسپارک

دنیای بیگ دیتا

سلام دوستان عزیزم. قبل از هر چیزی بگم من این مطلب رو در وب سایت دنیای بیگ دیتا نشر دادم که اونجا هم می تونین راجع به مباحث بیگ دیتا بیشتر مطالعه کنید.

مقدمه

دیگه همه مون می دونیم چرا مدل هایی مثل مپ ردیوس بوجود اومدن. علتش همین بیگ دیتا است. دیتامون بیگ میشه، نمی تونیم روی یه سرور جاش بدیم، مجبوریم بریم سمت مقیاس پذیری افقی و Distributed Systems و به امثال San Storage بگیم بای بای. گوگل اعظم اول Google File System – GFS رو ارائه داد و بعدش HDFS از روی همین GFS ساخته شد و بعدش گوگل Big Table رو داد و HBase هم از روی همین BT ساخته شد که چیز خفنیه.

هدوپ و مپ ردیوس

فلسفه کار در مپ ردیوس این است: بجای اینکه دیتا رو روی سرورهامون پخش کنیم میآیم کدمون رو میاریم روی سرورهامون و اجرا می کنیم. گذشته از نحوه اجرای مپ ردیوس که از مپ، شافل و ردیوس تشکیل شده باید گفت کُند هست. علتش هم اینه که دیتا رو از روی HDFS میخونه دوباره توی HDFS قرارش میده و به همین منوال ادامه پیدا می کنه. HDFS هم ذاتا کُند هست چرا؟ چون دیسک بیس هست. در ضمن بصورت پیش فرض باید Replication=3 باشه و همین مزید بر علت میشه. پس از اونجایی که هدوپ دیسک بیس هست نرخ I/O بالایی داره و برای کارهای Batch خوبه و باید بدونیم از Streaming پشتیبانی نمی کنه. ضمنا برای کارهای Iterative مثل الگوریتم های ماشین لرنینگی از جمله خوشه بندی مناسب نیست اونم بخاطر سرعت پایینی که داره. یه نقطه ضعفی که هدوپ داره اینه که برای هرکاری هی ابزار خاص خودشو داره که باعث میشه ملت قاطی کنن خدایی. میخوای کار x انجام بدی برو ابزار xn رو یاد بگیر. همین کارو سخت میکنه. هرکدوم از این ابزارها رو باید نصب کرد که خودش ماجرایی داره بعدش باید یاد گرفت که زمان بره و Maintenance اون هم بحث خاص خودشو داره.

در تصویر زیر مثال معروف و سنتی Word Count رو با استفاده از Map Reduce مشاهده می کنید. یه نکته ای که هست و بعضا دیدم فراموش میشه اینه که فازهای Shuffle and Sort توی MR فراموش میشه. گام های فرآیند MR عبارتند از:

Map > Partition > Shuffle > Sort > Reduce

بعد از اینکه متن ورودی ما در قالب nسطر مجزی شد(کاری که تو فاز اولیه خیلی پروژه های پردازش متن میبینیم) تو فاز Map در واقع یه Key Value درست میشه و ابتدا با نگاشت 1 به همه کلمات کار شروع میشه. ورودی فاز Shuffling در واقع خروجی فاز Map هستش و میاد کلمات یکسان رو ادغام میکنه و کنار هم قرار میده تا بعنوان ورودی بفرسته به فاز Reduce.  و تو فاز Reduce هم فرآیند تجمیع مقادیر تکرار کلمات موجود در متن متناظر با کلمات که کلید هستند انجام میشه.

آپاچی اسپارک

اما اسپارک که فاقد لایه Storage هست و از HDFS بعنوان لایه ذخیره سازی استفاده می کنه یه بار دیتای اصلی رو از روی HDFS میخونه و بقیه گام ها رو روی حافظه RAM انجام میده و همین In-Memory بودن اسپارک باعث میشه سرعت بالایی داشته باشه و میل و رغبت ملت ? تو صنعت بره سمتش. ضمنا اسپارک یه موتوره پردازشیه لامبورگینی بیس هستش ? که اکثرا نیازمندی های ما رو در قالب یک چارچوب یکپارچه برطرف می کنه همونطور که خودش اینطور خودشو معرفی میکنه:

Apache Spark™ is a unified analytics engine for large-scale data processing.

تو شکل ذیل هم ماژول های کلیدی اسپارک رو می بینید که در مقالات بعدی مفصل راجع بهشون می نویسم و مثال حل می کنیم باهاشون. همونطور که می بینید تو فیلدهای ذیل ورود کرده که نشان از جامعیت و یکپارچگی اش داره. ملت بدونید پشت هر کدوم از این ابزارها یه مقاله توپ از جاهای خفن دنیا خوابیده در راس اونها دانشگاه کالیفرنیا، برکلی، مهد اسپارک.

• تحلیل گراف: اول GraphX بود و الگوریتم های علم شبکه مثل Page Rank و … رو پشتیبانی میکنه که الآن GraphFrame اومده بیرون که می تونید یه دیتا فریم از گراف بسازید و روش کوئری SQL بزنید. خدایی خفن نیست. رو گرافمون با SQL کوئری بزنیم؟
• یادگیری ماشین: دو تا ماژول MLlib و ML رو داره با یه عالمه الگوریتم، از TF-IDF بگیر تا K-Menas و …
• پردازش Stream: فیت پردازش دیتاهای استریم مثل توییتر و … هست. فرق اسپارک استریم با مپ ردیوس اینه که دیتاهاش بصورت پیوسته و Continuous هست. هرچند واقعا استریم نیست و میکروبچ هستش به این صورت که مثلا میگه برو هر 30 ثانیه یه بار 10000 تا دیتا از اسپارک رو بخون و بیار و روش MR بزن. برخلاف Storm که واقعا استریمه چون میره و دونه دونه دیتا رو میخونه. اما چرا باعث میشه اسپارک سریع تر باشه؟ چون میره یه دفعه یه حجم زیادی از دیتا رو میخونه برمیداره میاره.
• و SQL : می تونیم روی دیتامون کوئری بزنیم که از Hive استفاده می کنه.

معماری اسپارک

اون کلاستر منیجر که وسط می بینید می تونه یکی از موارد ذیل باشه:

1. Spark Standalone (خود اسپارک)
2. Apache Mesos
3. YARN

کلاستر مینجر میاد چک میکنه چه ورکرهایی داریم چقدر RAM دارن چقدر Core دارن. کلا کارش مدیریت منابع مصرفی مون هست و بقول معروف Resource Manager هستش. یاد سیستم عامل افتادم.

اما دروازه ورود ما به منظور ارتباط برنامه ای که نوشتیم با آپاچی اسپارک Spark Context هستش. به این شکل که آدرس Master رو می دیم به Spark Context که خود SC میره به Master وصل میشه، دیتا رو می گیره و بررسی می کنه Workerها کجا قرار دارن.

این شکل خیلی خوبه. Driver برنامه رو اجرا می کنه. Task ها رو خودش تعریف می کنه اگه به فلش ها دقت کنید. هر Task هم همونطور که می بینید تحت یک JVM اومده بالا. Driver مشخص میکنه که هر Task که تو شکل با T مشخص شده به چه Workerی باید Assign بشه. این هم تو کد ما هست که تو فایل Jar مون وجود داره.

ببینید اون Executor که تو شکل معلومه یه دونه برنامه جاوایی هست. تابلویه دیگه. کنارش زده JVM ? شما وقتی برنامه تون رو Submit می کنید (با Spark-Submit) تو Master Spark، اتومات میاد توی Worker مربوطمون و یه دونه Executor JVM میاره بالا مخصوص برنامه شما.

مهم است بدانیم هر Node و Cluster یه JVM مجزی داره.

مفهوم RDD در اسپارک

بیس و پایه و اساس اسپارک همین مفهوم RDD هستش که اولین APIش هست و بعد از اون  Dataset و  DataFrame اومدن. RDD درواقع Abstraction اصلی تو اسپارک هستش که به دیتا نگاه می کنه. یعنی چی؟ یعنی وقتی به HBase وصل می شیم می گیم برو دیتا رو بخون بردار بیار به کل دیتای HBase به چشم یه RDD نگاه میکنه. همین سطح انتزاع تو رفتارش باعث سادگی کار ما میشه

اینجا R به Resilient اشاره داره که به انعطاف پذیری اشاره داره. یعنی چی؟ خب ما گفتیم اسپارک بر اساس RAM کار میکنه. حالا یه موقع موشک خورد به RAM و ترکید رفت رو هوا. چی کنیم؟ اینجا R میاد کمکون و می تونه دیتا رو دوباره برامون بسازه.

RDD ها Immutable هستند بر خلاف X-Men که Mutable بودند ?

?

اما Operation ها تو RDD به دو دسته تقسیم میشن:

Transformation
Map
Filter
ReduceByKey

ActionCollect
Count
Reduce
Take(n)

همونطور که بارها تکرار کردیم اسپارک روی RAM کار میکنه و هر لحظه ممکنه سرور خاموش بشه یا بترکه بره رو هوا و دیتا از روی RAM بپره دود شه بره هوا. اسپارک یه مکانیزمی داره که می تونه RDD رو Persist کنه روی دیسک و هر موقع لازم شد سریع اونو از روی دیسک بخونه و عملیات هاش رو انجام بده. بصورت پیشفرض Persist() میاد دیتا رو تو JVM Heap به عنوان unserialized objects ذخیره میکنه. اگه از Persist استفاده کنیم می تونیم از سطوح ذخیره سازی متعددی مثل

• MEMORY_ONLY
• MEMORY_AND_DISK
• MEMORY_ONLY_SER
• MEMORY_AND_DISK_SER
• DISK_ONLY

استفاده کنیم.

اما ما می تونیم از Cache هم استفاده کنیم که در این صورت تنها از سطح ذخیره سازی MEMORY_ONLY. بهره مند میشیم.

یه مفهومی هم داریم به اسم Broadcast Variables، وقتی یه چیزی رو می خوایم روی همه سرورها Share کنیم ازش استفاده می کنیم. مثلا می خوایم روی دیتا Map بزنیم یا Filter کنیم.