تشغيل مهمة wordcount في Hadoop على مجموعة Apache Spark مُدارة

1. مقدمة

تُعدّ مسارات العمل من حالات الاستخدام الشائعة في تحليلات البيانات، وهي تتضمّن استيعاب البيانات وتحويلها وتحليلها للعثور على المعلومات المفيدة فيها. في Google Cloud Platform، الأداة المستخدَمة لتنظيم سير العمل هي Managed Apache Airflow، وهي نسخة مستضافة من أداة سير العمل الشائعة والمفتوحة المصدر Apache Airflow. في هذا الدرس التطبيقي، ستستخدم خدمة Managed Apache Airflow لإنشاء سير عمل بسيط ينشئ مجموعة Managed Apache Spark، ويحلّلها باستخدام Managed Apache Spark وApache Hadoop، ثم يحذف مجموعة Managed Apache Spark بعد ذلك.

ما هي خدمة Managed Apache Airflow؟

‫Managed Apache Airflow هي خدمة تنسيق سير عمل مُدارة بالكامل تتيح لك إنشاء سلاسل من الإجراءات وجدولتها ومراقبتها، وتتضمّن هذه السلاسل السحابات ومراكز البيانات المحلية. تستند هذه الخدمة إلى مشروع Apache Airflow الشهير المفتوح المصدر، ويتم تشغيلها باستخدام لغة البرمجة Python، وهي لا تتطلّب أي التزامات تعاقدية كما أنّها سهلة الاستخدام.

من خلال استخدام Managed Apache Airflow بدلاً من نسخة محلية من Apache Airflow، يمكن للمستخدمين الاستفادة من أفضل ميزات Airflow بدون الحاجة إلى تثبيت أو إدارة.

لمزيد من المعلومات حول Managed Apache Airflow، يُرجى الاطّلاع على المستندات.

ما هي أداة Apache Airflow؟

‫Apache Airflow هي أداة مفتوحة المصدر تُستخدَم لإنشاء مهام سير العمل وجدولتها ومراقبتها آليًا. في ما يلي بعض المصطلحات الأساسية التي يجب تذكُّرها والمتعلّقة بـ Airflow والتي ستصادفها خلال الدرس التطبيقي:

• مخطط موجه غير دوري: DAG هو مجموعة من المهام المنظَّمة التي تريد جدولتها وتنفيذها. يتم تحديد الرسوم البيانية الموجّهة غير الدورية (DAG)، المعروفة أيضًا باسم سير العمل، في ملفات Python عادية.
• عامل التشغيل: يصف عامل التشغيل مهمة واحدة في سير العمل

ما هي خدمة Managed Apache Spark؟

Managed Apache Spark هي خدمة Apache Spark وApache Hadoop المُدارة بالكامل من Google Cloud Platform. تتكامل خدمة Apache Spark المُدارة بسهولة مع خدمات GCP الأخرى، ما يمنحك منصة قوية وكاملة لمعالجة البيانات والإحصاءات وتعلُّم الآلة.

لمزيد من المعلومات حول Managed Apache Spark، يُرجى الاطّلاع على المستندات.

الإجراءات التي ستتّخذها

يوضّح لك هذا الدرس التطبيقي حول الترميز كيفية إنشاء سير عمل Apache Airflow وتشغيله في Managed Apache Airflow الذي يكمل المهام التالية:

• تنشئ هذه الطريقة مجموعة Apache Spark مُدارة
• تشغيل مهمة wordcount في Apache Hadoop على المجموعة، وإخراج نتائجها إلى Cloud Storage
• يحذف المجموعة

ما ستتعلمه

• كيفية إنشاء سير عمل Apache Airflow وتشغيله في Managed Apache Airflow
• كيفية استخدام Managed Apache Airflow وManaged Apache Spark لتشغيل عملية تحليل على مجموعة بيانات
• كيفية الوصول إلى بيئة Managed Apache Airflow من خلال Google Cloud Platform Console وCloud SDK وواجهة مستخدم الويب Airflow

