🥇Apache Airflow: ETL सुलभ करणे | प्रोहोस्टर

हाय, मी दिमित्री लॉगविनेंको आहे - व्हेझेट ग्रुप ऑफ कंपन्यांच्या विश्लेषण विभागाचा डेटा अभियंता.

मी तुम्हाला ETL प्रक्रिया विकसित करण्यासाठी एक अद्भुत साधन सांगेन - Apache Airflow. परंतु एअरफ्लो इतका अष्टपैलू आणि बहुआयामी आहे की तुम्ही डेटा प्रवाहात गुंतलेले नसले तरीही तुम्ही त्यावर बारकाईने लक्ष दिले पाहिजे, परंतु वेळोवेळी कोणतीही प्रक्रिया सुरू करणे आणि त्यांच्या अंमलबजावणीचे निरीक्षण करणे आवश्यक आहे.

आणि होय, मी फक्त सांगणार नाही, तर दर्शवितो: प्रोग्राममध्ये बरेच कोड, स्क्रीनशॉट आणि शिफारसी आहेत.

एअरफ्लो / विकिमीडिया कॉमन्स हा शब्द गुगल करताना तुम्ही सहसा काय पाहता

सामग्री सारणी

परिचय
मुख्य भाग, व्यावहारिक (आणि थोडा सैद्धांतिक)
भाग अंतिम, संदर्भ आणि माहिती
- आम्ही तुमच्यासाठी गोळा केलेला रेक
- अधिक ऑटोमेशन साधने
संदर्भ

परिचय

अपाचे एअरफ्लो जॅंगोप्रमाणेच आहे:

अजगरात लिहिलेले
एक उत्तम अॅडमिन पॅनल आहे,
अनिश्चित काळासाठी विस्तारण्यायोग्य

- फक्त चांगले, आणि ते पूर्णपणे भिन्न हेतूंसाठी बनवले गेले होते, म्हणजे (जसे ते कॅटच्या आधी लिहिलेले आहे):

अमर्यादित मशीनवर कार्ये चालवणे आणि देखरेख करणे (बरेच सेलेरी / कुबर्नेट्स आणि तुमचा विवेक तुम्हाला अनुमती देईल)
डायनॅमिक वर्कफ्लो निर्मितीसह पायथन कोड लिहिण्यास आणि समजण्यास अगदी सोपे आहे
आणि तयार केलेले घटक आणि होममेड प्लगइन (जे अत्यंत सोपे आहे) वापरून कोणताही डेटाबेस आणि API एकमेकांशी कनेक्ट करण्याची क्षमता.

आम्ही Apache Airflow याप्रमाणे वापरतो:

आम्ही डीडब्ल्यूएच आणि ओडीएस (आमच्याकडे व्हर्टिका आणि क्लिकहाऊस आहेत) मधील विविध स्त्रोतांकडून (अनेक SQL सर्व्हर आणि पोस्टग्रेएसक्यूएल उदाहरणे, अनुप्रयोग मेट्रिकसह विविध API, अगदी 1C) डेटा गोळा करतो.
किती प्रगत cron, जे ODS वर डेटा एकत्रीकरण प्रक्रिया सुरू करते आणि त्यांच्या देखभालीवर देखील लक्ष ठेवते.

अलीकडे पर्यंत, आमच्या गरजा 32 कोर आणि 50 GB RAM असलेल्या एका छोट्या सर्व्हरद्वारे पूर्ण केल्या जात होत्या. एअरफ्लोमध्ये, हे कार्य करते:

अधिक 200 डॅग्ज (वास्तविक कार्यप्रवाह, ज्यामध्ये आम्ही कार्ये भरली आहेत)
प्रत्येकामध्ये सरासरी 70 कार्ये,
हा चांगुलपणा सुरू होतो (सरासरी देखील) तासातून एकदा.

आणि आम्ही कसे विस्तारित केले याबद्दल, मी खाली लिहीन, परंतु आता आपण über-समस्या परिभाषित करूया जी आपण सोडवू:

तीन मूळ SQL सर्व्हर आहेत, प्रत्येकामध्ये 50 डेटाबेसेस आहेत - अनुक्रमे एका प्रकल्पाची उदाहरणे, त्यांची रचना समान आहे (जवळजवळ सर्वत्र, mua-ha-ha), म्हणजे प्रत्येकाकडे ऑर्डर टेबल आहे (सुदैवाने, त्यासह एक टेबल नाव कोणत्याही व्यवसायात ढकलले जाऊ शकते). आम्ही सर्व्हिस फील्ड (स्रोत सर्व्हर, सोर्स डेटाबेस, ईटीएल टास्क आयडी) जोडून डेटा घेतो आणि अगदी सहजतेने व्हर्टिकामध्ये टाकतो.

चला जाऊया!

मुख्य भाग, व्यावहारिक (आणि थोडा सैद्धांतिक)

आम्ही (आणि तुम्ही) का?

जेव्हा झाडं मोठी होती आणि मी साधा होतो SQL-एक रशियन किरकोळ विक्रीमध्ये, आम्ही आमच्याकडे उपलब्ध दोन साधनांचा वापर करून ईटीएल प्रक्रिया उर्फ डेटा प्रवाहाचा घोटाळा केला:

इन्फॉर्मेटिका पॉवर सेंटर - एक अत्यंत पसरणारी प्रणाली, अत्यंत उत्पादक, स्वतःच्या हार्डवेअरसह, स्वतःची आवृत्ती. मी त्याच्या क्षमतेपैकी 1% गॉड फॉरबिड वापरला. का? बरं, सर्व प्रथम, 380 च्या दशकापासून या इंटरफेसने आपल्यावर मानसिक दबाव आणला. दुसरे म्हणजे, हे कॉन्ट्रॅप्शन अत्यंत फॅन्सी प्रक्रियांसाठी डिझाइन केलेले आहे, तीव्र घटकांचा पुनर्वापर आणि इतर अत्यंत-महत्त्वाच्या-एंटरप्राइझ-युक्त्या. एअरबस AXNUMX / वर्षाच्या पंखाप्रमाणे त्याची किंमत किती आहे याबद्दल आम्ही काहीही बोलणार नाही.
सावध रहा, स्क्रीनशॉट 30 वर्षांपेक्षा कमी वयाच्या लोकांना त्रास देऊ शकतो

SQL सर्व्हर एकत्रीकरण सर्व्हर - आम्ही आमच्या इंट्रा-प्रोजेक्ट फ्लोमध्ये या कॉमरेडचा वापर केला. बरं, खरं तर: आम्ही आधीच एसक्यूएल सर्व्हर वापरतो आणि त्याची ईटीएल साधने न वापरणे काहीसे अवास्तव ठरेल. त्यात सर्व काही चांगले आहे: इंटरफेस सुंदर आहे आणि प्रगती अहवाल ... परंतु यामुळे आम्हाला सॉफ्टवेअर उत्पादने आवडत नाहीत, अरेरे, यासाठी नाही. त्याची आवृत्ती dtsx (जे एक्सएमएल आहे ज्यामध्ये सेव्हवर नोड्स बदलले आहेत) आम्ही करू शकतो, पण मुद्दा काय आहे? शेकडो टेबल्स एका सर्व्हरवरून दुसऱ्या सर्व्हरवर ड्रॅग करणारं टास्क पॅकेज बनवण्याबद्दल काय? होय, शंभर काय, माऊसच्या बटणावर क्लिक करून वीस तुकड्यांमधून तुमची तर्जनी खाली पडेल. परंतु हे निश्चितपणे अधिक फॅशनेबल दिसते:

आम्ही नक्कीच मार्ग शोधत होतो. केस अगदी जवळपास स्व-लिखित SSIS पॅकेज जनरेटरवर आले ...

…आणि मग मला नवीन नोकरी मिळाली. आणि त्यावर अपाचे एअरफ्लोने मला मागे टाकले.

जेव्हा मला कळले की ETL प्रक्रियेचे वर्णन सोपे पायथन कोड आहेत, तेव्हा मी आनंदाने नाचलो नाही. अशाप्रकारे डेटा स्ट्रीम्सचे व्हर्जन आणि फरक केले गेले आणि शेकडो डेटाबेसेसमधून एकाच स्ट्रक्चरसह टेबल्स एका टार्गेटमध्ये टाकणे ही दीड किंवा दोन 13” स्क्रीनमध्ये पायथन कोडची बाब बनली.

क्लस्टर एकत्र करणे

चला पूर्णपणे बालवाडीची व्यवस्था करू नका आणि येथे पूर्णपणे स्पष्ट गोष्टींबद्दल बोलू नका, जसे की एअरफ्लो स्थापित करणे, तुमचा निवडलेला डेटाबेस, सेलेरी आणि डॉक्समध्ये वर्णन केलेल्या इतर केसेस.

जेणेकरून आम्ही ताबडतोब प्रयोग सुरू करू शकू, मी रेखाटन केले docker-compose.yml ज्यामध्ये:

चला प्रत्यक्षात वाढवू वायुप्रवाह: शेड्युलर, वेबसर्व्हर. सेलरीच्या कामांचे निरीक्षण करण्यासाठी फ्लॉवर देखील तेथे फिरत असेल (कारण ते आधीच ढकलले गेले आहे apache/airflow:1.10.10-python3.7पण आमची हरकत नाही)
पोस्टग्रे एसक्यूएल, ज्यामध्ये एअरफ्लो त्याची सेवा माहिती (शेड्यूलर डेटा, अंमलबजावणी आकडेवारी इ.) लिहील आणि सेलेरी पूर्ण झालेल्या कार्यांना चिन्हांकित करेल;
Redis, जे सेलेरीसाठी टास्क ब्रोकर म्हणून काम करेल;
सेलेरी कामगार, जे कार्यांच्या थेट अंमलबजावणीमध्ये गुंतलेले असेल.
फोल्डर करण्यासाठी ./dags आम्ही आमच्या फायली डॅग्जच्या वर्णनासह जोडू. ते माशीवर उचलले जातील, म्हणून प्रत्येक शिंकल्यानंतर संपूर्ण स्टॅकला हात लावण्याची गरज नाही.

काही ठिकाणी, उदाहरणांमधील कोड पूर्णपणे दर्शविला जात नाही (जेणेकरुन मजकूर गोंधळून जाऊ नये), परंतु कुठेतरी प्रक्रियेत तो सुधारित केला जातो. संपूर्ण कार्य कोड उदाहरणे रेपॉजिटरीमध्ये आढळू शकतात https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

version: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

services:
  # Redis as a Celery broker
  broker:
    image: redis:6.0.5-alpine

  # DB for the Airflow metadata
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Main container with Airflow Webserver, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <<: *airflow-config

      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'
      AIRFLOW__SCHEDULER__MAX_THREADS: 8

      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10

    depends_on:
      - airflow-db
      - broker

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Airflow Webserver
      - 8080:8080

  # Celery worker, will be scaled using `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <<: *airflow-config

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

टिपा:

रचनेच्या असेंब्लीमध्ये, मी मोठ्या प्रमाणावर सुप्रसिद्ध प्रतिमेवर अवलंबून होतो पुकल/डॉकर-एअरफ्लो - हे नक्की पहा. कदाचित तुम्हाला तुमच्या आयुष्यात इतर कशाचीही गरज नसेल.
सर्व एअरफ्लो सेटिंग्ज केवळ द्वारे उपलब्ध नाहीत airflow.cfg, परंतु पर्यावरणीय व्हेरिएबल्सद्वारे देखील (विकासकांना धन्यवाद), ज्याचा मी दुर्भावनापूर्णपणे फायदा घेतला.
स्वाभाविकच, ते उत्पादनासाठी तयार नाही: मी जाणीवपूर्वक कंटेनरवर हृदयाचे ठोके ठेवले नाहीत, मला सुरक्षिततेचा त्रास झाला नाही. पण मी आमच्या प्रयोगकर्त्यांसाठी किमान योग्य ते केले.
लक्षात ठेवा की:
- डॅग फोल्डर शेड्यूलर आणि कामगार दोघांनाही प्रवेशयोग्य असणे आवश्यक आहे.
- हे सर्व तृतीय-पक्ष लायब्ररींना लागू होते - ते सर्व शेड्यूलर आणि कामगारांसह मशीनवर स्थापित केले जाणे आवश्यक आहे.

बरं, आता हे सोपे आहे:

$ docker-compose up --scale worker=3

सर्वकाही वाढल्यानंतर, आपण वेब इंटरफेस पाहू शकता:

एअरफ्लो: http://127.0.0.1:8080/admin/
फ्लॉवर: http://127.0.0.1:5555/dashboard

मूलभूत संकल्पना

जर तुम्हाला या सर्व "डॅग्ज" मधील काहीही समजले नसेल, तर येथे एक लहान शब्दकोश आहे:

शेड्यूलर - एअरफ्लोमधील सर्वात महत्वाचे काका, जे रोबोट कठोर परिश्रम करतात यावर नियंत्रण ठेवतात, व्यक्ती नाही: वेळापत्रकाचे निरीक्षण करतात, डॅग्ज अपडेट करतात, कार्ये लाँच करतात.
सर्वसाधारणपणे, जुन्या आवृत्त्यांमध्ये, त्याला स्मरणशक्तीची समस्या होती (नाही, स्मृतिभ्रंश नाही, परंतु लीक) आणि लेगसी पॅरामीटर अगदी कॉन्फिगमध्ये राहिले. run_duration - त्याचे रीस्टार्ट मध्यांतर. पण आता सर्व काही ठीक आहे.
दिवस (उर्फ "डॅग") - "दिग्दर्शित अॅसायक्लिक आलेख", परंतु अशी व्याख्या काही लोकांना सांगेल, परंतु प्रत्यक्षात ते एकमेकांशी संवाद साधणार्‍या कार्यांसाठी कंटेनर आहे (खाली पहा) किंवा SSIS मधील पॅकेज आणि इन्फॉर्मेटिका मधील वर्कफ्लोचे अॅनालॉग .
डॅग्ज व्यतिरिक्त, अजूनही सबडॅग असू शकतात, परंतु बहुधा आम्हाला ते मिळणार नाहीत.
DAG धाव - आरंभ केलेला डॅग, जो स्वतःचा नियुक्त केला जातो execution_date. समान डेगचे डॅगन समांतरपणे कार्य करू शकतात (जर तुम्ही तुमची कार्ये निःसंशयपणे केली असतील तर).
ऑपरेटर विशिष्ट क्रिया करण्यासाठी जबाबदार असलेल्या कोडचे तुकडे आहेत. तीन प्रकारचे ऑपरेटर आहेत:
- कारवाईआमच्या आवडत्याप्रमाणे PythonOperator, जो कोणताही (वैध) पायथन कोड कार्यान्वित करू शकतो;
- हस्तांतरण, जे ठिकाणाहून डेटा वाहतूक करतात, म्हणतात, MsSqlToHiveTransfer;
- सेन्सर दुसरीकडे, एखादी घटना घडेपर्यंत ते तुम्हाला प्रतिक्रिया देण्यास किंवा डॅगच्या पुढील अंमलबजावणीची गती कमी करण्यास अनुमती देईल. HttpSensor निर्दिष्ट एंडपॉइंट खेचू शकतो, आणि जेव्हा इच्छित प्रतिसाद प्रतीक्षा करत असेल तेव्हा हस्तांतरण सुरू करा GoogleCloudStorageToS3Operator. एक जिज्ञासू मन विचारेल: “का? शेवटी, तुम्ही ऑपरेटरमध्येच पुनरावृत्ती करू शकता!” आणि मग, निलंबित ऑपरेटरसह कार्यांचा पूल रोखू नये म्हणून. पुढील प्रयत्नापूर्वी सेन्सर सुरू होतो, तपासतो आणि मरतो.
कार्य - घोषित ऑपरेटर, प्रकाराकडे दुर्लक्ष करून, आणि डॅगशी संलग्न केलेल्या कार्याच्या रँकवर पदोन्नत केले जातात.
कार्य उदाहरण - जेव्हा सामान्य नियोजकाने ठरवले की कामगिरी करणार्‍या कामगारांवर लढाईत कार्ये पाठवण्याची वेळ आली आहे (जर आपण वापरत असाल तर LocalExecutor किंवा च्या बाबतीत रिमोट नोडला CeleryExecutor), ते त्यांना संदर्भ नियुक्त करते (म्हणजे, व्हेरिएबल्सचा एक संच - अंमलबजावणी पॅरामीटर्स), कमांड किंवा क्वेरी टेम्पलेट्स विस्तृत करते आणि त्यांना एकत्र करते.

