Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Managed Service for Apache Spark

1. نظرة عامة حول الحوسبة بدون خادم لـ Apache Spark

الخدمة المُدارة لـ Apache Spark هي خدمة مُدارة بالكامل وقابلة للتوسّع بدرجة كبيرة لتشغيل Apache Spark وApache Flink وPresto والعديد من الأدوات وأُطر العمل الأخرى المفتوحة المصدر. يمكنك استخدام الخدمة المُدارة لـ Apache Spark لتحديث بحيرة البيانات وعمليات استخراج البيانات وتحويلها وتحميلها (ETL) أو استخراج البيانات وتحميلها (ELT) وعلوم البيانات الآمنة على مستوى العالم. تتكامل الخدمة المُدارة لـ Apache Spark أيضًا بشكل كامل مع العديد من خدمات Google Cloud، بما في ذلك BigQuery وCloud Storage وGemini Enterprise Agent Engine وKnowledge Catalog.

تتوفّر خدمة Apache Spark المُدارة بنوعَين:

تتيح لك خدمة Apache Spark المُدارة بدون خادم تشغيل مهام PySpark بدون الحاجة إلى ضبط البنية الأساسية والتوسّع التلقائي. تتيح خدمة Apache Spark المُدارة أحمال العمل المجمّعة في PySpark والجلسات أو دفاتر الملاحظات.
تتيح لك مجموعات Apache Spark المُدارة إدارة مجموعة Hadoop YARN لأحمال عمل Spark المستندة إلى YARN، بالإضافة إلى أدوات مفتوحة المصدر مثل Flink وPresto. يمكنك تخصيص مجموعاتك المستندة إلى السحابة الإلكترونية باستخدام أي قدر تريده من التوسيع العمودي أو الأفقي، بما في ذلك التوسيع التلقائي.

في هذا الدرس التطبيقي حول الترميز، ستتعرّف على عدة طرق مختلفة يمكنك من خلالها استخدام Dataproc Serverless.

تم إنشاء Apache Spark في الأصل للعمل على مجموعات Hadoop واستخدام YARN كمدير للموارد. يتطلّب الحفاظ على مجموعات Hadoop العنقودية مجموعة محدّدة من الخبرات وضمان ضبط العديد من عناصر التحكّم المختلفة في المجموعات العنقودية بشكل صحيح. بالإضافة إلى مجموعة منفصلة من عناصر التحكّم التي يطلب Spark من المستخدم ضبطها أيضًا. يؤدي ذلك إلى العديد من السيناريوهات التي يقضي فيها المطوّرون وقتًا أطول في إعداد البنية الأساسية بدلاً من العمل على رمز Spark نفسه.

تُغنيك خدمة Dataproc Serverless عن الحاجة إلى إعداد مجموعات Hadoop أو Spark يدويًا. لا تعمل خدمة Dataproc Serverless على Hadoop، بل تستخدم تخصيص الموارد الديناميكي الخاص بها لتحديد متطلبات الموارد، بما في ذلك التوسّع التلقائي. لا يزال بإمكانك تخصيص مجموعة فرعية صغيرة من خصائص Spark باستخدام Dataproc Serverless، ولكن في معظم الحالات لن تحتاج إلى تعديلها.

2. إعداد

ستبدأ بإعداد البيئة والموارد المستخدَمة في هذا الدرس التطبيقي حول الترميز.

أنشئ مشروعًا على Google Cloud، أو استخدِم مشروعًا حاليًا.

افتح Cloud Shell من خلال النقر عليه في شريط أدوات Cloud Console.

توفّر Cloud Shell بيئة Shell جاهزة للاستخدام يمكنك استخدامها في هذا الدرس التطبيقي حول الترميز.

سيضبط Cloud Shell اسم مشروعك تلقائيًا. تحقَّق من ذلك من خلال تنفيذ الأمر echo $GOOGLE_CLOUD_PROJECT. إذا لم يظهر رقم تعريف مشروعك في الناتج، اضبطه.

export GOOGLE_CLOUD_PROJECT=<your-project-id>