المتطلبات

• حساب Google Cloud Platform
• معرفة أساسية بواجهة سطر الأوامر
• فهم أساسي للغة Python

2. إعداد GCP

إنشاء المشروع

اختَر مشروعًا على Google Cloud Platform أو أنشئ مشروعًا.

سجِّل رقم تعريف مشروعك الذي ستستخدمه في الخطوات اللاحقة.

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

ملاحظة: إذا كنت لا تخطّط للاحتفاظ بالموارد التي تنشئها في هذا البرنامج التعليمي، أنشئ مشروعًا جديدًا بدلاً من اختيار مشروع حالي. وبعد الانتهاء، يمكنك حذف المشروع، ما يؤدي إلى إزالة جميع الموارد المرتبطة بالمشروع والبرنامج التعليمي.

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

فعِّل واجهات برمجة التطبيقات Managed Apache Airflow وManaged Apache Spark وCloud Storage.بعد تفعيلها، يمكنك تجاهل الزر "الانتقال إلى بيانات الاعتماد" والمتابعة إلى الخطوة التالية من البرنامج التعليمي.

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

إنشاء بيئة Apache Airflow مُدارة

أنشئ بيئة Apache Airflow مُدارة باستخدام الإعدادات التالية: الاسم: my-airflow-environment الموقع الجغرافي: us-central1 المنطقة: us-central1-a يمكنك ترك جميع الإعدادات الأخرى على قيمتها التلقائية. انقر على "إنشاء" في أسفل الصفحة.

تكتمل عملية إنشاء البيئة عندما تظهر علامة الاختيار الخضراء على يمين اسم البيئة في صفحة البيئات في GCP Console. سيستغرق إنشاء البيئة بضع دقائق، ولا بأس من الانتقال إلى صفحة أخرى أثناء عملية الإنشاء.

إنشاء حزمة Cloud Storage

في مشروعك، أنشئ حزمة Cloud Storage بالإعدادات التالية: الاسم: <your-project-id> فئة التخزين التلقائية: متعدّدة المناطق الموقع الجغرافي: الولايات المتحدة نموذج التحكّم في الوصول: دقيق انقر على "إنشاء" عندما تكون مستعدًا

3- إعداد Apache Airflow

إذا لم تنتهِ عملية إنشاء "بيئة Apache Airflow المُدارة"، انتقِل إلى خطوة "نموذج سير العمل" وارجع إلى هذا الجزء بعد ذلك.

عرض معلومات بيئة Apache Airflow المُدارة

في GCP Console، افتح صفحة البيئات.

انقر على اسم البيئة للاطّلاع على تفاصيلها.

تقدّم صفحة تفاصيل البيئة معلومات، مثل عنوان URL لواجهة مستخدم الويب في Airflow، ومعرّف مجموعة Google Kubernetes Engine، واسم حزمة Cloud Storage، ومسار المجلد ‎ /dags.

في Airflow، DAG (DAG) هو مجموعة من المهام المنظَّمة التي تريد جدولتها وتنفيذها. يتم تحديد الرسومات البيانية الموجّهة غير الدورية، المعروفة أيضًا باسم مهام سير العمل، في ملفات Python عادية. لا يجدول Managed Apache Airflow سوى الرسومات البيانية الموجّهة غير الدورية في المجلد /dags. يقع المجلد /dags في حزمة Cloud Storage التي ينشئها Managed Apache Airflow تلقائيًا عند إنشاء بيئتك.

ضبط متغيّرات بيئة Apache Airflow

متغيرات Apache Airflow هي مفهوم خاص بـ Airflow يختلف عن متغيرات البيئة. في هذه الخطوة، ستضبط متغيرات Airflow الثلاثة التالية: gcp_project وgcs_bucket وgce_zone.

استخدام gcloud لضبط المتغيّرات