आम्ही कार्ये व्युत्पन्न करतो

प्रथम, आमच्या डगच्या सामान्य योजनेची रूपरेषा काढूया, आणि नंतर आम्ही अधिकाधिक तपशीलांमध्ये जाऊ, कारण आम्ही काही क्षुल्लक उपाय लागू करतो.

तर, त्याच्या सर्वात सोप्या स्वरूपात, असा डग यासारखा दिसेल:

from datetime import timedelta, datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

from commons.datasources import sql_server_ds

dag = DAG('orders',
          schedule_interval=timedelta(hours=6),
          start_date=datetime(2020, 7, 8, 0))

def workflow(**context):
    print(context)

for conn_id, schema in sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

चला ते शोधूया:

प्रथम, आम्ही आवश्यक libs आयात करतो आणि काहीतरी;
sql_server_ds - हे आहे List[namedtuple[str, str]] एअरफ्लो कनेक्शन्सच्या कनेक्शनच्या नावांसह आणि डेटाबेसेस ज्यामधून आम्ही आमची प्लेट घेऊ;
dag - आमच्या डेगची घोषणा, जी अपरिहार्यपणे असणे आवश्यक आहे globals(), अन्यथा Airflow ला ते सापडणार नाही. डग हे देखील म्हणायचे आहे:
- त्याचे नाव काय orders - हे नाव नंतर वेब इंटरफेसमध्ये दिसेल,
- तो आठ जुलैच्या मध्यरात्रीपासून काम करेल,
- आणि ते चालले पाहिजे, अंदाजे दर 6 तासांनी (त्याऐवजी येथे कठीण मुलांसाठी timedelta() स्वीकार्य cron-ओळ 0 0 0/6 ? * * *, कमी थंड साठी - एक अभिव्यक्ती जसे @daily);
workflow() मुख्य काम करेल, पण आता नाही. आत्तासाठी, आम्ही आमचा संदर्भ लॉगमध्ये टाकू.
आणि आता कार्ये तयार करण्याची साधी जादू:
- आम्ही आमच्या स्त्रोतांमधून धावतो;
- आरंभ करा PythonOperator, जे आमच्या डमीची अंमलबजावणी करेल workflow(). टास्कचे अनन्य (डेगमध्ये) नाव निर्दिष्ट करण्यास विसरू नका आणि डॅग स्वतःच बांधा. झेंडा provide_context यामधून, फंक्शनमध्ये अतिरिक्त युक्तिवाद ओततो, ज्याचा वापर करून आम्ही काळजीपूर्वक गोळा करू **context.

आतासाठी, एवढेच. आम्हाला काय मिळाले:

वेब इंटरफेसमध्ये नवीन डेग,
दीडशे कार्ये समांतरपणे पार पाडली जातील (जर एअरफ्लो, सेलेरी सेटिंग्ज आणि सर्व्हरची क्षमता परवानगी देत असेल).

बरं, जवळजवळ समजलं.

अवलंबित्व कोण स्थापित करेल?

ही संपूर्ण गोष्ट सोपी करण्यासाठी, मी स्क्रू केले docker-compose.yml प्रक्रिया requirements.txt सर्व नोड्स वर.

आता ते गेले आहे:

ग्रे स्क्वेअर हे शेड्युलरद्वारे प्रक्रिया केलेले कार्य उदाहरण आहेत.

आम्ही थोडी प्रतीक्षा करतो, कार्ये कामगारांनी पूर्ण केली आहेत:

हिरवे, अर्थातच, त्यांचे काम यशस्वीरित्या पूर्ण केले आहे. रेड्स फारसे यशस्वी नाहीत.

तसे, आमच्या उत्पादनावर कोणतेही फोल्डर नाही ./dags, मशीन्समध्ये कोणतेही सिंक्रोनाइझेशन नाही - सर्व डॅग्ज आत आहेत git आमच्या Gitlab वर, आणि Gitlab CI मशिनमध्ये विलीन झाल्यावर अपडेट वितरित करते master.

फ्लॉवर बद्दल थोडे

कामगार आमच्या पॅसिफायरला मारत असताना, आपण आणखी एक साधन लक्षात ठेवू जे आपल्याला काहीतरी दर्शवू शकते - फ्लॉवर.

कामगार नोड्सवरील सारांश माहितीसह पहिले पृष्ठ:

कार्य करण्यासाठी गेलेल्या कार्यांसह सर्वात तीव्र पृष्ठ:

आमच्या ब्रोकरच्या स्थितीसह सर्वात कंटाळवाणा पृष्ठ:

सर्वात उजळ पृष्ठ हे कार्य स्थिती आलेख आणि त्यांच्या अंमलबजावणीच्या वेळेसह आहे:

आम्ही अंडरलोडेड लोड करतो

तर, सर्व कार्ये पूर्ण झाली आहेत, आपण जखमींना घेऊन जाऊ शकता.

आणि तेथे बरेच जखमी झाले - एक किंवा दुसर्या कारणास्तव. एअरफ्लोच्या योग्य वापराच्या बाबतीत, हे चौरस सूचित करतात की डेटा निश्चितपणे आला नाही.

तुम्हाला लॉग पाहण्याची आणि पडलेल्या कार्य उदाहरणे रीस्टार्ट करण्याची आवश्यकता आहे.

कोणत्याही स्क्वेअरवर क्लिक करून, आम्ही आमच्यासाठी उपलब्ध क्रिया पाहू:

तुम्ही घेऊ शकता आणि क्लिअर द फॉलन करू शकता. म्हणजेच, आम्ही विसरतो की तेथे काहीतरी अयशस्वी झाले आहे आणि तेच उदाहरण कार्य शेड्यूलरकडे जाईल.

