Airflowን እንዴት መጠቀም እንደሚቻል፡ አስተማማኝ የውሂብ መስመሮችን ለመገንባት ተግባራዊ፣ ከጫፍ እስከ ጫፍ መመሪያ
ውሂብን የሚያንቀሳቅሱ ወይም የ ML ስራዎችን የሚያቀናብሩ ከሆነ፣ ምናልባትም ተመሳሳይ ሐረግ ሰምተው ይሆናል፡ “በቃ በ Airflow ውስጥ ያስቀምጡት።” እውነቱ ሲናገር፣ Apache Airflow ውስብስብ በሆኑ የስራ ፍሰቶች ላይ ታይነት፣ አስተማማኝነት እና ቁጥጥር ሲፈልጉ ያበራል። በዚህ ተግባራዊ መመሪያ ውስጥ፣ Airflowን እንዴት መጠቀም እንደሚችሉ ደረጃ በደረጃ እንመለከታለን—ከዋና ፅንሰ-ሀሳቦች እስከ ምርት-ዝግጁ ቅጦች—ስለዚህ የሚያምኗቸውን የመስመር ዝርጋታዎች መላክ ይችላሉ።
ተግባራዊ እናደርገዋለን፡ ስለ DAGs እና ተግባራት የአዕምሮ ሞዴል፣ ከ TaskFlow API ጋር የተያያዙ ምሳሌዎች፣ የአሰማራ አማራጮች፣ የሙከራ ስልቶች እና ምርጥ ልምዶች ያገኛሉ። በመጨረሻ፣ ከ“ትምህርቱን ማስኬድ እችላለሁ” ወደ “ይህን በምርት ውስጥ ማስኬድ እችላለሁ” ትቀየራላችሁ።
ማስታወሻ፡ ጥልቅ ለመጥለቅ እና ለማጣቀሻ፣ ኦፊሴላዊ ሰነዶች በጣም ጥሩ እና በመደበኛነት የተዘመኑ ናቸው።
Apache Airflow ምንድን ነው፣ በእርግጥ?
Airflow አስተባባሪ ነው—የውሂብ ማቀነባበሪያ አይደለም። በሌላ ቦታ የሚያሄዱትን ስራ (ዳታቤዝ፣ መጋዘኖች፣ Spark ስራዎች፣ ኤፒአይዎች፣ ኮንቴይነሮች) መርሐግብር ያስይዛል፣ ያዛል እና ይቆጣጠራል። የስራ ፍሰቶችን እንደ DAGs (Directed Acyclic Graphs) ይገልፃሉ፣ እነዚህም ተግባራትን እና ጥገኝነታቸውን የሚያመሰጥሩ የ Python ፋይሎች ናቸው። Airflow ከዚያም እነዚያን ተግባራት በእርስዎ መርሐግብር፣ መለኪያዎች እና አካባቢ መሰረት ያስፈጽማል።
- DAG: የስራ ፍሰት ፍቺ (ከጥገኝነት ጋር የተግባራት ግራፍ)።
- ተግባር፡ የስራ ክፍል (የ Python ተግባር፣ የ SQL አፈጻጸም፣ የ Bash ትዕዛዝ፣ የውጭ ስራ ቀስቃሽ፣ ወዘተ)።
- ኦፕሬተር፡ ለተግባር አይነት አብነት (ለምሳሌ፣
PythonOperator፣ BashOperator፣ KubernetesPodOperator)።
- መርሐግብር አስያዥ፡ ምን መሮጥ እንዳለበት እና መቼ እንደሚሮጥ ይወስናል።
- አስፈፃሚ፡ ተግባራትን ያከናውናል (በአካባቢው፣ በ Celery፣ Kubernetes፣ ወዘተ)።
- UI: ለሩጫዎች፣ ምዝግብ ማስታወሻዎች፣ ዳግም ሙከራዎች እና የዘር ሐረግ መቆጣጠሪያ ማዕከልዎ።
Airflowን ከጫኑ በኋላ በኦፊሴላዊ ትምህርቶች ይጀምሩ; ትልቁን ምስል በፍጥነት ይሰጡዎታል።
Airflowን በትክክለኛው መንገድ መጫን እና ማስኬድ
Airflow ተለዋዋጭ ነው። ከመድረክዎ ጋር የሚስማማውን መንገድ ይምረጡ፡
- በፕሮጀክቱ የቀረበውን ፈጣን ጅምር Docker Compose ይጠቀሙ። የድር አገልጋዩን፣ መርሐግብር አስያዥውን፣ ዳታቤዙን እና ሌሎችንም በተገቢው ነባሪዎች ያሽከረክራል።
- DAGsን ለመማር እና ለመድገም በጣም ጥሩ ነው።
- Celery Executor ወይም Kubernetes Executor በአስተዳደር Postgres።
- ምዝግብ ማስታወሻዎችን በ S3/GCS ውስጥ ያስቀምጡ እና ጥገኝነቶችን በምስልዎ ወይም በ
requirements.txt ያሽጉ።
- ለተለዋዋጭነት Kubernetes Executor ወይም Celery Executor ከራስ-ሰር ልኬት ሠራተኞች ጋር።
- ውጫዊ ሚስጥሮች (ቮልት)፣ ጠንካራ ታዛቢነት (ምዝግብ ማስታወሻዎች + መለኪያዎች) እና ለማሻሻያዎች ሰማያዊ/አረንጓዴ ማሰማራት።
ጠቃሚ ምክር፡ የ Airflow ኮድ ቤዝዎን ከማስተዋወቅዎ በፊት በስሪት ቁጥጥር ስር ያድርጉት፣ ኮንቴይነራይዝ ያድርጉት እና ይፈትሹት። “ምርጥ ልምዶች” ገጽ ምርት-ዝግጁ ቅጦችን ያስቀምጣል።
በየቀኑ የሚጠቀሙባቸው ዋና ፅንሰ-ሀሳቦች
DAGs: የስራ ፍሰትዎ እንደ ኮድ
DAG የሚከተሉትን የሚገልጽ የ Python ፋይል ነው፦
- DAG metadata: መታወቂያ፣ መርሐግብር፣ የመነሻ ቀን፣ መለያዎች።
- ነባሪ ክርክሮች፡ ዳግም ሙከራዎች፣ ባለቤቶች፣ SLAs።
DAGን እንደ “ምን” እና “መቼ” እና ተግባራትን እንደ “እንዴት” አድርገው ያስቡ።
ተግባራት እና ኦፕሬተሮች
ኦፕሬተሮች ለተለመዱ ተግባራት ቅድመ ዝግጅት ናቸው። ምሳሌዎች:
- PythonOperator / TaskFlow
@task ለ Python ኮድ
- SimpleHttpOperator ለኤፒአይዎች
- KubernetesPodOperator ለኮንቴይነራይዝድ ስራዎች
- SQL አቅራቢዎች (ለምሳሌ፣ Snowflake፣ BigQuery፣ Postgres) ለመጋዘን ስራ
TaskFlow API: ዘመናዊው፣ Pythonic መንገድ
የ TaskFlow API ተግባራትን እንደ Python ተግባራት በ @task እንዲጽፉ፣ በ XCom በኩል የሚያልፉ እሴቶችን እንዲመልሱ እና በንጽህና እንዲያቀናብሩ ያስችልዎታል። ቦይለርፕሌትን ይቀንሳል እና ተነባቢነትን ያሻሽላል—በጣም ይመከራል።
የእርስዎ የመጀመሪያ Airflow DAG (TaskFlow እትም)
ከዚህ በታች ቁልፍ ሀሳቦችን ለማሳየት አነስተኛ የ ETL-style ምሳሌ አለ-መርሐግብር ማስያዝ፣ TaskFlow፣ ጥገኝነት እና XCom የውሂብ ማለፍ።
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(
dag_id="weather_etl_example",
start_date=datetime(2024, 1, 1),
schedule="0 * * * *", # hourly
catchup=False,
tags=.
መርሐግብር ማስያዝ፣ ማካካሻ እና የኋላ ሙላቶች
schedule: Cron ወይም ቅድመ-ቅምጦች (@daily, @hourly)።
start_date + catchup: catchup=True ከሆነ፣ Airflow ከጅምር ቀን ጀምሮ ሩጫዎችን ወደ ኋላ ይሞላል። ለዥረት-ቅጥ መስመሮች፣ catchup=False ያዘጋጁ።
- በእጅ የኋላ ሙላቶች፡ ታሪካዊ ክፍተቶችን እንደገና ለማስኬድ UI ወይም CLI ይጠቀሙ።
ተግባራዊ የአውራ ጣት ህግ፡ ለውሳኔ ሰጪ የቡድን ስራዎች ማካካሻን አንቃ፤ በእውነተኛ ጊዜ ወይም በኤፒአይ የፍጥነት-የተገደቡ መስመሮች ያሰናክሉ።
በተግባሮች መካከል ውሂብን በደህና ማለፍ (XCom)
- ትናንሽ ነገሮች፡ ከ TaskFlow ጋር የእሴት መመለሻዎች ጥሩ ናቸው።
- ትላልቅ ሸክሞች፡ በነገር ማከማቻ (S3/GCS) በ XCom ውስጥ ቁልፍ ያስቀምጡ።
- በ XCom ውስጥ ሚስጥራዊ መረጃዎችን ያስወግዱ; ሚስጥራዊ የጀርባ ፍጻሜዎችን (ለምሳሌ፣ ቮልት) እና የአካባቢ ተለዋዋጮችን ይጠቀሙ።
ተለዋዋጭ የተግባር ካርታ ስራ እና የደጋፊ-ውጭ የስራ ጫናዎች
Airflow በግቤት ላይ በመመስረት ተግባራትን በተለዋዋጭነት ማመንጨት ይችላል—ለተከፋፈሉ የውሂብ ስብስቦች ወይም ባለብዙ ተከራይ ስራዎች ተስማሚ ነው።
- DAGsን ቆራጥ እና idempotent ያድርጉ።
- ማስተባበርን (Airflow) ከስሌት (Spark, dbt, መጋዘኖች) ይለዩ።
- ለግልጽነት እና ለ XCom ንጽህና TaskFlow API ይጠቀሙ።
- DAGsን መለኪያ ያድርጉ; ተለዋዋጮችን በጥንቃቄ ይጠቀሙ።
- የመስመር ዝርጋታዎችዎን ይቆጣጠሩ፣ ያስጠነቅቁ እና ይመዝግቡ።
ከውሂብ መጋዘኖች እና ML ጋር እንዴት እንደሚሰራ
- የውሂብ መጋዘኖች፡ ለ SQL ስራዎች የአቅራቢ ኦፕሬተሮችን (ለምሳሌ፣ SnowflakeOperator፣ BigQueryInsertJobOperator) ይጠቀሙ። SQLን በፋይሎች ወይም በስሪት ሞጁሎች ውስጥ ያስቀምጡ።
- dbt: dbtን በ Bash/KubernetesPodOperator ወይም በተሰጡ dbt ኦፕሬተሮች በአቅራቢዎች በኩል ቀስቅስ።
- ML: የባህሪ ትውልድን፣ ስልጠናን እና የቡድን ድምዳሜን እንደ የተለየ ተግባራት ያስተባብሩ፤ ቅርሶችን በማከማቻ ውስጥ ያስቀምጡ እና መለኪያዎችን ይመዝግቡ።
የላቀ መርሐግብር ማስያዝ፡ የውሂብ ስብስቦች እና የ DAG ጥገኝነትን ማቋረጥ
- የውሂብ ስብስቦች አንድ DAG ሲዘመን ሌላ DAGን የሚቀሰቅስ ሎጂካዊ የውሂብ ስብስብ እንዲያመነጭ ያስችላሉ—ከማስታወቂያ ቀስቅሴዎች የበለጠ ንጹህ።
- ለአሮጌ ቅጦች፣ ExternalTaskSensor ይሰራል፣ ነገር ግን የውሂብ ስብስቦች የበለጠ መግለጫዎች ናቸው።
ደህንነት እና ተገዢነት
- በ UI ውስጥ ሚና ላይ የተመሰረተ የመዳረሻ መቆጣጠሪያ (RBAC) ይጠቀሙ።
- አካባቢዎችን በቡድን ወይም በታማኝነት ወሰን ይለዩ።
- በምዝግብ ማስታወሻዎች እና የግንኙነት ለውጥ ታሪክ በኩል የኦዲት ዱካዎችን ያስቀምጡ።
ማሻሻያዎች እና ስሪት መስጠት
- በምርት-መሰል የስራ ጫናዎች ደረጃ ላይ ማሻሻያዎችን ይሞክሩ።
- አቅራቢዎችን ይሰኩ እና ሆን ብለው ያሻሽሉ።
- ለአስፈፃሚ-ተኮር ለውጦች እና ውድቀቶች የተለቀቁ ማስታወሻዎችን ያንብቡ።
ለመጀመሪያው የምርት DAGዎ ፈጣን የማረጋገጫ ዝርዝር
- ግልጽ ባለቤትነት (
owner መለያ) እና ማንቂያዎች ተዋቅረዋል።
retries በተመጣጣኝ የኋላ ማፈግፈግ ተዘጋጅቷል።
- Idempotent ተግባራት እና ግልጽ ጥገኝነት።
- ትናንሽ XCom ሸክሞች; ትልቅ ውሂብ በማከማቻ ውስጥ።
- ምዝግብ ማስታወሻዎች ወደ ዘላቂ ማከማቻ ተልከዋል; መለኪያዎች ወደ ውጭ ተልከዋል።
- የማስጀመሪያ እቅድ (ካናሪ ወይም ሰማያዊ/አረንጓዴ) እና የመመለሻ ደረጃዎች።
ምሳሌ፡ እውነተኛ የመጋዘን ጭነት DAG
ይህ ንድፍ ዕለታዊ ፋይሎችን ያወጣል፣ ያረጋግጣል እና ወደ መጋዘን ሠንጠረዥ ይጭናል፣ በእያንዳንዱ ክፍልፍል ተለዋዋጭ ካርታ እና ሊዘገዩ የሚችሉ ዳሳሾች።
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- ወደ ምርት ከማስተዋወቅዎ በፊት ምርጥ ልምዶችን ይገምግሙ።
- ለስርዓቶችዎ (መጋዘኖች፣ ደመናዎች፣ ML መሳሪያዎች) የአቅራቢ ሰነዶችን ያስሱ።
በነገራችን ላይ፡ በ AI የጎን አጋዥ የደራሲነትን ፍጥነት ያሳድጉ
ልብ ሊባል የሚገባው፡ ብዙ DAGs የሚጽፉ ከሆነ፣ ኮድን የሚረዳ AI ረዳት ቦይለርፕሌትን ማፋጠን፣ የ TaskFlow ስታቦችን ማመንጨት እና ጥገኝነት ጥገናዎችን እንኳን ሊጠቁም ይችላል። ከአርታዒዎ እና ከአሳሽዎ ጎን ቀለል ያለ ረዳት ከፈለጉ፣ Sider.AI በልማት ወቅት ፈጣን የኮድ ድጋሚ ጽሑፎችን እና ማብራሪያዎችን ለመስጠት ጠቃሚ ሊሆን ይችላል። ቁልፍ መውሰጃዎች
- Airflowን ለማስተባበር ይጠቀሙ፣ ለማስላት አይደለም።
- ንጹህ፣ ሊሞከሩ የሚችሉ DAGs ለማግኘት TaskFlow APIን ይምረጡ።
- ውሂብን ከ XCom ያውጡ; በምትኩ ማጣቀሻዎችን ያስተላልፉ።
- ቦታዎችን ለመቆጠብ ሊዘገዩ የሚችሉ ዳሳሾችን/ኦፕሬተሮችን ይጠቀሙ።
- ኮንቴይነራይዝ ያድርጉ፣ ይፈትሹ እና በአካባቢዎች ያስተዋውቁ።
- እንደ ሰሜን ኮከብዎ በኦፊሴላዊ ትምህርቶች እና ምርጥ ልምዶች ላይ ይተማመኑ።
FAQ
Q1:Airflowን እንዴት መጠቀም እንደሚቻል ለመማር ቀላሉ መንገድ ምንድነው?
DAGs፣ ተግባራትን፣ መርሐግብርን እና UI ለመረዳት በኦፊሴላዊው ትምህርት ይጀምሩ። ከዚያ ትንሽ በ TaskFlow ላይ የተመሠረተ የመስመር ዝርጋታ ይገንቡ እና ለምርት ዝግጁነት ምርጥ ልምዶችን በመጠቀም ይድገሙት።
Q2:በ Airflow ውስጥ TaskFlow API ወይም ክላሲክ ኦፕሬተሮችን መጠቀም አለብኝ?
ለአብዛኛዎቹ የ Pythonic መስመሮች TaskFlow APIን ይጠቀሙ ምክንያቱም የበለጠ ንጹህ እና የ XCom መመለሻዎችን በተፈጥሮ ስለሚይዝ። ክላሲክ ኦፕሬተሮች እንደ Bash፣ SQL ወይም ኮንቴይነር ስራዎች ላሉ Python ላልሆኑ ተግባራት አሁንም በጣም ጥሩ ናቸው።
Q3:በ Airflow ተግባራት መካከል ትልቅ ውሂብን እንዴት ማለፍ እችላለሁ?
ትላልቅ ሸክሞችን በ XCom ውስጥ ከማስቀመጥ ይቆጠቡ። ተግባራትን ፈጣን እና አስተማማኝ ለማድረግ ውሂብን በ S3/GCS ወይም በዳታቤዝ ውስጥ ያስቀምጡ እና ማጣቀሻዎችን ወይም URIs በ XCom በኩል ብቻ ያስተላልፉ።
Q4:በምርት ውስጥ Airflow ለማግኘት የትኛውን አስፈፃሚ መምረጥ አለብኝ?
ለተለዋዋጭነት እና ለመነጠል፣ Kubernetes Executor ጠንካራ ነባሪ ነው። ቀለል ያሉ ጭነቶች ለማግኘት፣ Celery Executor በደንብ ይሰራል—ራስ-ሰር ልኬትን፣ ጠንካራ ምዝግብ ማስታወሻን እና ውጫዊ ሚስጥሮችን ማረጋገጥዎን እርግጠኛ ይሁኑ።
Q5:በብዙ Airflow DAGs ላይ ጥገኝነትን እንዴት መያዝ እችላለሁ?
አንድ የመስመር ዝርጋታ ለሌላ ውሂብ ሲያመነጭ መግለጫዎችን ለመሻገር-DAG ቀስቅሴዎች የውሂብ ስብስቦችን ይጠቀሙ። በአማራጭ፣ ExternalTaskSensor ሩጫዎችን ማስተባበር ይችላል፣ ነገር ግን የውሂብ ስብስቦች ለውሂብ-ተኮር ማስተባበር የበለጠ ንጹህ ናቸው።
