Dataproc در موتور محاسباتی گوگل

۱. مقدمه - گوگل دیتاپروک

Dataproc یک سرویس کاملاً مدیریت‌شده و بسیار مقیاس‌پذیر برای اجرای Apache Spark، Apache Flink، Presto و بسیاری از ابزارها و چارچوب‌های متن‌باز دیگر است. از Dataproc برای مدرن‌سازی دریاچه داده، ETL/ELT و علم داده امن در مقیاس جهانی استفاده کنید. Dataproc همچنین به طور کامل با چندین سرویس Google Cloud از جمله BigQuery ، Cloud Storage ، Vertex AI و Dataplex یکپارچه شده است.

Dataproc در سه نسخه موجود است:

• Dataproc Serverless به شما امکان می‌دهد کارهای PySpark را بدون نیاز به پیکربندی زیرساخت و مقیاس‌بندی خودکار اجرا کنید. Dataproc Serverless از بارهای کاری دسته‌ای PySpark و جلسات/نوت‌بوک‌ها پشتیبانی می‌کند.
• Dataproc در Google Compute Engine به شما امکان می‌دهد علاوه بر ابزارهای متن‌باز مانند Flink و Presto، یک خوشه Hadoop YARN را برای بارهای کاری Spark مبتنی بر YARN مدیریت کنید. می‌توانید خوشه‌های مبتنی بر ابر خود را با هر میزان مقیاس‌بندی عمودی یا افقی که می‌خواهید، از جمله مقیاس‌بندی خودکار، تنظیم کنید.
• Dataproc در موتور Google Kubernetes به شما امکان می‌دهد خوشه‌های مجازی Dataproc را در زیرساخت GKE خود برای ارسال کارهای Spark، PySpark، SparkR یا Spark SQL پیکربندی کنید.

۲. یک کلاستر Dataproc روی یک VPC گوگل کلود ایجاد کنید

در این مرحله، با استفاده از کنسول Google Cloud، یک کلاستر Dataproc روی Google Cloud ایجاد خواهید کرد.

به عنوان اولین قدم، API سرویس Dataproc را در کنسول فعال کنید. پس از فعال شدن، عبارت "Dataproc" را در نوار جستجو جستجو کرده و روی Create Cluster کلیک کنید.

برای استفاده از ماشین‌های مجازی Google Compute Engine(GCE) به عنوان زیرساخت اصلی برای اجرای کلاسترهای Dataproc، گزینه Cluster را در Compute Engine انتخاب کنید.

اکنون در صفحه ایجاد خوشه هستید.

در این صفحه:

• یک نام منحصر به فرد برای خوشه ارائه دهید.
• منطقه خاص را انتخاب کنید. شما همچنین می‌توانید یک منطقه (Zone) را انتخاب کنید، با این حال، Dataproc قابلیتی را فراهم می‌کند که به طور خودکار یکی را برای شما انتخاب کند. برای این codelab، "us-central1" و "us-central1-c" را انتخاب کنید.
• نوع خوشه "استاندارد" را انتخاب کنید. این تضمین می‌کند که یک گره اصلی وجود دارد.
• در برگه پیکربندی گره‌ها ، تأیید کنید که تعداد کارگران ایجاد شده دو نفر خواهد بود.
• در بخش سفارشی‌سازی خوشه ، کادر کنار «فعال کردن دروازه کامپوننت» را علامت بزنید. این کار دسترسی به رابط‌های وب روی خوشه، از جمله Spark UI، Yarn Node Manager و Jupyter notebooks را فعال می‌کند.
• در قسمت Optional Components ، گزینه Jupyter Notebook را انتخاب کنید. این گزینه، کلاستر را با یک سرور Jupyter notebook پیکربندی می‌کند.
• بقیه موارد را همانطور که هست بگذارید و روی ایجاد خوشه کلیک کنید.

این باعث راه‌اندازی یک کلاستر Dataproc می‌شود.

۳. کلاستر را اجرا کنید و SSH را به آن وارد کنید

زمانی که وضعیت کلاستر به Running تغییر کرد، روی نام کلاستر از کنسول Dataproc کلیک کنید.