हे स्पष्ट आहे की सर्व लाल चौरसांसह माऊससह हे करणे फारसे मानवीय नाही - हे आम्ही एअरफ्लोकडून अपेक्षित नाही. साहजिकच, आमच्याकडे मोठ्या प्रमाणावर विनाश करणारी शस्त्रे आहेत: Browse/Task Instances

चला सर्व काही एकाच वेळी निवडा आणि शून्यावर रीसेट करू, योग्य आयटमवर क्लिक करा:

साफसफाई केल्यानंतर, आमच्या टॅक्सी अशा दिसतात (ते शेड्यूलर शेड्यूल करण्यासाठी आधीच वाट पाहत आहेत):

कनेक्शन, हुक आणि इतर चल

पुढील डीएजी पाहण्याची वेळ आली आहे, update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='DWH Reports updated',
    html_content=dedent("""Господа хорошие, отчеты обновлены"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Наташ, просыпайся, мы {{ dag.dag_id }} уронили
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

प्रत्येकाने कधी अहवाल अपडेट केला आहे का? ही पुन्हा तिची आहे: डेटा कोठून मिळवायचा याची स्त्रोतांची यादी आहे; कुठे ठेवायचे याची यादी आहे; जेव्हा सर्वकाही घडले किंवा तुटले तेव्हा हॉंक करण्यास विसरू नका (ठीक आहे, हे आपल्याबद्दल नाही, नाही).

चला पुन्हा फाइलमध्ये जाऊ आणि नवीन अस्पष्ट सामग्री पाहू:

from commons.operators import TelegramBotSendMessage - आम्हाला आमचे स्वतःचे ऑपरेटर बनवण्यापासून काहीही प्रतिबंधित करत नाही, ज्याचा आम्ही अनब्लॉक केलेले संदेश पाठवण्यासाठी एक छोटा आवरण बनवून फायदा घेतला. (आम्ही खाली या ऑपरेटरबद्दल अधिक बोलू);
default_args={} - dag त्याच्या सर्व ऑपरेटरना समान युक्तिवाद वितरित करू शकते;
to='{{ var.value.all_the_kings_men }}' - फील्ड to आमच्याकडे हार्डकोड नसेल, परंतु जिन्जा वापरून डायनॅमिकली व्युत्पन्न केले जाईल आणि ईमेलच्या सूचीसह एक व्हेरिएबल, जे मी काळजीपूर्वक ठेवले आहे Admin/Variables;
trigger_rule=TriggerRule.ALL_SUCCESS — ऑपरेटर सुरू करण्याची अट. आमच्या बाबतीत, सर्व अवलंबित्व पूर्ण झाले असल्यासच पत्र बॉसकडे जाईल यशस्वीरित्या;
tg_bot_conn_id='tg_main' - युक्तिवाद conn_id आम्ही तयार केलेले कनेक्शन आयडी स्वीकारतो Admin/Connections;
trigger_rule=TriggerRule.ONE_FAILED - पडलेल्या कार्ये असल्यासच टेलिग्राममधील संदेश उडून जातील;
task_concurrency=1 - आम्ही एका कार्याच्या अनेक कार्य उदाहरणे एकाच वेळी लाँच करण्यास प्रतिबंधित करतो. अन्यथा, आम्हाला अनेकांचे एकाचवेळी प्रक्षेपण मिळेल VerticaOperator (एका टेबलकडे पहात आहे);
report_update >> [email, tg] - सर्व VerticaOperator याप्रमाणे पत्रे आणि संदेश पाठविण्यात एकत्र येणे:

परंतु नोटिफायर ऑपरेटर्सची लॉन्च परिस्थिती भिन्न असल्याने, फक्त एकच कार्य करेल. ट्री व्ह्यूमध्ये, सर्व काही थोडे कमी व्हिज्युअल दिसते:

मी याबद्दल काही शब्द सांगेन मॅक्रो आणि त्यांचे मित्र - चल.

मॅक्रो हे जिन्जा प्लेसहोल्डर आहेत जे ऑपरेटर वितर्कांमध्ये विविध उपयुक्त माहिती बदलू शकतात. उदाहरणार्थ, यासारखे:

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} संदर्भ व्हेरिएबलच्या सामग्रीमध्ये विस्तारित होईल execution_date स्वरूपात YYYY-MM-DD: 2020-07-14. सर्वोत्कृष्ट भाग असा आहे की संदर्भ व्हेरिएबल्स एका विशिष्ट कार्य उदाहरणावर (ट्री व्ह्यूमधील स्क्वेअर) नेल केले जातात आणि रीस्टार्ट केल्यावर, प्लेसहोल्डर समान मूल्यांमध्ये विस्तृत होतील.

नियुक्त केलेली मूल्ये प्रत्येक कार्य उदाहरणावर प्रस्तुत बटण वापरून पाहिली जाऊ शकतात. पत्र पाठवण्याचे कार्य असे आहे:

आणि म्हणून संदेश पाठवण्याच्या कार्यावर:

नवीनतम उपलब्ध आवृत्तीसाठी अंगभूत मॅक्रोची संपूर्ण यादी येथे उपलब्ध आहे: मॅक्रो संदर्भ

शिवाय, प्लगइनच्या मदतीने, आम्ही आमचे स्वतःचे मॅक्रो घोषित करू शकतो, परंतु ती दुसरी गोष्ट आहे.

पूर्वनिर्धारित गोष्टींव्यतिरिक्त, आम्ही आमच्या व्हेरिएबल्सची मूल्ये बदलू शकतो (मी हे आधीच वरील कोडमध्ये वापरले आहे). च्या मध्ये तयार करू Admin/Variables काही गोष्टी:

आपण वापरू शकता सर्वकाही:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

मूल्य स्केलर असू शकते किंवा ते JSON देखील असू शकते. JSON च्या बाबतीत:

bot_config

{
    "bot": {
        "token": 881hskdfASDA16641,
        "name": "Verter"
    },
    "service": "TG"
}

फक्त इच्छित कीचा मार्ग वापरा: {{ var.json.bot_config.bot.token }}.

मी अक्षरशः एक शब्द बोलेन आणि एक स्क्रीनशॉट दाखवेन कनेक्शन. येथे सर्व काही प्राथमिक आहे: पृष्ठावर Admin/Connections आम्ही एक कनेक्शन तयार करतो, आमचे लॉगिन / पासवर्ड आणि अधिक विशिष्ट पॅरामीटर्स तेथे जोडतो. याप्रमाणे:

पासवर्ड एन्क्रिप्ट केले जाऊ शकतात (डीफॉल्टपेक्षा अधिक पूर्णपणे), किंवा तुम्ही कनेक्शन प्रकार सोडू शकता (जसे मी केले tg_main) - वस्तुस्थिती अशी आहे की प्रकारांची यादी एअरफ्लो मॉडेल्समध्ये हार्डवायर केलेली आहे आणि स्त्रोत कोडमध्ये प्रवेश केल्याशिवाय विस्तारित केली जाऊ शकत नाही (अचानक मी काहीतरी गुगल केले नाही तर, कृपया मला दुरुस्त करा), परंतु काहीही आम्हाला क्रेडिट्स मिळण्यापासून रोखणार नाही नाव

आपण एकाच नावाने अनेक कनेक्शन देखील करू शकता: या प्रकरणात, पद्धत BaseHook.get_connection(), जे आम्हाला नावाने कनेक्शन मिळवते, देईल यादृच्छिक अनेक नावांमधून (राऊंड रॉबिन बनवणे अधिक तर्कसंगत असेल, परंतु एअरफ्लो डेव्हलपरच्या विवेकावर सोडूया).

व्हेरिएबल्स आणि कनेक्शन ही नक्कीच छान साधने आहेत, परंतु शिल्लक न गमावणे महत्वाचे आहे: तुमच्या प्रवाहाचे कोणते भाग तुम्ही कोडमध्येच साठवता आणि स्टोरेजसाठी तुम्ही एअरफ्लोला कोणते भाग देता. एकीकडे, UI द्वारे त्वरीत मूल्य बदलणे सोयीचे असू शकते, उदाहरणार्थ, मेलिंग बॉक्स. दुसरीकडे, हे अजूनही माऊस क्लिकवर परत येणे आहे, ज्यापासून आम्हाला (मी) सुटका हवी होती.

कनेक्शनसह कार्य करणे हे कार्यांपैकी एक आहे हुक. सर्वसाधारणपणे, एअरफ्लो हुक हे तृतीय-पक्ष सेवा आणि लायब्ररीशी कनेक्ट करण्यासाठी पॉइंट आहेत. उदा. JiraHook जिराशी संवाद साधण्यासाठी आमच्यासाठी क्लायंट उघडेल (तुम्ही कामे पुढे-मागे हलवू शकता), आणि त्याच्या मदतीने SambaHook तुम्ही स्थानिक फाइल पुश करू शकता smb-बिंदू.

सानुकूल ऑपरेटरचे विश्लेषण

आणि आम्ही ते कसे बनवले आहे ते पाहण्याच्या जवळ पोहोचलो TelegramBotSendMessage

कोड commons/operators.py वास्तविक ऑपरेटरसह:

from typing import Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Send message to chat_id using TelegramBotHook

    Example:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Union[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Send "{self.message}" to the chat {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

येथे, एअरफ्लोमधील इतर सर्व गोष्टींप्रमाणे, सर्वकाही अगदी सोपे आहे:

कडून वारसा मिळाला BaseOperator, जे काही एअरफ्लो-विशिष्ट गोष्टी लागू करते (तुमच्या विश्रांतीकडे पहा)
घोषित फील्ड template_fields, ज्यामध्ये Jinja प्रक्रिया करण्यासाठी मॅक्रो शोधेल.
साठी योग्य युक्तिवाद मांडले __init__(), आवश्यक तेथे डीफॉल्ट सेट करा.
आम्ही पूर्वजांच्या प्रारंभाबद्दल देखील विसरलो नाही.
संबंधित हुक उघडला TelegramBotHookत्यातून क्लायंट ऑब्जेक्ट प्राप्त झाला.
अधिलिखित (पुन्हा परिभाषित) पद्धत BaseOperator.execute(), जेव्हा ऑपरेटर लॉन्च करण्याची वेळ येईल तेव्हा कोणता Airfow ट्विच करेल - त्यात आम्ही लॉग इन करणे विसरून मुख्य क्रिया अंमलात आणू. (आम्ही लॉग इन करतो, तसे, अगदी आत stdout и stderr - एअरफ्लो सर्वकाही रोखेल, ते सुंदरपणे गुंडाळेल, आवश्यक असेल तेथे ते विघटित करेल.)

आपण काय आहे ते पाहूया commons/hooks.py. फाइलचा पहिला भाग, हुकसह:

from typing import Union

from airflow.hooks.base_hook import BaseHook
from requests_toolbelt.sessions import BaseUrlSession

class TelegramBotHook(BaseHook):
    """Telegram Bot API hook

    Note: add a connection with empty Conn Type and don't forget
    to fill Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

मला येथे काय स्पष्ट करावे हे देखील माहित नाही, मी फक्त महत्वाचे मुद्दे लक्षात ठेवतो:

आम्ही वारसा घेतो, वितर्कांचा विचार करा - बहुतेक प्रकरणांमध्ये ते एक असेल: conn_id;
मानक पद्धती ओव्हरराइड करणे: मी स्वतःला मर्यादित केले get_conn(), ज्यामध्ये मला नावानुसार कनेक्शन पॅरामीटर्स मिळतात आणि फक्त विभाग मिळतो extra (हे एक JSON फील्ड आहे), ज्यामध्ये मी (माझ्या स्वतःच्या सूचनांनुसार!) टेलीग्राम बॉट टोकन ठेवले आहे: {"bot_token": "YOuRAwEsomeBOtToKen"}.
मी आमचे एक उदाहरण तयार करतो TelegramBot, त्याला विशिष्ट टोकन देऊन.

इतकंच. आपण वापरून हुक पासून क्लायंट मिळवू शकता TelegramBotHook().clent किंवा TelegramBotHook().get_conn().

आणि फाइलचा दुसरा भाग, ज्यामध्ये मी टेलीग्राम REST API साठी मायक्रोरॅपर बनवतो, जेणेकरून ते ड्रॅग होऊ नये. python-telegram-bot एका पद्धतीसाठी sendMessage.

class TelegramBot:
    """Telegram Bot API wrapper

    Examples:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hi, darling')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hi, darling', chat_id=-1762374628374)
    """
    API_ENDPOINT = 'https://api.telegram.org/bot{}/'

    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):
        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)
        self.session = BaseUrlSession(self._base_url)
        self.chat_id = chat_id

    def send_message(self, message: str, chat_id: Union[int, str] = None):
        method = 'sendMessage'

        payload = {'chat_id': chat_id or self.chat_id,
                   'text': message,
                   'parse_mode': 'MarkdownV2'}

        response = self.session.post(method, data=payload).json()
        if not response.get('ok'):
            raise TelegramBotException(response)