أولاً، افتح Cloud Shell الذي تم تثبيت Cloud SDK عليه بشكل ملائم.

اضبط متغيّر البيئة COMPOSER_INSTANCE على اسم بيئة Managed Apache Airflow

COMPOSER_INSTANCE=my-airflow-environment

لضبط متغيرات Airflow باستخدام أداة سطر الأوامر gcloud، استخدِم الأمر gcloud composer environments run مع الأمر الفرعي variables. ينفِّذ الأمر gcloud composer الأمر الفرعي variables لواجهة سطر الأوامر في Airflow. يمرِّر الأمر الفرعي الوسيطات إلى أداة سطر الأوامر gcloud.

ستنفّذ هذا الأمر ثلاث مرات، مع استبدال المتغيرات بالمتغيرات ذات الصلة بمشروعك.

اضبط gcp_project باستخدام الأمر التالي، مع استبدال <your-project-id> برقم تعريف المشروع الذي دوّنته في الخطوة 2.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcp_project <your-project-id>

ستبدو المخرجات على النحو التالي

kubeconfig entry generated for us-central1-my-airflow-env-123abc-gke.
Executing within the following Kubernetes cluster namespace: composer-1-10-0-airflow-1-10-2-123abc
[2020-04-17 20:42:49,713] {settings.py:176} INFO - settings.configure_orm(): Using pool settings. pool_size=5, pool_recycle=1800, pid=449
[2020-04-17 20:42:50,123] {default_celery.py:90} WARNING - You have configured a result_backend of redis://airflow-redis-service.default.svc.cluste
r.local:6379/0, it is highly recommended to use an alternative result_backend (i.e. a database).
[2020-04-17 20:42:50,127] {__init__.py:51} INFO - Using executor CeleryExecutor
[2020-04-17 20:42:50,433] {app.py:52} WARNING - Using default Composer Environment Variables. Overrides have not been applied.
[2020-04-17 20:42:50,440] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg
[2020-04-17 20:42:50,452] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg

اضبط gcs_bucket باستخدام الأمر التالي، مع استبدال <your-bucket-name> بمعرّف الحزمة الذي دوّنته في الخطوة 2. إذا اتّبعت اقتراحنا، سيكون اسم الحزمة هو نفسه رقم تعريف مشروعك. ستكون النتيجة مشابهة للنتيجة التي حصلت عليها من الأمر السابق.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcs_bucket gs://<your-bucket-name>

اضبط gce_zone باستخدام الأمر التالي. ستكون النتيجة مشابهة للنتائج التي تم الحصول عليها من الأوامر السابقة.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gce_zone us-central1-a

(اختياري) استخدام gcloud لعرض متغيّر

للاطّلاع على قيمة متغيّر، شغِّل الأمر الفرعي variables في واجهة سطر الأوامر Airflow مع الوسيطة get أو استخدِم واجهة مستخدم Airflow.

على سبيل المثال:

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --get gcs_bucket

يمكنك إجراء ذلك باستخدام أيّ من المتغيّرات الثلاثة التي ضبطتها للتو: gcp_project وgcs_bucket وgce_zone.

4. نموذج سير العمل

لنلقِ نظرة على رمز DAG الذي سنستخدمه في الخطوة 5. لا تقلق بشأن تنزيل الملفات الآن، ما عليك سوى متابعة القراءة هنا.

هناك الكثير من التفاصيل التي يجب توضيحها، لذا لنشرحها قليلاً.

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

نبدأ ببعض عمليات استيراد Airflow:

• airflow.models: تسمح لنا بالوصول إلى البيانات وإنشائها في قاعدة بيانات Airflow.
• airflow.contrib.operators: المكان الذي يقيم فيه المشغّلون من المنتدى في هذه الحالة، نحتاج إلى dataproc_operator للوصول إلى واجهة برمجة التطبيقات Managed Apache Spark.
• airflow.utils.trigger_rule: لإضافة قواعد مشغّلة إلى عوامل التشغيل. تتيح القواعد المشغّلة التحكّم بدقة في ما إذا كان يجب أن يتم تنفيذ عامل التشغيل بالنسبة إلى حالة العناصر الرئيسية.