اضبط منطقة في Compute Engine لمواردك، مثل us-central1 أو europe-west2.

export REGION=<your-region>

تفعيل واجهات برمجة التطبيقات

يستخدم الدرس التطبيقي حول الترميز واجهات برمجة التطبيقات التالية:

BigQuery
Dataproc

فعِّل واجهات برمجة التطبيقات اللازمة. سيستغرق ذلك حوالي دقيقة واحدة، وستظهر رسالة تفيد بنجاح العملية عند اكتمالها.

gcloud services enable bigquery.googleapis.com
gcloud services enable dataproc.googleapis.com

ضبط إذن الوصول إلى الشبكة

يتطلّب استخدام Dataproc Serverless تفعيل الوصول الخاص من Google في المنطقة التي ستنفّذ فيها مهام Spark، لأنّ برامج التشغيل والمنفّذات في Spark لا تتضمّن سوى عناوين IP خاصة. نفِّذ ما يلي لتفعيلها في الشبكة الفرعية default.

gcloud compute networks subnets update default \
  --region=${REGION} \
  --enable-private-ip-google-access

يمكنك التأكّد من تفعيل ميزة "الوصول الخاص من Google" من خلال ما يلي، وسيتم عرض True أو False.

gcloud compute networks subnets describe default \
  --region=${REGION} \
  --format="get(privateIpGoogleAccess)"

إنشاء حزمة تخزين

أنشئ حزمة تخزين سيتم استخدامها لتخزين مواد العرض التي تم إنشاؤها في هذا الدرس العملي.

اختَر اسمًا للحزمة، ويجب أن تكون أسماء الحِزم فريدة على مستوى جميع المستخدمين.

export BUCKET=<your-bucket-name>

أنشئ الحزمة في المنطقة التي تريد تشغيل مهام Spark فيها.

gsutil mb -l ${REGION} gs://${BUCKET}

يمكنك الاطّلاع على الحزمة في وحدة تحكّم Cloud Storage. يمكنك أيضًا تشغيل gsutil ls للاطّلاع على الحزمة.

إنشاء خادم "سجلّ دائم"

توفّر واجهة مستخدم Spark مجموعة كبيرة من أدوات تصحيح الأخطاء والإحصاءات حول مهام Spark. لعرض واجهة مستخدم Spark لوظائف Dataproc Serverless المكتملة، يجب إنشاء مجموعة Dataproc بعقدة واحدة لاستخدامها كـ خادم سجلّ دائم.

حدِّد اسمًا لخادم سجلّاتك الدائمة.

PHS_CLUSTER_NAME=my-phs

نفِّذ ما يلي.

gcloud dataproc clusters create ${PHS_CLUSTER_NAME} \
    --region=${REGION} \
    --single-node \
    --enable-component-gateway \
    --properties=spark:spark.history.fs.logDirectory=gs://${BUCKET}/phs/*/spark-job-history

سنتناول واجهة مستخدم Spark وخادم السجلّ الدائم بمزيد من التفصيل لاحقًا في هذا الدرس العملي.

3- تشغيل مهام Serverless Spark باستخدام Dataproc Batches

في هذا النموذج، ستعمل مع مجموعة بيانات من مجموعة البيانات العامة الخاصة برحلات Citi Bike في مدينة نيويورك. إنّ NYC Citi Bikes هو نظام مدفوع لمشاركة الدراجات في مدينة نيويورك. ستجري بعض عمليات التحويل البسيطة وستطبع أرقام تعريف محطات Citi Bike العشر الأكثر شيوعًا. يستخدم هذا النموذج أيضًا spark-bigquery-connector المفتوح المصدر لقراءة البيانات وكتابتها بسلاسة بين Spark وBigQuery.

استنسِخ مستودع Github التالي وcd في الدليل الذي يحتوي على الملف citibike.py.

git clone https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos/data-analytics/next-2022-workshop/dataproc-serverless

citibike.py

import sys

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import BooleanType

if len(sys.argv) == 1:
    print("Please provide a GCS bucket name.")