class TelegramBotException(Exception):
    def __init__(self, *args, **kwargs):
        super().__init__((args, kwargs))

हे सर्व जोडण्याचा योग्य मार्ग आहे: TelegramBotSendMessage, TelegramBotHook, TelegramBot - प्लगइनमध्ये, सार्वजनिक भांडारात ठेवा आणि ते मुक्त स्त्रोताला द्या.

आम्ही या सर्वांचा अभ्यास करत असताना, आमचे अहवाल अपडेट यशस्वीरित्या अयशस्वी झाले आणि मला चॅनेलमध्ये एक त्रुटी संदेश पाठवला. ते चुकीचे आहे का ते मी तपासणार आहे...

आमच्या कुत्र्यात काहीतरी तुटले! आम्हाला तेच अपेक्षित होतं ना? नक्की!

ओतणार आहात का?

माझे काही चुकले असे तुम्हाला वाटते का? असे दिसते की त्याने एसक्यूएल सर्व्हरवरून व्हर्टिकामध्ये डेटा हस्तांतरित करण्याचे वचन दिले होते, आणि नंतर त्याने तो घेतला आणि विषय सोडून दिला, बदमाश!

हा अत्याचार हेतुपुरस्सर होता, मला तुमच्यासाठी काही शब्दावली उलगडायची होती. आता आपण पुढे जाऊ शकता.

आमची योजना अशी होती:

डेग करा
कार्ये व्युत्पन्न करा
सर्व काही किती सुंदर आहे ते पहा
भरण्यासाठी सत्र क्रमांक नियुक्त करा
SQL सर्व्हर वरून डेटा मिळवा
व्हर्टिका मध्ये डेटा ठेवा
आकडेवारी गोळा करा

म्हणून, हे सर्व चालू ठेवण्यासाठी, मी आमच्यामध्ये एक छोटीशी भर टाकली docker-compose.yml:

docker-compose.db.yml

version: '3.4'

x-mssql-base: &mssql-base
  image: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04
  restart: always
  environment:
    ACCEPT_EULA: Y
    MSSQL_PID: Express
    SA_PASSWORD: SayThanksToSatiaAt2020
    MSSQL_MEMORY_LIMIT_MB: 1024