output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep

يحدد هذا السطر موقع ملف الإخراج. السطر المهم هنا هو models.Variable.get('gcs_bucket') الذي سيجلب قيمة المتغير gcs_bucket من قاعدة بيانات Airflow.

WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)

input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

• WORDCOUNT_JAR: موقع ملف ‎.jar الذي سنشغّله في النهاية على مجموعة Apache Spark المُدارة تتم استضافته على "منصة Google السحابية" (GCP) نيابةً عنك.
• input_file: موقع الملف الذي يحتوي على البيانات التي ستتم معالجتها في مهمة Hadoop. سنحمّل البيانات إلى هذا الموقع معًا في الخطوة 5.
• wordcount_args: الوسيطات التي سنمرّرها إلى ملف jar.

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

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

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

يجب توفير المتغيّر default_dag_args في شكل قاموس كلّما تم إنشاء رسم بياني موجّه غير دوري جديد:

• ‫'email_on_failure': يشير إلى ما إذا كان يجب إرسال تنبيهات عبر البريد الإلكتروني عند تعذُّر إكمال مهمة
• 'email_on_retry': تشير إلى ما إذا كان يجب إرسال تنبيهات عبر البريد الإلكتروني عند إعادة محاولة تنفيذ مهمة
• 'retries': يشير إلى عدد محاولات إعادة التشغيل التي يجب أن يجريها Airflow في حال تعذُّر تنفيذ DAG
• 'retry_delay': يشير إلى المدة التي يجب أن تنتظرها Airflow قبل محاولة إعادة التشغيل
• 'project_id': يحدّد هذا الرمز معرّف مشروع Google Cloud الذي سيتم ربط DAG به، وهو ما ستحتاج إليه لاحقًا مع Managed Apache Spark Operator.