bucket = sys.argv[1]
table = "bigquery-public-data:new_york_citibike.citibike_trips"

spark = SparkSession.builder \
          .appName("pyspark-example") \
          .config("spark.jars","gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar") \
          .getOrCreate()

df = spark.read.format("bigquery").load(table)

top_ten = df.filter(col("start_station_id") \
            .isNotNull()) \
            .groupBy("start_station_id") \
            .count() \
            .orderBy("count", ascending=False) \
            .limit(10) \
            .cache()

top_ten.show()

top_ten.write.option("header", True).csv(f"gs://{bucket}/citibikes_top_ten_start_station_ids")

أرسِل المهمة إلى Serverless Spark باستخدام Cloud SDK المتاح في Cloud Shell تلقائيًا. نفِّذ الأمر التالي في واجهة الأوامر الذي يستخدم Cloud SDK وDataproc Batches API لإرسال مهام Serverless Spark.

gcloud dataproc batches submit pyspark citibike.py \
  --batch=citibike-job \
  --region=${REGION} \
  --deps-bucket=gs://${BUCKET} \
  --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar \
--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER_NAME} \
  -- ${BUCKET}

إليك التفاصيل:

يشير gcloud dataproc batches submit إلى Dataproc Batches API.
يشير pyspark إلى أنّك بصدد إرسال مهمة PySpark.
--batch هو اسم الوظيفة. في حال عدم توفيرها، سيتم استخدام معرّف فريد عالمي (UUID) تم إنشاؤه عشوائيًا.
--region=${REGION} هي المنطقة الجغرافية التي ستتم فيها معالجة الوظيفة.
--deps-bucket=${BUCKET} هو المكان الذي يتم تحميل ملف Python المحلي إليه قبل تشغيله في بيئة Serverless.
يتضمّن --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar ملف jar الخاص بـ spark-bigquery-connector في بيئة تشغيل Spark.
‫--history-server-cluster=projects/${GOOGLE_CLOUD_PROJECT}/regions/${REGION}/clusters/${PHS_CLUSTER} هو الاسم المؤهّل بالكامل لخادم سجلّ المحفوظات الدائم. هذا هو المكان الذي يتم فيه تخزين بيانات أحداث Spark (بشكل منفصل عن ناتج وحدة التحكّم) ويمكن عرضها من واجهة مستخدم Spark.
يشير الرمز -- في النهاية إلى أنّ أي شيء بعد ذلك سيكون وسيطات وقت التشغيل للبرنامج. في هذه الحالة، أنت ترسل اسم الحزمة، كما هو مطلوب في المهمة.

ستظهر لك النتيجة التالية عند إرسال الدفعة.

Batch [citibike-job] submitted.

بعد بضع دقائق، ستظهر لك النتيجة التالية مع البيانات الوصفية من المهمة.

+----------------+------+
|start_station_id| count|
+----------------+------+
|             519|551078|
|             497|423334|
|             435|403795|
|             426|384116|
|             293|372255|
|             402|367194|
|             285|344546|
|             490|330378|
|             151|318700|
|             477|311403|
+----------------+------+

Batch [citibike-job] finished.

في القسم التالي، ستتعرّف على كيفية العثور على سجلّات هذه المهمة.

ميزات إضافية

باستخدام Spark Serverless، تتوفّر لك خيارات إضافية لتشغيل مهامك.

يمكنك إنشاء صورة Docker مخصّصة يتم تشغيل مهمتك عليها. هذه طريقة رائعة لتضمين تبعيات إضافية، بما في ذلك مكتبات Python وR.
يمكنك ربط مثيل Dataproc Metastore بمهمتك للوصول إلى بيانات Hive الوصفية.
للحصول على المزيد من التحكّم، تتيح خدمة Dataproc Serverless ضبط مجموعة صغيرة من خصائص Spark.

4. مقاييس Dataproc وإمكانية تتبُّع البيانات