services:
  dwh:
    image: jbfavre/vertica:9.2.0-7_ubuntu-16.04

  mssql_0:
    <<: *mssql-base

  mssql_1:
    <<: *mssql-base

  mssql_2:
    <<: *mssql-base

  mssql_init:
    image: mio101/py3-sql-db-client-base
    command: python3 ./mssql_init.py
    depends_on:
      - mssql_0
      - mssql_1
      - mssql_2
    environment:
      SA_PASSWORD: SayThanksToSatiaAt2020
    volumes:
      - ./mssql_init.py:/mssql_init.py
      - ./dags/commons/datasources.py:/commons/datasources.py

तेथे आम्ही वाढवतो:

व्हर्टिका यजमान म्हणून dwh सर्वात डीफॉल्ट सेटिंग्जसह,
SQL सर्व्हरची तीन उदाहरणे,
आम्ही नंतरचे डेटाबेस काही डेटासह भरतो (कोणत्याही परिस्थितीत पाहू नका mssql_init.py!)

आम्ही मागील वेळेपेक्षा किंचित अधिक क्लिष्ट कमांडच्या मदतीने सर्व चांगले लॉन्च करतो:

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

आमच्या चमत्कारी यादृच्छिकतेने काय तयार केले, आपण आयटम वापरू शकता Data Profiling/Ad Hoc Query:

मुख्य गोष्ट विश्लेषकांना दाखवणे नाही

च्या वर तपशीलवार ETL सत्रे मी करणार नाही, तिथे सर्व काही क्षुल्लक आहे: आम्ही एक आधार बनवतो, त्यात एक चिन्ह आहे, आम्ही प्रत्येक गोष्ट संदर्भ व्यवस्थापकासह गुंडाळतो आणि आता आम्ही हे करतो:

with Session(task_name) as session:
    print('Load', session.id, 'started')

    # Load worflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

from sys import stderr

class Session:
    """ETL workflow session

    Example:
        with Session(task_name) as session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Well done'
    """

    def __init__(self, connection, task_name):
        self.connection = connection
        self.connection.autocommit = True

        self._task_name = task_name
        self._id = None

        self.loaded_rows = None
        self.successful = None
        self.comment = None

    def __enter__(self):
        return self.open()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if any(exc_type, exc_val, exc_tb):
            self.successful = False
            self.comment = f'{exc_type}: {exc_val}n{exc_tb}'
            print(exc_type, exc_val, exc_tb, file=stderr)
        self.close()

    def __repr__(self):
        return (f'<{self.__class__.__name__} ' 
                f'id={self.id} ' 
                f'task_name="{self.task_name}">')

    @property
    def task_name(self):
        return self._task_name

    @property
    def id(self):
        return self._id

    def _execute(self, query, *args):
        with self.connection.cursor() as cursor:
            cursor.execute(query, args)
            return cursor.fetchone()[0]

    def _create(self):
        query = """
            CREATE TABLE IF NOT EXISTS sessions (
                id          SERIAL       NOT NULL PRIMARY KEY,
                task_name   VARCHAR(200) NOT NULL,

                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,
                finished    TIMESTAMPTZ           DEFAULT current_timestamp,
                successful  BOOL,

                loaded_rows INT,
                comment     VARCHAR(500)
            );
            """
        self._execute(query)

    def open(self):
        query = """
            INSERT INTO sessions (task_name, finished)
            VALUES (%s, NULL)
            RETURNING id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'opened')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Session is not open')
        query = """
            UPDATE sessions
            SET
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            WHERE
                id = %s
            RETURNING id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'closed',
              ', successful: ', self.successful,
              ', Loaded: ', self.loaded_rows,
              ', comment:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

वेळ आली आहे आमचा डेटा गोळा करा आमच्या दीडशे टेबलांवरून. चला हे अगदी नम्र ओळींच्या मदतीने करूया:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()