برای مشاهده گره اصلی و دو گره کارگر کلاستر، روی تب VM Instance کلیک کنید.

برای ورود به گره اصلی، روی SSH کنار گره اصلی کلیک کنید.

برای مشاهده ساختار دایرکتوری، دستورات hdfs را اجرا کنید.

hadoop_commands_example

sudo hadoop fs -ls /

sudo hadoop version

sudo hadoop fs -mkdir /test51 

sudo hadoop fs -ls /

۴. رابط‌های وب و درگاه‌های کامپوننت

از کنسول کلاستر Dataproc روی نام کلاستر خود کلیک کنید، سپس روی تب WEB INTERFACES کلیک کنید.

این رابط‌های وب موجود، از جمله Jupyter را نشان می‌دهد. برای باز کردن یک دفترچه یادداشت Jupyter، روی Jupyter کلیک کنید. می‌توانید از این برای ایجاد دفترچه یادداشت در PySpark ذخیره شده در GCS استفاده کنید. دفترچه یادداشت خود را در Google Cloud Storage ذخیره کنید و یک دفترچه یادداشت PySpark را برای استفاده در این codelab باز کنید.

۵. نظارت و مشاهده‌ی کارهای اسپارک

با فعال و در حال اجرا بودن کلاستر Dataproc، یک کار دسته‌ای PySpark ایجاد کنید و آن را به کلاستر Dataproc ارسال کنید.

یک سطل (Bucket) برای ذخیره‌سازی اسکریپت PySpark در فضای ابری گوگل (Google Cloud Storage) یا به اختصار GCS ایجاد کنید . مطمئن شوید که سطل را در همان ناحیه‌ای که کلاستر Dataproc قرار دارد، ایجاد می‌کنید.

حالا که GCS bucket ایجاد شده است، فایل زیر را در این bucket کپی کنید.

https://raw.githubusercontent.com/diptimanr/spark-on-gce/main/test-spark-1.py

این اسکریپت یک نمونه Spark DataFrame ایجاد می‌کند و آن را به عنوان یک جدول Hive می‌نویسد.

hive_job.py

from pyspark.sql import SparkSession
from datetime import datetime, date
from pyspark.sql import Row

spark = SparkSession.builder.master("local").enableHiveSupport().getOrCreate()

df = spark.createDataFrame([ (1, 2., 'string1', date(2000, 1, 1), datetime(2000, 1, 1, 12, 0)),
        (2, 3., 'string2', date(2000, 2, 1), datetime(2000, 1, 2, 12, 0)), (3, 4., 'string3', date(2000, 3, 1), datetime(2000, 1, 3, 12, 0))
    ], schema='a long, b double, c string, d date, e timestamp')
print("..... Writing data .....")
df.write.mode("overwrite").saveAsTable("test_table_1")
print("..... Complete .....")

این اسکریپت را به عنوان یک کار دسته‌ای Spark در Dataproc ارسال کنید. در منوی ناوبری سمت چپ روی Jobs کلیک کنید و سپس روی Submit Job کلیک کنید.

شناسه و منطقه‌ی اجرای کار را وارد کنید. کلاستر خود را انتخاب کنید و محل GCS اسکریپت Spark که کپی کرده‌اید را وارد کنید. این کار به عنوان یک کار دسته‌ای Spark در Dataproc اجرا خواهد شد.

در قسمت Properties، کلید spark.submit.deployMode و مقدار client اضافه کنید تا مطمئن شوید که درایور در گره اصلی Dataproc اجرا می‌شود و نه در گره‌های کارگر. برای ارسال کار دسته‌ای به Dataproc، روی Submit کلیک کنید.

اسکریپت Spark یک Dataframe ایجاد می‌کند و در جدول Hive test_table_1 می‌نویسد.

پس از اجرای موفقیت‌آمیز کار، می‌توانید دستورات چاپ کنسول را در زیر برگه نظارت (Monitoring) مشاهده کنید.

اکنون که جدول Hive ایجاد شده است، یک کار پرس و جوی Hive دیگر ارسال کنید تا محتوای جدول انتخاب شده و در کنسول نمایش داده شود.