with models.DAG(
        'composer_hadoop_tutorial',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

يؤدي استخدام with models.DAG إلى توجيه النص البرمجي لتضمين كل ما يليه داخل الرسم البياني الموجّه غير الدوري نفسه. نرى أيضًا ثلاث وسيطات تم تمريرها:

• الأول هو سلسلة نصية، وهو الاسم الذي سيتم إطلاقه على الرسم البياني الموجّه غير الدوري الذي ننشئه. في هذه الحالة، نستخدم composer_hadoop_tutorial.
• ‫schedule_interval: عنصر datetime.timedelta، وقد ضبطناه هنا على يوم واحد. وهذا يعني أنّ هذا الرسم البياني الموجّه غير الدوري سيحاول التنفيذ مرة واحدة يوميًا بعد 'start_date' الذي تم ضبطه سابقًا في 'default_dag_args'
• ‫default_args: القاموس الذي أنشأناه سابقًا والذي يحتوي على الوسيطات التلقائية للرسم البياني الموجّه غير الدوري

إنشاء مجموعة Apache Spark مُدارة

بعد ذلك، سننشئ dataproc_operator.DataprocClusterCreateOperator الذي ينشئ مجموعة Apache Spark مُدارة.

    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

ضمن عامل التشغيل هذا، نرى بعض الوسيطات، وكلها باستثناء الوسيطة الأولى خاصة بعامل التشغيل هذا:

• task_id: كما هو الحال في BashOperator، هذا هو الاسم الذي نحدّده لعامل التشغيل، ويمكن الاطّلاع عليه من واجهة مستخدم Airflow.
• cluster_name: هو الاسم الذي نحدّده لمجموعة Apache Spark المُدارة. في هذا المثال، أطلقنا عليها الاسم composer-hadoop-tutorial-cluster-{{ ds_nodash }} (راجِع مربّع المعلومات للاطّلاع على معلومات إضافية اختيارية)
• استبدِل num_workers بعدد العمال الذين نخصّصهم لمجموعة Managed Apache Spark
• zone: المنطقة الجغرافية التي نريد أن يكون فيها المجموعة، كما هو محفوظ في قاعدة بيانات Airflow. سيتم قراءة المتغير 'gce_zone' الذي ضبطناه في الخطوة 3.
• master_machine_type - نوع الجهاز الذي نريد تخصيصه للعقدة الرئيسية المُدارة في Apache Spark
• worker_machine_type: نوع الآلة التي نريد تخصيصها لعامل Managed Apache Spark

معلومات إضافية عن المتغيّر cluster_name

يظهر الجزء {{ ds_nodash }} من المَعلمة لأنّ Airflow يتيح استخدام قوالب jinja2. وهي مَعلمة يعرضها Airflow في وقت التشغيل في كل مرة يبدأ فيها المشغّل. في هذه الحالة، يتم استبدال {{ ds_nodash }} بتاريخ التنفيذ الخاص بمخطط DAG بتنسيق YYYYMMDD. إذا لم تكن على دراية بقوالب jinja2، لا بأس في ذلك، إذ لا تحتاج إلى معرفة أي شيء عنها لإكمال هذا التمرين العملي.

إرسال مهمة Apache Hadoop

تتيح لنا السمة dataproc_operator.DataProcHadoopOperator إرسال مهمة إلى مجموعة Managed Apache Spark.

    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

نقدّم عدة مَعلمات:

• task_id - الاسم الذي نحدّده لهذا الجزء من الرسم البياني الموجّه غير الدوري
• استبدِل main_jar بموقع ملف ‎ .jar الذي نريد تشغيله على المجموعة
• cluster_name: اسم المجموعة التي سيتم تنفيذ المهمة عليها، وستلاحظ أنّها مطابقة لما نجده في المشغّل السابق
• arguments - الوسيطات التي يتم تمريرها إلى ملف jar، كما لو كنت تنفّذ ملف ‎ .jar من سطر الأوامر

حذف المجموعة

آخر عامل سننشئه هو dataproc_operator.DataprocClusterDeleteOperator.

    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

وكما يظهر من اسمها، ستؤدي هذه العملية إلى حذف مجموعة معيّنة من Apache Spark المُدارة. نرى هنا ثلاث وسيطات:

• task_id: كما هو الحال في BashOperator، هذا هو الاسم الذي نحدّده لعامل التشغيل، ويمكن الاطّلاع عليه من واجهة مستخدم Airflow.
• cluster_name: هو الاسم الذي نحدّده لمجموعة Managed Apache Spark. في هذا المثال، أطلقنا عليها الاسم composer-hadoop-tutorial-cluster-{{ ds_nodash }} (راجِع مربّع المعلومات بعد "إنشاء مجموعة Managed Apache Spark" للحصول على معلومات إضافية اختيارية)
• trigger_rule: أشرنا إلى "قواعد المشغّلات" بإيجاز أثناء عمليات الاستيراد في بداية هذه الخطوة، ولكن لدينا هنا مثال على استخدامها. بشكلٍ تلقائي، لا يتم تنفيذ عامل تشغيل Airflow إلا بعد إكمال جميع عوامل التشغيل السابقة بنجاح. لا يشترط ALL_DONE عامل تفعيل القاعدة إلا أن تكون جميع عوامل التشغيل السابقة قد اكتملت، بغض النظر عمّا إذا كانت ناجحة أم لا. وهذا يعني هنا أنّه حتى إذا تعذّر تنفيذ مهمة Hadoop، سنظل نريد إيقاف تشغيل المجموعة.

  create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

أخيرًا، نريد أن يتم تنفيذ هذه المشغّلات بترتيب معيّن، ويمكننا الإشارة إلى ذلك باستخدام مشغّلات إزاحة البت في Python. في هذه الحالة، سيتم تنفيذ create_dataproc_cluster أولاً دائمًا، يليه run_dataproc_hadoop ثم delete_dataproc_cluster.

بعد تجميع كل ما سبق، سيبدو الرمز البرمجي على النحو التالي:

# Copyright 2018 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# [START composer_hadoop_tutorial]
"""Example Airflow DAG that creates a Managed Apache Spark cluster, runs the Hadoop
wordcount example, and deletes the cluster.

This DAG relies on three Airflow variables
https://airflow.apache.org/concepts.html#variables
* gcp_project - Google Cloud Project to use for the Managed Apache Spark cluster.
* gce_zone - Google Compute Engine zone where Managed Apache Spark cluster should be
  created.
* gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
  See https://cloud.google.com/storage/docs/creating-buckets for creating a
  bucket.
"""

import datetime
import os

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

# Output file for Managed Apache Spark job.
output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
# Path to Hadoop wordcount example available on every Managed Apache Spark cluster.
WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)
# Arguments to pass to Managed Apache Spark job.
input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date': yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