query = f"""
    SELECT 
        id, start_time, end_time, type, data
    FROM dbo.Orders
    WHERE
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)

हुकच्या मदतीने आम्ही एअरफ्लोमधून मिळवतो pymssql- कनेक्ट करा
चला विनंतीमध्ये तारखेच्या रूपात प्रतिबंध बदलू - ते टेम्पलेट इंजिनद्वारे फंक्शनमध्ये टाकले जाईल.
आमच्या विनंतीला आहार देणे pandasकोण आम्हाला मिळेल DataFrame - भविष्यात ते आम्हाला उपयुक्त ठरेल.

मी प्रतिस्थापन वापरत आहे {dt} विनंती पॅरामीटर ऐवजी %s मी एक दुष्ट पिनोचियो आहे म्हणून नाही तर pandas हाताळू शकत नाही pymssql आणि शेवटचा सरकतो params: Listजरी त्याला खरोखर हवे आहे tuple.
हे देखील लक्षात ठेवा की विकसक pymssql त्याला यापुढे पाठिंबा न देण्याचा निर्णय घेतला आणि आता बाहेर पडण्याची वेळ आली आहे pyodbc.

एअरफ्लोने आमच्या फंक्शन्सचे युक्तिवाद कशासह भरले ते पाहूया:

जर डेटा नसेल, तर पुढे चालू ठेवण्यात काही अर्थ नाही. परंतु भरणे यशस्वी मानणे देखील विचित्र आहे. पण ही चूक नाही. ए-आह-आह, काय करावे ?! आणि येथे काय आहे:

if df.empty:
    raise AirflowSkipException('No rows to load')

AirflowSkipException एअरफ्लोला सांगते की कोणत्याही त्रुटी नाहीत, परंतु आम्ही कार्य वगळतो. इंटरफेसमध्ये हिरवा किंवा लाल चौरस नसून गुलाबी असेल.

चला आमचा डेटा टाकूया एकाधिक स्तंभ:

df['etl_source'] = src_schema
df['etl_id'] = session.id
df['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

म्हणजे:

आम्ही ज्या डेटाबेसमधून ऑर्डर घेतली,
आमच्या पूर सत्राचा आयडी (तो वेगळा असेल प्रत्येक कामासाठी),
स्त्रोत आणि ऑर्डर आयडी वरून हॅश - जेणेकरून अंतिम डेटाबेसमध्ये (जेथे सर्व काही एका टेबलमध्ये ओतले जाते) आमच्याकडे एक अद्वितीय ऑर्डर आयडी आहे.

शेवटची पायरी शिल्लक आहे: सर्व काही व्हर्टिकामध्ये घाला. आणि, विचित्रपणे, हे करण्याचा सर्वात नेत्रदीपक आणि कार्यक्षम मार्गांपैकी एक म्हणजे CSV!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

आम्ही एक विशेष रिसीव्हर बनवत आहोत StringIO.
pandas कृपया आमच्या ठेवू DataFrame फॉर्म मध्ये CSV-रेषा.
चला आमच्या आवडत्या व्हर्टिकाला हुकसह कनेक्शन उघडूया.
आणि आता मदतीने copy() आमचा डेटा थेट Vertika वर पाठवा!

आम्ही ड्रायव्हरकडून किती ओळी भरल्या आहेत ते घेऊ आणि सत्र व्यवस्थापकाला सांगू की सर्व काही ठीक आहे:

session.loaded_rows = cursor.rowcount
session.successful = True

एवढेच.

विक्रीवर, आम्ही लक्ष्य प्लेट स्वतः तयार करतो. येथे मी स्वत: ला एक लहान मशीन परवानगी दिली:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

मी वापरत आहे VerticaOperator() मी डेटाबेस स्कीमा आणि टेबल तयार करतो (जर ते आधीपासून अस्तित्वात नसेल तर नक्कीच). मुख्य गोष्ट म्हणजे अवलंबनांची योग्यरित्या व्यवस्था करणे:

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

गोळा करीत आहे

- बरं, - लहान उंदीर म्हणाला, - आता नाही का
मी जंगलातील सर्वात भयंकर प्राणी आहे याची तुम्हाला खात्री पटली आहे का?

ज्युलिया डोनाल्डसन, द ग्रुफेलो

मला वाटते की माझ्या सहकार्‍यांमध्ये आणि माझ्यात स्पर्धा असेल तर: कोण पटकन सुरवातीपासून ETL प्रक्रिया तयार करेल आणि लॉन्च करेल: ते त्यांच्या SSIS आणि माउससह आणि मी एअरफ्लोसह ... आणि मग आम्ही देखभाल सुलभतेची तुलना देखील करू ... व्वा, मला वाटते की मी त्यांना सर्व आघाड्यांवर पराभूत करेन हे तुम्ही मान्य कराल!

जर थोडे अधिक गंभीरपणे पाहिले तर, अपाचे एअरफ्लो - प्रोग्राम कोडच्या रूपात प्रक्रियांचे वर्णन करून - माझे काम केले खूप अधिक आरामदायक आणि आनंददायक.

त्याची अमर्यादित विस्तारक्षमता, प्लग-इन आणि स्केलेबिलिटीची पूर्वस्थिती या दोन्ही बाबतीत, आपल्याला जवळजवळ कोणत्याही क्षेत्रात एअरफ्लो वापरण्याची संधी देते: अगदी डेटा गोळा करणे, तयार करणे आणि प्रक्रिया करणे या संपूर्ण चक्रात, अगदी रॉकेट्स (मंगळावर, च्या प्रक्षेपणात) अर्थात).

भाग अंतिम, संदर्भ आणि माहिती

आम्ही तुमच्यासाठी गोळा केलेला रेक

start_date. होय, हे आधीच स्थानिक मेम आहे. वाया डगचा मुख्य युक्तिवाद start_date सर्व पास. थोडक्यात, आपण निर्दिष्ट केल्यास start_date वर्तमान तारीख, आणि schedule_interval - एक दिवस, नंतर डीएजी उद्या सुरू होईल.
```
start_date = datetime(2020, 7, 7, 0, 1, 2)
```
आणि आणखी समस्या नाहीत.

त्याच्याशी संबंधित आणखी एक रनटाइम त्रुटी आहे: Task is missing the start_date parameter, जे बहुतेकदा सूचित करते की आपण डॅग ऑपरेटरला बांधायला विसरलात.
सर्व एकाच मशीनवर. होय, आणि बेस (एअरफ्लो स्वतः आणि आमचे कोटिंग), आणि एक वेब सर्व्हर, आणि शेड्यूलर आणि कामगार. आणि ते कामही केले. परंतु कालांतराने, सेवांसाठी कार्यांची संख्या वाढली, आणि जेव्हा PostgreSQL ने 20 ms ऐवजी 5 s मध्ये निर्देशांकाला प्रतिसाद देण्यास सुरुवात केली, तेव्हा आम्ही ते घेतले आणि दूर नेले.
LocalExecutor. होय, आम्ही अजूनही त्यावर बसलो आहोत आणि आम्ही आधीच अथांग डोहाच्या काठावर आलो आहोत. LocalExecutor आतापर्यंत आमच्यासाठी पुरेसे आहे, परंतु आता किमान एका कामगारासह विस्तारित करण्याची वेळ आली आहे आणि आम्हाला CeleryExecutor वर जाण्यासाठी कठोर परिश्रम करावे लागतील. आणि आपण एका मशीनवर त्याच्यासह कार्य करू शकता हे लक्षात घेता, सर्व्हरवर देखील सेलेरी वापरण्यापासून काहीही थांबवत नाही, जे "अर्थात, प्रामाणिकपणे कधीही उत्पादनात जाणार नाही!"
न वापरणे अंगभूत साधने:
- जोडण्या सेवा क्रेडेन्शियल्स साठवण्यासाठी,
- SLA चुकतो वेळेवर पूर्ण न झालेल्या कामांना प्रतिसाद देणे,
- xcom मेटाडेटा एक्सचेंजसाठी (मी म्हणालो मेटाडेटा!) डॅग टास्क दरम्यान.
मेलचा गैरवापर. बरं, मी काय सांगू? पडलेल्या कामांच्या सर्व पुनरावृत्तीसाठी अलर्ट सेट केले गेले. आता माझे कार्य Gmail मध्ये Airflow कडून >90k ईमेल आहेत आणि वेब मेल थूथन एका वेळी 100 हून अधिक उचलण्यास आणि हटविण्यास नकार देते.

आणखी तोटे: अपाचे एअरफ्लो पिटफेल्स

अधिक ऑटोमेशन साधने

आपल्या हातांनी नव्हे तर आपल्या डोक्याने अधिक कार्य करण्यासाठी, एअरफ्लोने आपल्यासाठी हे तयार केले आहे:

आरईएसटी API - त्याच्याकडे अजूनही प्रायोगिक स्थिती आहे, जी त्याला काम करण्यापासून रोखत नाही. त्याद्वारे, तुम्ही केवळ डॅग्ज आणि टास्कची माहिती मिळवू शकत नाही, तर डॅग थांबवू/सुरू करू शकता, डीएजी रन किंवा पूल तयार करू शकता.
CLI - कमांड लाइनद्वारे अनेक साधने उपलब्ध आहेत जी फक्त WebUI द्वारे वापरण्यासाठी गैरसोयीची नाहीत, परंतु सामान्यतः अनुपस्थित आहेत. उदाहरणार्थ:
- backfill कार्य उदाहरणे रीस्टार्ट करण्यासाठी आवश्यक आहे.
  उदाहरणार्थ, विश्लेषक आले आणि म्हणाले: “आणि कॉम्रेड, तुमच्याकडे 1 ते 13 जानेवारीच्या डेटामध्ये मूर्खपणा आहे! दुरुस्त करा, दुरुस्त करा, दुरुस्त करा, दुरुस्त करा!" आणि आपण असे हॉब आहात:
```
airflow backfill -s '2020-01-01' -e '2020-01-13' orders
```
- बेस सेवा: initdb, resetdb, upgradedb, checkdb.
- run, जे तुम्हाला एक उदाहरण कार्य चालविण्यास आणि सर्व अवलंबनांवर स्कोअर करण्यास अनुमती देते. शिवाय, आपण ते द्वारे चालवू शकता LocalExecutor, तुमच्याकडे सेलेरी क्लस्टर असला तरीही.
- जवळजवळ समान गोष्ट करते test, फक्त बेस मध्ये देखील काहीही लिहित नाही.
- connections शेलमधून मोठ्या प्रमाणात कनेक्शन तयार करण्यास अनुमती देते.
पायथन एपीआय - संवाद साधण्याचा एक ऐवजी हार्डकोर मार्ग, जो प्लगइन्ससाठी आहे आणि त्यात लहान हातांनी झुंडू देत नाही. पण आम्हाला जाण्यापासून कोण रोखणार /home/airflow/dags, धावणे ipython आणि गोंधळ सुरू? आपण, उदाहरणार्थ, खालील कोडसह सर्व कनेक्शन निर्यात करू शकता:
```
from airflow import settings
from airflow.models import Connection

fields = 'conn_id conn_type host port schema login password extra'.split()

session = settings.Session()
for conn in session.query(Connection).order_by(Connection.conn_id):
  d = {field: getattr(conn, field) for field in fields}
  print(conn.conn_id, '=', d)
```

एअरफ्लो मेटाडेटाबेसशी कनेक्ट करत आहे. मी त्यावर लिहिण्याची शिफारस करत नाही, परंतु विविध विशिष्ट मेट्रिक्ससाठी कार्य स्थिती मिळवणे कोणत्याही API वापरण्यापेक्षा बरेच जलद आणि सोपे असू शकते.

आपण असे म्हणूया की आपली सर्व कार्ये उदात्त नसतात, परंतु ती कधीकधी पडू शकतात आणि हे सामान्य आहे. परंतु काही अडथळे आधीच संशयास्पद आहेत आणि ते तपासणे आवश्यक आहे.

एसक्यूएल सावध रहा!

WITH last_executions AS (
SELECT
    task_id,
    dag_id,
    execution_date,
    state,
        row_number()
        OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC) AS rn
FROM public.task_instance
WHERE
    execution_date > now() - INTERVAL '2' DAY
),
failed AS (
    SELECT
        task_id,
        dag_id,
        execution_date,
        state,
        CASE WHEN rn = row_number() OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC)
                 THEN TRUE END AS last_fail_seq
    FROM last_executions
    WHERE
        state IN ('failed', 'up_for_retry')
)
SELECT
    task_id,
    dag_id,
    count(last_fail_seq)                       AS unsuccessful,
    count(CASE WHEN last_fail_seq
        AND state = 'failed' THEN 1 END)       AS failed,
    count(CASE WHEN last_fail_seq
        AND state = 'up_for_retry' THEN 1 END) AS up_for_retry
FROM failed
GROUP BY
    task_id,
    dag_id
HAVING
    count(last_fail_seq) > 0

संदर्भ

आणि अर्थातच, Google जारी केल्यापासून पहिले दहा दुवे माझ्या बुकमार्कमधील एअरफ्लो फोल्डरमधील सामग्री आहेत.

अपाचे एअरफ्लो दस्तऐवजीकरण - अर्थातच, आपण कार्यालयापासून सुरुवात केली पाहिजे. दस्तऐवजीकरण, पण सूचना कोण वाचतो?
चांगला सराव - बरं, कमीतकमी निर्मात्यांच्या शिफारसी वाचा.
एअरफ्लो UI - अगदी सुरुवात: चित्रांमधील वापरकर्ता इंटरफेस
Apache Airflow च्या प्रमुख संकल्पना समजून घेणे - मूलभूत संकल्पना चांगल्या प्रकारे वर्णन केल्या आहेत, जर (अचानक!) तुम्हाला माझ्याकडून काहीतरी समजले नाही.
Tianlong's Blog — एअरफ्लो सर्व्हर/क्लस्टर कसे तयार करावे याबद्दल मार्गदर्शक - एअरफ्लो क्लस्टर सेट करण्यासाठी एक लहान मार्गदर्शक.
Lyft वर Apache Airflow चालू आहे - जवळजवळ समान मनोरंजक लेख, कदाचित अधिक औपचारिकता आणि कमी उदाहरणे वगळता.
अपाचे एअरफ्लो सेलेरी कामगारांवर नोकऱ्यांचे वितरण कसे करते - सेलेरीच्या संयोगाने काम करण्याबद्दल.
अपाचे एअरफ्लोमध्ये DAG लेखन सर्वोत्तम पद्धती - कार्यांची क्षमता, तारखेऐवजी आयडीद्वारे लोड करणे, परिवर्तन, फाइल संरचना आणि इतर मनोरंजक गोष्टी.
अपाचे एअरफ्लोमध्ये अवलंबित्व व्यवस्थापित करणे - कार्यांचे अवलंबन आणि ट्रिगर नियम, ज्याचा मी फक्त पासिंगमध्ये उल्लेख केला आहे.
एअरफ्लो: जेव्हा तुमचा DAG शेड्यूलच्या मागे असतो - शेड्युलरमधील काही "उद्देशानुसार कार्य" कसे मात करावे, गमावलेला डेटा लोड करा आणि कार्यांना प्राधान्य कसे द्या.
Apache Airflow साठी उपयुक्त SQL क्वेरी — एअरफ्लो मेटाडेटाला उपयुक्त SQL क्वेरी.
Apache Airflow सह वर्कफ्लो विकसित करण्यास सुरुवात करा - सानुकूल सेन्सर तयार करण्याबद्दल एक उपयुक्त विभाग आहे.
Presto आणि Airflow सह AWS वर Fetchr डेटा सायन्स इन्फ्रा तयार करणे — डेटा सायन्ससाठी AWS वर पायाभूत सुविधा तयार करण्याबद्दल एक मनोरंजक छोटी टीप.
एअरफ्लो DAGs डीबग करताना तपासण्यासाठी 7 सामान्य त्रुटी - सामान्य चुका (जेव्हा कोणीतरी सूचना वाचत नाही).
Apache Airflow वापरून पासवर्ड साठवा आणि प्रवेश करा - तुम्ही फक्त कनेक्शन वापरू शकता तरीही लोक पासवर्ड साठवून ठेवतात कसे ते स्मित करा.
पायथन आणि अपाचे एअरफ्लोचा झेन - अंतर्निहित डीएजी फॉरवर्डिंग, फंक्शन्समध्ये कॉन्टेक्स्ट थ्रोइंग, पुन्हा अवलंबनांबद्दल आणि टास्क लॉन्च वगळण्याबद्दल.
वायुप्रवाह: कमी ज्ञात टिपा, युक्त्या आणि सर्वोत्तम पद्धती - वापराबद्दल default arguments и params टेम्पलेट्स, तसेच व्हेरिएबल्स आणि कनेक्शन्समध्ये.
एअरफ्लो शेड्यूलर प्रोफाइल करणे - प्लॅनर एअरफ्लो 2.0 साठी कशी तयारी करत आहे याची एक कथा.
डॉकर-कंपोजमध्ये 3 सेलेरी कामगारांसह Apache Airflow - मध्ये आमचे क्लस्टर तैनात करण्याबद्दल थोडा जुना लेख docker-compose.
एअरफ्लो संदर्भ वापरून 4 टेम्प्लेटिंग कार्ये - टेम्पलेट्स आणि संदर्भ फॉरवर्डिंग वापरून डायनॅमिक कार्ये.
एअरफ्लोमध्ये त्रुटी सूचना — मेल आणि स्लॅकद्वारे मानक आणि सानुकूल सूचना.
एअरफ्लो वर्कशॉप: क्रॅचशिवाय जटिल DAGs - शाखा कार्ये, मॅक्रो आणि XCom.

आणि लेखात वापरलेले दुवे:

मॅक्रो संदर्भ - टेम्पलेट्समध्ये वापरण्यासाठी प्लेसहोल्डर्स उपलब्ध आहेत.
सामान्य नुकसान - हवेचा प्रवाह - डॅग तयार करताना सामान्य चुका.
puckel/docker-airflow: Docker Apache Airflow - docker-compose प्रयोग, डीबगिंग आणि अधिकसाठी.
python-telegram-bot/python-telegram-bot: आम्ही तुम्हाला एक आवरण बनवले आहे जे तुम्ही नाकारू शकत नाही — टेलिग्राम REST API साठी पायथन रॅपर.

स्त्रोत: www.habr.com