یک job دیگر با مشخصات زیر ایجاد کنید:

توجه داشته باشید که نوع کار (Job Type) روی Hive و نوع منبع پرس‌وجو (query source type) روی Query Text تنظیم شده است، به این معنی که ما کل عبارت HiveQL را در کادر متنی Query Text خواهیم نوشت.

کار را ارسال کنید و بقیه پارامترها را به صورت پیش‌فرض نگه دارید.

توجه کنید که HiveQL چگونه تمام رکوردها را انتخاب کرده و در کنسول نمایش می‌دهد.

۶. مقیاس‌بندی خودکار

مقیاس‌بندی خودکار، وظیفه تخمین تعداد «مناسب» گره‌های کارگر خوشه برای یک بار کاری است.

API مربوط به Dataproc AutoscalingPolicies مکانیزمی برای خودکارسازی مدیریت منابع کلاستر ارائه می‌دهد و امکان مقیاس‌بندی خودکار ماشین مجازی کارگر کلاستر را فراهم می‌کند. یک سیاست مقیاس‌بندی خودکار، پیکربندی قابل استفاده مجددی است که نحوه مقیاس‌بندی کارگران کلاستر با استفاده از سیاست مقیاس‌بندی خودکار را توصیف می‌کند. این سیاست، مرزهای مقیاس‌بندی، فراوانی و میزان تهاجمی بودن را تعریف می‌کند تا کنترل دقیقی بر منابع کلاستر در طول عمر کلاستر فراهم کند.

سیاست‌های مقیاس‌بندی خودکار Dataproc با استفاده از فایل‌های YAML نوشته می‌شوند و این فایل‌های YAML یا در دستور CLI برای ایجاد خوشه ارسال می‌شوند یا هنگام ایجاد یک خوشه از Cloud Console از یک سطل GCS انتخاب می‌شوند.

در اینجا مثالی از سیاست مقیاس‌بندی خودکار Dataproc آورده شده است:

policy.yaml

workerConfig:
  minInstances: 10
  maxInstances: 10
secondaryWorkerConfig:
  maxInstances: 50
basicAlgorithm:
  cooldownPeriod: 4m
  yarnConfig:
    scaleUpFactor: 0.05
    scaleDownFactor: 1.0
    gracefulDecommissionTimeout: 1h

۷. پیکربندی اجزای اختیاری Dataproc

این باعث راه‌اندازی یک کلاستر Dataproc می‌شود.

وقتی یک کلاستر Dataproc ایجاد می‌کنید، اجزای استاندارد اکوسیستم Apache Hadoop به طور خودکار روی کلاستر نصب می‌شوند (به فهرست نسخه‌های Dataproc مراجعه کنید). می‌توانید هنگام ایجاد کلاستر، اجزای اضافی به نام اجزای اختیاری (Optional Components) را روی کلاستر نصب کنید.

هنگام ایجاد خوشه Dataproc از کنسول، اجزای اختیاری را فعال کرده و Jupyter Notebook را به عنوان جزء اختیاری انتخاب کرده‌ایم.

۸. منابع را پاکسازی کنید

برای پاک‌سازی کلاستر، پس از انتخاب کلاستر از کنسول Dataproc، روی Stop کلیک کنید. پس از توقف کلاستر، برای حذف کلاستر، روی Delete کلیک کنید.

پس از حذف خوشه Dataproc، سطل‌های GCS را که کد در آنها کپی شده است، حذف کنید.

برای پاکسازی منابع و متوقف کردن هرگونه هزینه اضافی، خوشه Dataproc ابتدا باید متوقف و سپس حذف شود.

قبل از متوقف کردن و حذف خوشه، اطمینان حاصل کنید که تمام داده‌های نوشته شده در ذخیره‌سازی HDFS برای ذخیره‌سازی پایدار به GCS کپی می‌شوند.

برای متوقف کردن خوشه، روی «توقف» کلیک کنید.

پس از توقف خوشه، برای حذف خوشه، روی حذف کلیک کنید.

در پنجره تأیید، روی حذف کلیک کنید تا خوشه حذف شود.