# [START composer_hadoop_schedule]
with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    # [END composer_hadoop_schedule]

    # Create a Managed Apache Spark cluster.
    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/code.html#default-variables
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    # Run the Hadoop wordcount example installed on the Managed Apache Spark cluster
    # master node.
    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

    # Delete Managed Apache Spark cluster.
    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
        # even if the Managed Apache Spark job fails.
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

    # [START composer_hadoop_steps]
    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster
    # [END composer_hadoop_steps]

# [END composer_hadoop]

إذا تخطّيت الخطوة 3 لأنّ بيئتك لم يتم إنشاؤها، احرص على الرجوع إليها الآن قبل الانتقال إلى الخطوة 5.

5- تحميل ملفات Airflow إلى Cloud Storage

نسخ DAG إلى مجلد /dags

1. أولاً، افتح Cloud Shell الذي تم تثبيت Cloud SDK عليه بشكل ملائم.
2. استنساخ مستودع نماذج Python والانتقال إلى دليل composer/workflows

git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git && cd python-docs-samples/composer/workflows

3. نفِّذ الأمر التالي لضبط اسم مجلد DAG على متغيّر بيئة

DAGS_FOLDER=$(gcloud composer environments describe ${COMPOSER_INSTANCE} \
--location us-central1 --format="value(config.dagGcsPrefix)")

4. نفِّذ الأمر gsutil التالي لنسخ رمز البرنامج التعليمي إلى المكان الذي تم فيه إنشاء مجلد /dags

gsutil cp hadoop_tutorial.py $DAGS_FOLDER

ستبدو المخرجات على النحو التالي:

Copying file://hadoop_tutorial.py [Content-Type=text/x-python]...
/ [1 files][  4.1 KiB/  4.1 KiB]
Operation completed over 1 objects/4.1 KiB.

ملاحظة: يشير هذا الدليل الآن إلى سير العمل باسم composer_hadoop_tutorial

6. استخدام واجهة مستخدم Airflow

للوصول إلى واجهة مستخدم Airflow على الويب باستخدام وحدة تحكّم Google Cloud، اتّبِع الخطوات التالية: افتح صفحة البيئات. في عمود خادم الويب Airflow للبيئة، انقر على رمز النافذة الجديدة. ستفتح واجهة مستخدم الويب Airflow في نافذة متصفّح جديدة.

للحصول على معلومات حول واجهة مستخدم Airflow، يُرجى الاطّلاع على الوصول إلى واجهة الويب.

عرض المتغيّرات

يتم الاحتفاظ بالمتغيرات التي ضبطتها سابقًا في بيئتك. يمكنك عرض المتغيّرات من خلال اختيار المشرف > المتغيّرات من شريط قائمة واجهة مستخدم Airflow.

استكشاف عمليات تنفيذ DAG