تعرض وحدة تحكّم Dataproc Batches جميع مهام Dataproc Serverless. في وحدة التحكّم، سيظهر لك رقم تعريف الدفعة والموقع الجغرافي والحالة ووقت الإنشاء والوقت المنقضي والنوع لكل مهمة. انقر على رقم تعريف الدفعة الخاص بمهمتك للاطّلاع على مزيد من المعلومات عنها.

في هذه الصفحة، ستظهر لك معلومات مثل رصد التي تعرض عدد عمليات تنفيذ Batch Spark التي استخدمتها مهمتك بمرور الوقت (ما يشير إلى مقدار القياس التلقائي).

في علامة التبويب التفاصيل، ستظهر لك المزيد من البيانات الوصفية حول المهمة، بما في ذلك أي وسيطات ومعلَمات تم إرسالها مع المهمة.

يمكنك أيضًا الوصول إلى جميع السجلّات من هذه الصفحة. عند تشغيل مهام Dataproc Serverless، يتم إنشاء ثلاث مجموعات مختلفة من السجلّات:

مستوى الخدمة
نتائج وحدة التحكّم
تسجيل أحداث Spark

على مستوى الخدمة: يتضمّن هذا المستوى السجلات التي أنشأتها خدمة Dataproc Serverless، مثل طلب Dataproc Serverless لوحدات معالجة مركزية إضافية من أجل التوسّع التلقائي. يمكنك الاطّلاع على هذه السجلات من خلال النقر على عرض السجلات، ما سيؤدي إلى فتح Cloud Logging.

يمكن الاطّلاع على نتائج وحدة التحكّم ضمن النتائج.هذه هي النتائج التي تم إنشاؤها بواسطة المهمة، بما في ذلك البيانات الوصفية التي تطبعها Spark عند بدء مهمة أو أي عبارات طباعة مضمّنة في المهمة.

يمكن الوصول إلى سجلّ أحداث Spark من واجهة مستخدم Spark. بما أنّك وفّرت لخادم سجلّ Spark خادمًا دائمًا، يمكنك الوصول إلى واجهة مستخدم Spark من خلال النقر على عرض خادم سجلّ Spark الذي يحتوي على معلومات عن مهام Spark التي تم تنفيذها سابقًا. يمكنك الاطّلاع على مزيد من المعلومات حول واجهة مستخدم Spark من مستندات Spark الرسمية.

5- نماذج Dataproc: BQ -> GCS

نماذج Dataproc هي أدوات مفتوحة المصدر تساعد في تبسيط مهام معالجة البيانات على السحابة الإلكترونية. تعمل هذه النماذج كغلاف لخدمة Dataproc Serverless وتتضمّن نماذج للعديد من مهام استيراد البيانات وتصديرها، بما في ذلك:

BigQuerytoGCS وGCStoBigQuery
GCStoBigTable
GCStoJDBC وJDBCtoGCS
HivetoBigQuery
MongotoGCS وGCStoMongo

تتوفّر القائمة الكاملة في ملف README.

في هذا القسم، ستستخدم "نماذج Dataproc" لتصدير البيانات من BigQuery إلى Cloud Storage.

إنشاء نسخة طبق الأصل من المستودع

استنسِخ المستودع وغيِّر إلى المجلد python.

git clone https://github.com/GoogleCloudPlatform/dataproc-templates.git
cd dataproc-templates/python

ضبط البيئة

عليك الآن ضبط متغيّرات البيئة. تستخدم "نماذج Dataproc" متغيّر البيئة GCP_PROJECT لرقم تعريف مشروعك، لذا اضبط هذا المتغيّر على GOOGLE_CLOUD_PROJECT.

export GCP_PROJECT=${GOOGLE_CLOUD_PROJECT}

يجب ضبط المنطقة في البيئة من وقت سابق، وإذا لم يتم ضبطها، يمكنك ضبطها هنا.

export REGION=<region>

تستخدم "نماذج Dataproc" spark-bigquery-conector لمعالجة مهام BigQuery، وتتطلّب تضمين معرّف الموارد المنتظم (URI) في متغيّر بيئة JARS. اضبط المتغيّر JARS.

export JARS="gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.26.0.jar"

