۱. مقدمه
گردشهای کاری یک مورد استفاده رایج در تجزیه و تحلیل دادهها هستند - آنها شامل دریافت، تبدیل و تجزیه و تحلیل دادهها برای یافتن اطلاعات معنادار درون آنها میشوند. در پلتفرم ابری گوگل، ابزاری که برای هماهنگسازی گردشهای کاری استفاده میشود، Managed Apache Airflow است که نسخه میزبانی شده ابزار گردش کار متنباز محبوب Apache Airflow است. در این آزمایشگاه، شما از Managed Apache Airflow برای ایجاد یک گردش کار ساده استفاده خواهید کرد که یک خوشه Managed Apache Spark ایجاد میکند، آن را با استفاده از Managed Apache Spark و Apache Hadoop تجزیه و تحلیل میکند، سپس خوشه Managed Apache Spark را حذف میکند.
جریان هوای مدیریتشده آپاچی چیست؟
Apache Airflow مدیریتشده ، یک سرویس هماهنگسازی گردش کار کاملاً مدیریتشده است که به شما امکان میدهد خطوط لولهای را که در سراسر ابرها و مراکز داده داخلی گسترده شدهاند، ایجاد، زمانبندی و نظارت کنید. Managed Apache Airflow که بر اساس پروژه متنباز محبوب Apache Airflow ساخته شده و با استفاده از زبان برنامهنویسی پایتون اداره میشود، هیچ محدودیتی ندارد و استفاده از آن آسان است.
با استفاده از Managed Apache Airflow به جای یک نمونه محلی از Apache Airflow، کاربران میتوانند از بهترین مزایای Airflow بدون هیچ گونه سربار نصب یا مدیریتی بهرهمند شوند.
آپاچی ایرفلو چیست؟
آپاچی ایرفلو یک ابزار متنباز است که برای برنامهنویسی، برنامهریزی و نظارت بر گردشهای کاری استفاده میشود. چند اصطلاح کلیدی در رابطه با ایرفلو وجود دارد که باید به خاطر داشته باشید و در طول تمرین با آنها مواجه خواهید شد:
- DAG - یک DAG (گراف جهتدار غیرمدور) مجموعهای از وظایف سازمانیافته است که میخواهید آنها را زمانبندی و اجرا کنید. DAGها که به آنها گردش کار نیز گفته میشود، در فایلهای استاندارد پایتون تعریف میشوند.
- عملگر - یک عملگر یک وظیفه واحد را در یک گردش کار توصیف میکند.
آپاچی اسپارک مدیریتشده چیست؟
آپاچی اسپارک مدیریتشده، سرویس کاملاً مدیریتشدهی آپاچی اسپارک و آپاچی هادوپ پلتفرم ابری گوگل است. آپاچی اسپارک مدیریتشده به راحتی با سایر سرویسهای GCP ادغام میشود و پلتفرمی قدرتمند و کامل برای پردازش دادهها، تجزیه و تحلیل و یادگیری ماشینی در اختیار شما قرار میدهد.
کاری که انجام خواهید داد
این کدلب به شما نشان میدهد که چگونه یک گردش کار Apache Airflow را در Managed Apache Airflow ایجاد و اجرا کنید که وظایف زیر را انجام میدهد:
- یک خوشه مدیریتشده آپاچی اسپارک ایجاد میکند.
- یک کار شمارش کلمات توسط آپاچی هادوپ را روی کلاستر اجرا میکند و نتایج آن را به فضای ذخیرهسازی ابری ارسال میکند.
- خوشه را حذف میکند
آنچه یاد خواهید گرفت
- نحوه ایجاد و اجرای گردش کار Apache Airflow در Managed Apache Airflow
- نحوه استفاده از Managed Apache Airflow و Managed Apache Spark برای اجرای تحلیلی روی یک مجموعه داده
- نحوه دسترسی به محیط مدیریتشده Apache Airflow از طریق کنسول پلتفرم ابری گوگل، Cloud SDK و رابط وب Airflow
آنچه نیاز دارید
- حساب GCP
- دانش پایه CLI
- درک اولیه از پایتون
۲. راهاندازی GCP
ایجاد پروژه
یک پروژه پلتفرم ابری گوگل (Google Cloud Platform Project) انتخاب یا ایجاد کنید.
شناسه پروژه خود را یادداشت کنید، که در مراحل بعدی از آن استفاده خواهید کرد.
اگر در حال ایجاد یک پروژه جدید هستید، شناسه پروژه درست زیر نام پروژه در صفحه ایجاد پروژه قرار دارد. |
|
اگر قبلاً پروژهای ایجاد کردهاید، میتوانید شناسه آن را در صفحه اصلی کنسول در کارت اطلاعات پروژه پیدا کنید. |
|
فعال کردن APIها
APIهای Managed Apache Airflow، Managed Apache Spark و Cloud Storage را فعال کنید . پس از فعال شدن آنها، میتوانید دکمهای که میگوید «برو به اعتبارنامهها» را نادیده بگیرید و به مرحله بعدی آموزش بروید. |
|
ایجاد محیط مدیریتشدهی جریان هوای آپاچی
یک محیط Managed Apache Airflow با پیکربندی زیر ایجاد کنید :
تمام تنظیمات دیگر میتوانند به صورت پیشفرض باقی بمانند. روی «ایجاد» در پایین کلیک کنید. |
|
ایجاد سطل ذخیرهسازی ابری
در پروژه خود، یک مخزن ذخیرهسازی ابری با پیکربندی زیر ایجاد کنید :
وقتی آماده بودید، روی «ایجاد» کلیک کنید |
|
۳. راهاندازی جریان هوای آپاچی
مشاهده اطلاعات محیط جریان هوای مدیریتشده آپاچی
در کنسول GCP، صفحه محیطها (Environments) را باز کنید.
برای مشاهده جزئیات هر محیط، روی نام آن کلیک کنید.
صفحه جزئیات محیط ، اطلاعاتی مانند URL رابط وب Airflow، شناسه خوشه موتور Kubernetes گوگل، نام مخزن ذخیرهسازی ابری و مسیر پوشه /dags را ارائه میدهد.
در Airflow، یک DAG (گراف جهتدار غیرمدور) مجموعهای از وظایف سازمانیافته است که میخواهید زمانبندی و اجرا کنید. DAGها که به آنها گردش کار نیز گفته میشود، در فایلهای استاندارد پایتون تعریف میشوند. Apache Airflow مدیریتشده فقط DAGهای موجود در پوشه /dags را زمانبندی میکند. پوشه /dags در سطل ذخیرهسازی ابری است که Apache Airflow مدیریتشده هنگام ایجاد محیط شما بهطور خودکار ایجاد میکند.
تنظیم متغیرهای محیطی Apache Airflow
متغیرهای Apache Airflow یک مفهوم مختص Airflow هستند که با متغیرهای محیطی متمایزند. در این مرحله، سه متغیر Airflow زیر را تنظیم خواهید کرد: gcp_project ، gcs_bucket و gce_zone .
استفاده از gcloud برای تنظیم متغیرها
ابتدا، Cloud Shell خود را که SDK Cloud به راحتی برای شما نصب شده است، باز کنید.
متغیر محیطی COMPOSER_INSTANCE روی نام محیط Managed Apache Airflow خود تنظیم کنید.
COMPOSER_INSTANCE=my-airflow-environment
برای تنظیم متغیرهای Airflow با استفاده از ابزار خط فرمان gcloud، از دستور gcloud composer environments run به همراه زیردستور variables استفاده کنید. این دستور gcloud composer variables زیردستور Airflow CLI را اجرا میکند. زیردستور، آرگومانها را به ابزار خط فرمان gcloud منتقل میکند.
شما این دستور را سه بار اجرا خواهید کرد و متغیرها را با متغیرهای مرتبط با پروژه خود جایگزین خواهید کرد.
gcp_project را با استفاده از دستور زیر تنظیم کنید و <your-project-id> را با شناسه پروژهای که در مرحله ۲ یادداشت کردهاید، جایگزین کنید.
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> با شناسه سطلی که در مرحله ۲ یادداشت کردهاید جایگزین کنید. اگر توصیه ما را دنبال کرده باشید، نام سطل شما همان شناسه پروژه شما خواهد بود. خروجی شما مشابه دستور قبلی خواهد بود.
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 برای مشاهده یک متغیر
برای دیدن مقدار یک متغیر، زیردستور Airflow CLI variables با آرگومان get اجرا کنید یا از رابط کاربری Airflow استفاده کنید.
برای مثال:
gcloud composer environments run ${COMPOSER_INSTANCE} \
--location us-central1 variables -- --get gcs_bucket
شما میتوانید این کار را با هر یک از سه متغیری که تنظیم کردهاید انجام دهید: gcp_project ، gcs_bucket و gce_zone .
۴. نمونه گردش کار
بیایید نگاهی به کد DAG که در مرحله ۵ استفاده خواهیم کرد، بیندازیم. فعلاً نگران دانلود فایلها نباشید، فقط مراحل زیر را دنبال کنید.
اینجا خیلی چیزها هست که باید باز شود، پس بیایید کمی آن را تجزیه و تحلیل کنیم.
from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule
ما با واردات برخی از جریانهای هوا شروع میکنیم:
-
airflow.models- به ما امکان دسترسی و ایجاد دادهها در پایگاه داده Airflow را میدهد. -
airflow.contrib.operators- جایی که اپراتورهای جامعه زندگی میکنند. در این مورد، برای دسترسی به Managed Apache Spark API بهdataproc_operatorنیاز داریم. -
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 که در نهایت روی خوشه مدیریتشده آپاچی اسپارک اجرا خواهیم کرد. این فایل از قبل برای شما روی GCP میزبانی شده است. -
input_file- محل فایلی که حاوی دادههایی است که کار Hadoop ما در نهایت روی آنها محاسبه خواهد کرد. دادهها را در مرحله 5 با هم در آن مکان آپلود خواهیم کرد. -
wordcount_args- آرگومانهایی که به فایل jar ارسال خواهیم کرد.
yesterday = datetime.datetime.combine(
datetime.datetime.today() - datetime.timedelta(1),
datetime.datetime.min.time())
این به ما یک شیء datetime معادل میدهد که نیمهشب روز قبل را نشان میدهد. برای مثال، اگر این دستور در ساعت ۱۱:۰۰ روز ۴ مارس اجرا شود، شیء datetime ساعت ۰۰:۰۰ روز ۳ مارس را نشان میدهد. این مربوط به نحوه مدیریت زمانبندی توسط 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 به شکل یک دیکشنری باید هر زمان که یک DAG جدید ایجاد میشود، ارائه شود:
-
'email_on_failure'- مشخص میکند که آیا در صورت عدم موفقیت یک وظیفه، باید هشدارهای ایمیل ارسال شود یا خیر. -
'email_on_retry'- مشخص میکند که آیا هنگام تلاش مجدد برای انجام یک کار، باید هشدارهای ایمیل ارسال شود یا خیر. -
'retries'- نشان میدهد که در صورت خرابی DAG، Airflow باید چند بار تلاش مجدد انجام دهد. -
'retry_delay'- نشان میدهد که Airflow قبل از تلاش برای تلاش مجدد، چه مدت باید منتظر بماند. -
'project_id'- به DAG میگوید که با چه شناسه پروژه GCP مرتبط شود، که بعداً با اپراتور Managed Apache Spark مورد نیاز خواهد بود.
with models.DAG(
'composer_hadoop_tutorial',
schedule_interval=datetime.timedelta(days=1),
default_args=default_dag_args) as dag:
استفاده with models.DAG به اسکریپت میگوید که هر چیزی که در زیر آن قرار دارد را درون همان DAG قرار دهد. همچنین سه آرگومان ارسال شده را میبینیم:
- اولین مورد، یک رشته، نامی است که به DAG که ایجاد میکنیم اختصاص میدهیم. در این مورد، ما
composer_hadoop_tutorialاستفاده میکنیم. -
schedule_interval- یک شیءdatetime.timedeltaکه در اینجا روی یک روز تنظیم کردهایم. این بدان معناست که این DAG تلاش میکند روزی یک بار پس از'start_date'که قبلاً در'default_dag_args'تنظیم شده بود، اجرا شود. -
default_args- دیکشنری که قبلاً ایجاد کردیم و شامل آرگومانهای پیشفرض برای DAG است
ایجاد یک خوشه مدیریتشده آپاچی اسپارک
در مرحله بعد، یک 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- نامی که به خوشه Managed Apache Spark اختصاص میدهیم. در اینجا، ما آن راcomposer-hadoop-tutorial-cluster-نامگذاری کردهایم (برای اطلاعات بیشتر اختیاری به کادر اطلاعات مراجعه کنید) -
num_workers- تعداد workerهایی که به خوشه Managed Apache Spark اختصاص میدهیم -
zone- منطقه جغرافیایی که میخواهیم خوشه در آن قرار گیرد، همانطور که در پایگاه داده Airflow ذخیره شده است. این متغیر'gce_zone'را که در مرحله 3 تنظیم کردیم، میخواند. -
master_machine_type- نوع ماشینی که میخواهیم به سرور اصلی Managed Apache Spark اختصاص دهیم -
worker_machine_type- نوع ماشینی که میخواهیم به کارگر Managed Apache Spark اختصاص دهیم
ارسال یک شغل آپاچی هادوپ
dataproc_operator.DataProcHadoopOperator به ما اجازه میدهد تا یک کار را به یک خوشه مدیریتشده 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- نامی که به این بخش از DAG اختصاص میدهیم -
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)
همانطور که از نامش پیداست، این عملگر یک خوشه مدیریتشده آپاچی اسپارک مشخص را حذف میکند. در اینجا سه آرگومان میبینیم:
-
task_id- درست مانند BashOperator، این نامی است که ما به اپراتور اختصاص میدهیم و از رابط کاربری Airflow قابل مشاهده است. -
cluster_name- نامی که به خوشه مدیریتشده آپاچی اسپارک اختصاص میدهیم. در اینجا، ما آن راcomposer-hadoop-tutorial-cluster-(برای اطلاعات بیشتر اختیاری، به کادر اطلاعات بعد از «ایجاد یک خوشه مدیریتشده آپاچی اسپارک» مراجعه کنید) -
trigger_rule- ما در ابتدای این مرحله به طور خلاصه در مورد Trigger Rules صحبت کردیم، اما در اینجا یکی از آنها را در عمل داریم. به طور پیشفرض، یک عملگر Airflow اجرا نمیشود مگر اینکه همه عملگرهای بالادستی آن با موفقیت تکمیل شده باشند. قانون ماشهALL_DONEفقط مستلزم آن است که همه عملگرهای بالادستی تکمیل شده باشند، صرف نظر از اینکه آیا آنها موفق بودهاند یا خیر. در اینجا این بدان معناست که حتی اگر کار Hadoop با شکست مواجه شود، ما همچنان میخواهیم خوشه را از بین ببریم.
create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster
در نهایت، ما میخواهیم این عملگرها به ترتیب خاصی اجرا شوند و میتوانیم این را با استفاده از عملگرهای bitshift پایتون نشان دهیم. در این حالت، 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]
۵. فایلهای Airflow را در فضای ذخیرهسازی ابری آپلود کنید
DAG را در پوشه /dags خود کپی کنید
- ابتدا، Cloud Shell خود را که SDK Cloud به راحتی برای شما نصب شده است، باز کنید.
- مخزن پایتون سمپلز را کلون کنید و به دایرکتوری composer/workflows بروید.
git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git && cd python-docs-samples/composer/workflows
- دستور زیر را اجرا کنید تا نام پوشه DAGs خود را به یک متغیر محیطی تغییر دهید.
DAGS_FOLDER=$(gcloud composer environments describe ${COMPOSER_INSTANCE} \
--location us-central1 --format="value(config.dagGcsPrefix)")
- دستور
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.
۶. استفاده از رابط کاربری جریان هوا
برای دسترسی به رابط وب Airflow با استفاده از کنسول GCP:
|
|
برای اطلاعات بیشتر در مورد رابط کاربری Airflow، به دسترسی به رابط وب مراجعه کنید.
مشاهده متغیرها
متغیرهایی که قبلاً تنظیم کردهاید در محیط شما باقی میمانند. میتوانید متغیرها را با انتخاب Admin > Variables از نوار منوی رابط کاربری Airflow مشاهده کنید.
بررسی مسیرهای DAG
وقتی فایل DAG خود را در پوشه dags در فضای ذخیرهسازی ابری آپلود میکنید، Cloud Composer فایل را تجزیه میکند. اگر خطایی یافت نشود، نام گردش کار در فهرست DAG ظاهر میشود و گردش کار بلافاصله در صف اجرا قرار میگیرد. برای مشاهده DAGهای خود، روی DAGها در بالای صفحه کلیک کنید.
برای باز کردن صفحه جزئیات DAG، روی composer_hadoop_tutorial کلیک کنید. این صفحه شامل نمایش گرافیکی وظایف و وابستگیهای گردش کار است.
حالا، در نوار ابزار، روی نمای نمودار کلیک کنید و سپس ماوس را روی نمودار هر وظیفه ببرید تا وضعیت آن را ببینید. توجه داشته باشید که حاشیه اطراف هر وظیفه نیز وضعیت را نشان میدهد (حاشیه سبز = در حال اجرا؛ قرمز = ناموفق و غیره).
برای اجرای مجدد گردش کار از نمای گراف :
- در نمای نمودار رابط کاربری Airflow، روی نمودار
create_dataproc_clusterکلیک کنید. - برای تنظیم مجدد سه وظیفه، روی Clear کلیک کنید و سپس برای تأیید، روی OK کلیک کنید.
همچنین میتوانید وضعیت و نتایج گردش کار composer-hadoop-tutorial را با مراجعه به صفحات کنسول GCP زیر بررسی کنید:
- خوشههای مدیریتشدهی آپاچی اسپارک برای نظارت بر ایجاد و حذف خوشه. توجه داشته باشید که خوشه ایجاد شده توسط گردش کار، زودگذر است: فقط برای مدت گردش کار وجود دارد و به عنوان بخشی از آخرین وظیفه گردش کار حذف میشود.
- برای مشاهده یا نظارت بر کار شمارش کلمات آپاچی هادوپ ، از کارهای مدیریتشدهی آپاچی اسپارک استفاده کنید. برای مشاهدهی خروجی گزارش کار، روی شناسهی کار کلیک کنید.
- مرورگر فضای ابری (Cloud Storage Browser) برای دیدن نتایج شمارش کلمات در پوشهی
wordcountدر فضای ابری که برای این آزمایشگاه کد ایجاد کردهاید.
۷. پاکسازی
برای جلوگیری از تحمیل هزینه به حساب GCP خود برای منابع استفاده شده در این آزمایشگاه کد:
- (اختیاری) برای ذخیره دادههایتان، دادهها را از مخزن ذخیرهسازی ابری برای محیط Managed Apache Airflow و مخزن ذخیرهسازی که برای این codelab ایجاد کردهاید، دانلود کنید .
- سطل ذخیرهسازی ابری که برای این آزمایشگاه کد ایجاد کردهاید را حذف کنید .
- سطل ذخیرهسازی ابری را برای محیط حذف کنید .
- محیط Managed Apache Airflow را حذف کنید . توجه داشته باشید که حذف محیط، فضای ذخیرهسازی آن محیط را حذف نمیکند.
همچنین میتوانید به صورت اختیاری پروژه را حذف کنید:
- در کنسول GCP، به صفحه پروژهها بروید.
- در لیست پروژهها، پروژهای را که میخواهید حذف کنید انتخاب کرده و روی حذف کلیک کنید.
- در کادر، شناسه پروژه را تایپ کنید و سپس برای حذف پروژه، روی خاموش کردن کلیک کنید.