عند تحميل ملف DAG إلى المجلد dags في Cloud Storage، يحلّل Cloud Composer الملف. في حال عدم العثور على أي أخطاء، سيظهر اسم سير العمل في قائمة DAG، وسيتم وضع سير العمل في قائمة الانتظار لتشغيله على الفور. للاطّلاع على الرسوم البيانية الموجّهة غير الدورية، انقر على الرسوم البيانية الموجّهة غير الدورية في أعلى الصفحة.

انقر على composer_hadoop_tutorial لفتح صفحة تفاصيل الرسم البياني الموجّه غير الدوري (DAG). تتضمّن هذه الصفحة تمثيلاً رسوميًا لمهام سير العمل والطلبات التابعة.

الآن، في شريط الأدوات، انقر على عرض الرسم البياني ثم مرِّر مؤشر الماوس فوق الرسم لكل مهمة للاطّلاع على حالتها. يُرجى العِلم أنّ الحدود المحيطة بكل مهمة تشير أيضًا إلى الحالة (حدود خضراء = قيد التنفيذ، حدود حمراء = تعذُّر التنفيذ، وما إلى ذلك).

لتشغيل سير العمل مرة أخرى من عرض الرسم البياني:

1. في "عرض الرسم البياني" لواجهة مستخدم Airflow، انقر على الرسم create_dataproc_cluster.
2. انقر على محو لإعادة ضبط المهام الثلاث، ثم انقر على حسنًا للتأكيد.

يمكنك أيضًا الاطّلاع على حالة ونتائج سير عمل composer-hadoop-tutorial من خلال الانتقال إلى صفحات GCP Console التالية:

• مجموعات Apache Spark المُدارة لمراقبة إنشاء المجموعات وحذفها يُرجى العِلم أنّ المجموعة التي يتم إنشاؤها من خلال سير العمل تكون مؤقتة: فهي لا تبقى إلا لمدة سير العمل ويتم حذفها كجزء من مهمة سير العمل الأخيرة.
• مهام Apache Spark المُدارة لعرض مهمة wordcount في Apache Hadoop أو مراقبتها. انقر على معرّف الوظيفة للاطّلاع على ناتج سجلّ المهمة.
• متصفّح Cloud Storage للاطّلاع على نتائج عدد الكلمات في المجلد wordcount في حزمة Cloud Storage التي أنشأتها لهذا الدرس العملي

7. تنظيف

لتجنُّب تحمّل رسوم في حسابك على GCP مقابل الموارد المستخدَمة في هذا الدرس العملي:

1. (اختياري) لحفظ بياناتك، نزِّل البيانات من حزمة Cloud Storage لبيئة Managed Apache Airflow وحزمة التخزين التي أنشأتها لهذا الدرس العملي.
2. احذف حزمة Cloud Storage التي أنشأتها لهذا الدرس العملي.
3. احذف حزمة Cloud Storage الخاصة بالبيئة.
4. احذف بيئة Managed Apache Airflow. يُرجى العِلم أنّ حذف بيئتك لا يؤدي إلى حذف حزمة التخزين الخاصة بها.

يمكنك أيضًا حذف المشروع بشكل اختياري:

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

تنبيه: يؤدي حذف مشروع إلى ما يلي:

• يتم حذف كل شيء في المشروع. إذا كنت قد استخدمت مشروعًا حاليًا لهذا البرنامج التعليمي، سيؤدي حذفه إلى حذف أي عمل آخر أنجزته في المشروع.
• فقدان أرقام تعريف المشاريع المخصّصة: عند إنشاء هذا المشروع، ربما أنشأت رقم تعريف مشروع مخصّصًا تريد استخدامه في المستقبل. للحفاظ على عناوين URL التي تستخدم رقم تعريف المشروع، مثل عنوان URL appspot.com، احذف الموارد المحدّدة داخل المشروع بدلاً من حذف المشروع بأكمله.