ضبط مَعلمات النموذج

اضبط اسمًا لحزمة مرحلية لتستخدمها الخدمة.

export GCS_STAGING_LOCATION=gs://${BUCKET}

بعد ذلك، ستضبط بعض المتغيرات الخاصة بالوظيفة. بالنسبة إلى جدول الإدخال، ستشير مرة أخرى إلى مجموعة بيانات NYC Citibike في BigQuery.

BIGQUERY_GCS_INPUT_TABLE=bigquery-public-data.new_york_citibike.citibike_trips

يمكنك اختيار csv أو parquet أو avro أو json. بالنسبة إلى هذا الدرس التطبيقي حول الترميز، اختَر CSV. في القسم التالي، سنشرح كيفية استخدام "نماذج Dataproc" لتحويل أنواع الملفات.

BIGQUERY_GCS_OUTPUT_FORMAT=csv

اضبط وضع الإخراج على overwrite. يمكنك الاختيار بين overwrite أو append أو ignore أو errorifexists.

BIGQUERY_GCS_OUTPUT_MODE=overwrite

اضبط موقع الإخراج في GCS ليكون مسارًا في الحزمة.

BIGQUERY_GCS_OUTPUT_LOCATION=gs://${BUCKET}/BQtoGCS

تشغيل النموذج

شغِّل نموذج BIGQUERYTOGCS من خلال تحديده أدناه وتقديم مَعلمات الإدخال التي ضبطتها.

./bin/start.sh \
-- --template=BIGQUERYTOGCS \
        --bigquery.gcs.input.table=${BIGQUERY_GCS_INPUT_TABLE} \
        --bigquery.gcs.output.format=${BIGQUERY_GCS_OUTPUT_FORMAT} \
        --bigquery.gcs.output.mode=${BIGQUERY_GCS_OUTPUT_MODE} \
        --bigquery.gcs.output.location=${BIGQUERY_GCS_OUTPUT_LOCATION}

سيكون الناتج مزعجًا إلى حدّ ما، ولكن بعد حوالي دقيقة ستظهر لك النتائج التالية.

Batch [5766411d6c78444cb5e80f305308d8f8] submitted.
...
Batch [5766411d6c78444cb5e80f305308d8f8] finished.

يمكنك التأكّد من أنّ الملفات تم إنشاؤها من خلال تنفيذ ما يلي.

gsutil ls ${BIGQUERY_GCS_OUTPUT_LOCATION}

يكتب Spark تلقائيًا في ملفات متعددة، وذلك حسب مقدار البيانات. في هذه الحالة، ستظهر لك حوالي 30 ملفًا تم إنشاؤها. يتم تنسيق أسماء ملفات إخراج Spark باستخدام part- متبوعًا برقم مكوّن من خمسة أرقام (يشير إلى رقم الجزء) وسلسلة تجزئة. بالنسبة إلى كميات كبيرة من البيانات، سيكتب Spark عادةً إلى عدة ملفات. مثال على اسم الملف هو part-00000-cbf69737-867d-41cc-8a33-6521a725f7a0-c000.csv.

6. نماذج Dataproc: تحويل ملف CSV إلى ملف parquet

ستستخدم الآن Dataproc Templates لتحويل البيانات في GCS من نوع ملف إلى آخر باستخدام GCSTOGCS. يستخدم هذا النموذج SparkSQL ويتيح أيضًا إمكانية إرسال طلب بحث SparkSQL ليتمّ تنفيذه أثناء عملية التحويل من أجل إجراء معالجة إضافية.

تأكيد متغيّرات البيئة

تأكَّد من ضبط GCP_PROJECT وREGION وGCS_STAGING_BUCKET من القسم السابق.

echo ${GCP_PROJECT}
echo ${REGION}
echo ${GCS_STAGING_LOCATION}

ضبط مَعلمات النموذج

عليك الآن ضبط مَعلمات الإعدادات لـ GCStoGCS. ابدأ بموقع الملفات المصدر. يُرجى العِلم أنّ هذا دليل وليس ملفًا محدّدًا، لأنّه ستتم معالجة جميع الملفات في الدليل. اضبط هذا الخيار على BIGQUERY_GCS_OUTPUT_LOCATION.

GCS_TO_GCS_INPUT_LOCATION=${BIGQUERY_GCS_OUTPUT_LOCATION}

اضبط تنسيق ملف الإدخال.

GCS_TO_GCS_INPUT_FORMAT=csv

اضبط تنسيق الناتج المطلوب. يمكنك اختيار parquet أو json أو avro أو csv.

GCS_TO_GCS_OUTPUT_FORMAT=parquet

اضبط وضع الإخراج على overwrite. يمكنك الاختيار بين overwrite أو append أو ignore أو errorifexists.

GCS_TO_GCS_OUTPUT_MODE=overwrite

حدِّد موقع الإخراج.

GCS_TO_GCS_OUTPUT_LOCATION=gs://${BUCKET}/GCStoGCS

تشغيل النموذج

نفِّذ نموذج GCStoGCS.

./bin/start.sh \
-- --template=GCSTOGCS \
        --gcs.to.gcs.input.location=${GCS_TO_GCS_INPUT_LOCATION} \
        --gcs.to.gcs.input.format=${GCS_TO_GCS_INPUT_FORMAT} \
        --gcs.to.gcs.output.format=${GCS_TO_GCS_OUTPUT_FORMAT} \
        --gcs.to.gcs.output.mode=${GCS_TO_GCS_OUTPUT_MODE} \
        --gcs.to.gcs.output.location=${GCS_TO_GCS_OUTPUT_LOCATION}

ستكون النتيجة مشوّشة إلى حدّ ما، ولكن بعد دقيقة واحدة تقريبًا، من المفترض أن تظهر لك رسالة نجاح على النحو التالي.

Batch [c198787ba8e94abc87e2a0778c05ec8a] submitted.
...
Batch [c198787ba8e94abc87e2a0778c05ec8a] finished.

يمكنك التأكّد من أنّ الملفات تم إنشاؤها من خلال تنفيذ ما يلي.

gsutil ls ${GCS_TO_GCS_OUTPUT_LOCATION}

باستخدام هذا النموذج، يمكنك أيضًا تقديم طلبات بحث SparkSQL من خلال تمرير gcs.to.gcs.temp.view.name وgcs.to.gcs.sql.query إلى النموذج، ما يتيح تشغيل طلب بحث SparkSQL على البيانات قبل الكتابة إلى GCS.

7. تنظيف الموارد

لتجنُّب تكبُّد رسوم غير ضرورية في حسابك على Google Cloud Platform بعد إكمال هذا الدرس التطبيقي حول الترميز، اتّبِع الخطوات التالية:

احذف حزمة Cloud Storage للبيئة التي أنشأتها.

gsutil rm -r gs://${BUCKET}

احذف مجموعة Dataproc المستخدَمة لخادم السجلّ الدائم.

gcloud dataproc clusters delete ${PHS_CLUSTER_NAME} \
  --region=${REGION}

احذف مهام Dataproc Serverless. انتقِل إلى وحدة تحكّم الدفعات، وانقر على المربّع بجانب كل مهمة تريد حذفها، ثم انقر على حذف.

إذا أنشأت مشروعًا خصيصًا لهذا الدرس التطبيقي حول الترميز، يمكنك أيضًا حذف المشروع اختياريًا:

في وحدة تحكّم Google Cloud Platform، انتقِل إلى صفحة المشاريع.
في قائمة المشاريع، اختَر المشروع الذي تريد حذفه وانقر على "حذف".
في المربّع، اكتب رقم تعريف المشروع، ثم انقر على "إيقاف" لحذف المشروع.

8. الخطوات التالية

تقدّم المراجع التالية طرقًا إضافية يمكنك من خلالها الاستفادة من Serverless Spark:

تعرَّف على كيفية تنظيم سير عمل Dataproc Serverless باستخدام Cloud Composer.
تعرَّف على كيفية دمج Dataproc Serverless مع خطوط أنابيب Kubeflow.