Apache Airflow: ETL को आसान बनाना

नमस्ते, मैं दिमित्री लोगविनेंको हूं - वेज़ेट समूह की कंपनियों के एनालिटिक्स विभाग का डेटा इंजीनियर।

मैं आपको ईटीएल प्रक्रियाओं को विकसित करने के लिए एक अद्भुत उपकरण - अपाचे एयरफ्लो के बारे में बताऊंगा। लेकिन एयरफ़्लो इतना बहुमुखी और बहुमुखी है कि आपको इस पर करीब से नज़र डालनी चाहिए, भले ही आप डेटा प्रवाह में शामिल न हों, लेकिन समय-समय पर किसी भी प्रक्रिया को लॉन्च करने और उनके निष्पादन की निगरानी करने की आवश्यकता हो।

और हां, मैं न केवल बताऊंगा, बल्कि दिखाऊंगा: कार्यक्रम में बहुत सारे कोड, स्क्रीनशॉट और सिफारिशें हैं।

जब आप गूगल पर एयरफ़्लो/विकिमीडिया कॉमन्स शब्द देखते हैं तो आप आमतौर पर क्या देखते हैं

लेख-सूची

• परिचय
• मुख्य भाग, व्यावहारिक (और थोड़ा सैद्धांतिक)
  • हम (और आप) ऐसा क्यों करते हैं
  • क्लस्टर को असेंबल करना
  • बुनियादी अवधारणाओं
  • हम कार्य उत्पन्न करते हैं
  • फूल के बारे में थोड़ा
  • हम अंडरलोडेड लोड करते हैं
  • कनेक्शन, हुक और अन्य चर
  • कस्टम ऑपरेटर को पार्स करना
  • क्या आप डालने जा रहे हैं?
  • उपसंहार
• भाग अंतिम, संदर्भ एवं जानकारी
  • रेक हमने आपके लिए एकत्र किया है
  • अधिक स्वचालन उपकरण
• संदर्भों

परिचय

अपाचे एयरफ़्लो बिल्कुल Django की तरह है:

• पायथन में लिखा गया है
• एक बेहतरीन एडमिन पैनल है,
• अनिश्चित काल तक विस्तार योग्य

- केवल बेहतर, और इसे पूरी तरह से अलग उद्देश्यों के लिए बनाया गया था, अर्थात् (जैसा कि कैट से पहले लिखा गया है):

• असीमित संख्या में मशीनों पर कार्य चलाना और निगरानी करना (जितनी सेलेरी/कुबेरनेट्स और आपका विवेक आपको अनुमति देगा)
• पाइथन कोड को लिखने और समझने में बहुत आसान से गतिशील वर्कफ़्लो पीढ़ी के साथ
• और तैयार घटकों और घरेलू प्लगइन्स (जो बेहद सरल है) दोनों का उपयोग करके किसी भी डेटाबेस और एपीआई को एक दूसरे से जोड़ने की क्षमता।

हम Apache Airflow का उपयोग इस प्रकार करते हैं:

• हम DWH और ODS (हमारे पास वर्टिका और क्लिकहाउस हैं) में विभिन्न स्रोतों (कई SQL सर्वर और PostgreSQL उदाहरण, एप्लिकेशन मेट्रिक्स के साथ विभिन्न API, यहां तक ​​कि 1C) से डेटा एकत्र करते हैं।
• कितना उन्नत cron, जो ओडीएस पर डेटा समेकन प्रक्रिया शुरू करता है, और उनके रखरखाव की निगरानी भी करता है।

कुछ समय पहले तक, हमारी ज़रूरतें 32 कोर और 50 जीबी रैम वाले एक छोटे सर्वर द्वारा पूरी की जाती थीं। एयरफ्लो में, यह काम करता है:

• अधिक है 200 डेग (वास्तव में वर्कफ़्लो, जिसमें हमने कार्य भरे हैं),
• प्रत्येक में औसतन 70 कार्य,
• यह अच्छाई शुरू होती है (औसतन भी) एक घंटे में एक बार.

और हमने कैसे विस्तार किया, इसके बारे में मैं नीचे लिखूंगा, लेकिन अब उबर-समस्या को परिभाषित करते हैं जिसे हम हल करेंगे:

तीन स्रोत SQL सर्वर हैं, प्रत्येक में 50 डेटाबेस हैं - क्रमशः एक परियोजना के उदाहरण, उनकी संरचना समान है (लगभग हर जगह, मुआ-हा-हा), जिसका अर्थ है कि प्रत्येक में एक ऑर्डर तालिका है (सौभाग्य से, उसके साथ एक तालिका) नाम को किसी भी व्यवसाय में धकेला जा सकता है)। हम सेवा फ़ील्ड (स्रोत सर्वर, स्रोत डेटाबेस, ईटीएल कार्य आईडी) जोड़कर डेटा लेते हैं और उन्हें वर्टिका कहते हैं, में फेंक देते हैं।

चलो चलें!

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

हम (और आप) ऐसा क्यों करते हैं

जब पेड़ बड़े थे और मैं सरल था SQL-एक रूसी रिटेल में, हमने हमारे पास उपलब्ध दो उपकरणों का उपयोग करके ईटीएल प्रक्रियाओं उर्फ ​​​​डेटा प्रवाह में घोटाला किया:

• इंफॉर्मेटिका पावर सेंटर - एक बेहद फैलने वाली प्रणाली, बेहद उत्पादक, अपने स्वयं के हार्डवेयर, अपने स्वयं के संस्करण के साथ। मैंने भगवान न करे इसकी क्षमताओं का 1% उपयोग किया। क्यों? खैर, सबसे पहले, यह इंटरफ़ेस, लगभग 380 के दशक का, मानसिक रूप से हम पर दबाव डालता है। दूसरे, यह उपकरण अत्यधिक फैंसी प्रक्रियाओं, उग्र घटक पुन: उपयोग और अन्य बहुत महत्वपूर्ण-उद्यम-चालों के लिए डिज़ाइन किया गया है। इस तथ्य के बारे में कि इसकी लागत एयरबस एXNUMX/वर्ष के विंग की तरह है, हम कुछ नहीं कहेंगे।

  सावधान, एक स्क्रीनशॉट 30 वर्ष से कम उम्र के लोगों को थोड़ा नुकसान पहुंचा सकता है

  

• SQL सर्वर एकीकरण सर्वर - हमने इस कॉमरेड का उपयोग अपने इंट्रा-प्रोजेक्ट फ्लो में किया। खैर, वास्तव में: हम पहले से ही SQL सर्वर का उपयोग करते हैं, और इसके ETL टूल का उपयोग न करना किसी तरह से अनुचित होगा। इसमें सब कुछ अच्छा है: इंटरफ़ेस सुंदर है, और प्रगति रिपोर्ट ... लेकिन यही कारण है कि हम सॉफ़्टवेयर उत्पादों को पसंद करते हैं, ओह, इसके लिए नहीं। इसका संस्करण dtsx (जो XML है जिसमें सेव पर नोड्स को फेरबदल किया गया है) हम कर सकते हैं, लेकिन इसका मतलब क्या है? एक कार्य पैकेज बनाने के बारे में क्या ख्याल है जो सैकड़ों तालिकाओं को एक सर्वर से दूसरे सर्वर तक खींच लेगा? हाँ, सौ क्या, माउस बटन पर क्लिक करते ही आपकी तर्जनी के बीस टुकड़े हो जायेंगे। लेकिन यह निश्चित रूप से अधिक फैशनेबल दिखता है:

  

हमने निश्चित रूप से बाहर निकलने के रास्ते तलाशे। मामला भी लगभग एक स्व-लिखित एसएसआईएस पैकेज जनरेटर आया ...

...और फिर मुझे एक नई नौकरी मिल गई। और Apache Airflow इस मामले में मुझसे आगे निकल गया।

जब मुझे पता चला कि ईटीएल प्रक्रिया विवरण सरल पायथन कोड हैं, तो मैं खुशी से नाच नहीं पाया। इस प्रकार डेटा स्ट्रीम को संस्करणित और भिन्न किया गया, और सैकड़ों डेटाबेस से एक ही संरचना के साथ तालिकाओं को एक लक्ष्य में डालना डेढ़ या दो 13 ”स्क्रीन में पायथन कोड का मामला बन गया।

क्लस्टर को असेंबल करना

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

ताकि हम तुरंत प्रयोग शुरू कर सकें, मैंने रेखाचित्र बनाया docker-compose.yml जिसमें:

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

कुछ स्थानों पर, उदाहरणों में कोड पूरी तरह से नहीं दिखाया गया है (ताकि पाठ अव्यवस्थित न हो), लेकिन कहीं-कहीं इसे प्रक्रिया में संशोधित किया गया है। संपूर्ण कार्य कोड उदाहरण रिपॉजिटरी में पाए जा सकते हैं https://github.com/dm-logv/airflow-tutorial.

डोकर-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 - इसका पुनरारंभ अंतराल। लेकिन अब सब कुछ ठीक है.

• डेग (उर्फ "डेग") - "निर्देशित एसाइक्लिक ग्राफ", लेकिन ऐसी परिभाषा कुछ लोगों को बताएगी, लेकिन वास्तव में यह एक दूसरे के साथ बातचीत करने वाले कार्यों के लिए एक कंटेनर है (नीचे देखें) या एसएसआईएस में पैकेज और इंफॉर्मेटिका में वर्कफ़्लो का एक एनालॉग .

  डैग के अलावा, अभी भी सबडैग हो सकते हैं, लेकिन सबसे अधिक संभावना है कि हम उन तक नहीं पहुंच पाएंगे।

• डीएजी रन - प्रारंभिक डैग, जिसे अपना स्वयं का असाइन किया गया है 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(), अन्यथा एयरफ़्लो इसे नहीं ढूंढ पाएगा। डौग को यह भी कहना होगा:
  • इसका नाम क्या है 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={} - डैग अपने सभी ऑपरेटरों को समान तर्क वितरित कर सकता है;
• 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(), जो हमें नाम से कनेक्शन दिलवाता है, देगा अनियमित कई नामों से (राउंड रॉबिन बनाना अधिक तर्कसंगत होगा, लेकिन आइए इसे एयरफ्लो डेवलपर्स के विवेक पर छोड़ दें)।

वेरिएबल्स और कनेक्शंस निश्चित रूप से अच्छे उपकरण हैं, लेकिन यह महत्वपूर्ण है कि संतुलन न खोएं: आपके प्रवाह के कौन से हिस्से आप कोड में ही संग्रहीत करते हैं, और कौन से हिस्से आप भंडारण के लिए एयरफ्लो को देते हैं। एक ओर, यूआई के माध्यम से मूल्य को जल्दी से बदलना सुविधाजनक हो सकता है, उदाहरण के लिए, एक मेलिंग बॉक्स। दूसरी ओर, यह अभी भी माउस क्लिक की वापसी है, जिससे हम (मैं) छुटकारा पाना चाहते थे।

कनेक्शन के साथ काम करना कार्यों में से एक है हुक. सामान्य तौर पर, एयरफ्लो हुक इसे तृतीय-पक्ष सेवाओं और पुस्तकालयों से जोड़ने के लिए बिंदु हैं। जैसे, 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, जिसमें जिंजा प्रोसेस करने के लिए मैक्रोज़ की तलाश करेगा।
• के लिए उचित तर्कों की व्यवस्था की __init__(), जहां आवश्यक हो वहां डिफ़ॉल्ट सेट करें।
• हम पूर्वज के आरंभीकरण के बारे में भी नहीं भूले।
• संबंधित हुक खोला TelegramBotHookइससे एक क्लाइंट ऑब्जेक्ट प्राप्त हुआ।
• ओवरराइड (पुनर्परिभाषित) विधि BaseOperator.execute(), जब ऑपरेटर लॉन्च करने का समय आएगा तो एयरफ़ॉ चिकोटी काटेगा - इसमें हम लॉग इन करना भूलकर मुख्य क्रिया लागू करेंगे। (वैसे, हम सीधे लॉग इन करते हैं 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().

और फ़ाइल का दूसरा भाग, जिसमें मैं टेलीग्राम रेस्ट एपीआई के लिए एक माइक्रोरैपर बनाता हूं, ताकि उसे खींचना न पड़े 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 सर्वर से वर्टिका में डेटा स्थानांतरित करने का वादा किया था, और फिर उसने इसे ले लिया और विषय से हट गया, बदमाश!

यह अत्याचार जानबूझकर किया गया था, मुझे बस आपके लिए कुछ शब्दावली समझनी थी। अब आप आगे बढ़ सकते हैं.

हमारी योजना यह थी:

1. डेग करो
2. कार्य उत्पन्न करें
3. देखो सब कुछ कितना सुंदर है
4. भरने के लिए सत्र संख्याएँ निर्दिष्ट करें
5. SQL सर्वर से डेटा प्राप्त करें
6. वर्टिका में डेटा डालें
7. आँकड़े एकत्र करें

तो, यह सब चालू करने के लिए, मैंने हमारे में एक छोटा सा योगदान किया 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:

मुख्य बात यह है कि इसे विश्लेषकों को न दिखाया जाए

इसे विस्तार से समझाइए ईटीएल सत्र मैं नहीं करूंगा, वहां सब कुछ तुच्छ है: हम एक आधार बनाते हैं, उसमें एक संकेत होता है, हम सब कुछ एक संदर्भ प्रबंधक के साथ लपेटते हैं, और अब हम यह करते हैं:

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)

1. एक हुक की मदद से हम एयरफ्लो से प्राप्त करते हैं pymssql-जोड़ना
2. आइए अनुरोध में दिनांक के रूप में एक प्रतिबंध को प्रतिस्थापित करें - इसे टेम्पलेट इंजन द्वारा फ़ंक्शन में डाल दिया जाएगा।
3. हमारे अनुरोध को खिलाना 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']])

अर्थात्

• जिस डेटाबेस से हमने ऑर्डर लिया,
• हमारे बाढ़ सत्र की आईडी (यह अलग होगी) हर कार्य के लिए),
• स्रोत और ऑर्डर आईडी से एक हैश - ताकि अंतिम डेटाबेस में (जहां सब कुछ एक तालिका में डाला जाता है) हमारे पास एक अद्वितीय ऑर्डर आईडी हो।

अंतिम चरण शेष है: सब कुछ वर्टिका में डालें। और, अजीब तरह से, ऐसा करने का सबसे शानदार और कुशल तरीकों में से एक सीएसवी के माध्यम से है!

# 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)

1. हम एक विशेष रिसीवर बना रहे हैं StringIO.
2. pandas कृपया अपना डाल देंगे DataFrame जैसा CSV-लाइनें.
3. आइए एक हुक के साथ हमारे पसंदीदा वर्टिका से एक कनेक्शन खोलें।
4. और अब मदद से copy() हमारा डेटा सीधे वर्टिका को भेजें!

हम ड्राइवर से लेंगे कि कितनी लाइनें भरी गईं, और सत्र प्रबंधक को बताएंगे कि सब कुछ ठीक है:

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

उपसंहार

- अच्छा, - छोटे चूहे ने कहा, - है ना, अभी
क्या तुम्हें यकीन है कि मैं जंगल का सबसे भयानक जानवर हूं?

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

मुझे लगता है कि अगर मेरे सहयोगियों और मेरे बीच एक प्रतियोगिता होती: कौन जल्दी से ईटीएल प्रक्रिया को शुरू से बनाएगा और लॉन्च करेगा: वे अपने एसएसआईएस और एक माउस के साथ और मैं एयरफ्लो के साथ ... और फिर हम रखरखाव में आसानी की भी तुलना करेंगे ... वाह, मुझे लगता है आप इस बात से सहमत होंगे कि मैं उन्हें सभी मोर्चों पर हरा दूंगा!

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

इसकी असीमित विस्तारशीलता, प्लग-इन और स्केलेबिलिटी की प्रवृत्ति दोनों के संदर्भ में, आपको लगभग किसी भी क्षेत्र में एयरफ्लो का उपयोग करने का अवसर देती है: यहां तक ​​​​कि डेटा एकत्र करने, तैयार करने और संसाधित करने के पूर्ण चक्र में, यहां तक ​​​​कि रॉकेट (मंगल ग्रह पर) लॉन्च करने में भी अवधि)।

भाग अंतिम, संदर्भ एवं जानकारी

रेक हमने आपके लिए एकत्र किया है

• 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 एमएस के बजाय 5 एस में सूचकांक पर प्रतिक्रिया देना शुरू किया, तो हमने इसे ले लिया और इसे दूर ले गए।
• स्थानीय निष्पादक. हाँ, हम अभी भी उस पर बैठे हैं, और हम पहले ही रसातल के किनारे पर आ चुके हैं। LocalExecutor अब तक हमारे लिए पर्याप्त रहा है, लेकिन अब कम से कम एक कार्यकर्ता के साथ विस्तार करने का समय आ गया है, और हमें CeleryExecutor में जाने के लिए कड़ी मेहनत करनी होगी। और इस तथ्य को ध्यान में रखते हुए कि आप इसके साथ एक मशीन पर काम कर सकते हैं, आपको सर्वर पर भी सेलेरी का उपयोग करने से कोई नहीं रोकता है, जो "बेशक, ईमानदारी से कभी भी उत्पादन में नहीं जाएगा!"
• अनुपयोगी अंतर्निर्मित उपकरण:
  • कनेक्शन सेवा क्रेडेंशियल संग्रहीत करने के लिए,
  • एसएलए चूक गया उन कार्यों का उत्तर देना जो समय पर पूरे नहीं हुए,
  • एक्सकॉम मेटाडेटा विनिमय के लिए (मैंने कहा मेटाडेटा!) डेग कार्यों के बीच।
• मेल दुरुपयोग. खैर, मेरी ओर से क्या कहा जा सकता है? गिरे हुए कार्यों की सभी पुनरावृत्तियों के लिए अलर्ट स्थापित किए गए थे। अब मेरे काम जीमेल में एयरफ्लो से 90 हजार ईमेल हैं, और वेब मेल थूथन एक समय में 100 से अधिक को लेने और हटाने से इनकार करता है।

अधिक नुकसान: अपाचे एयरफ्लो पिटफेल

अधिक स्वचालन उपकरण

हम अपने हाथों से नहीं बल्कि अपने सिरों से और भी अधिक काम कर सकें, इसके लिए एयरफ्लो ने हमारे लिए यह तैयारी की है:

• बाकी एपीआई - उनके पास अभी भी एक्सपेरिमेंटल का दर्जा है, जो उन्हें काम करने से नहीं रोकता है। इसके साथ, आप न केवल डैग और कार्यों के बारे में जानकारी प्राप्त कर सकते हैं, बल्कि एक डैग को रोक/शुरू भी कर सकते हैं, एक डीएजी रन या एक पूल बना सकते हैं।
• सीएलआई - कई उपकरण कमांड लाइन के माध्यम से उपलब्ध हैं जो वेबयूआई के माध्यम से उपयोग करने के लिए न केवल असुविधाजनक हैं, बल्कि आम तौर पर अनुपस्थित हैं। उदाहरण के लिए:
  • 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)

• एयरफ्लो मेटाडेटाबेस से कनेक्ट हो रहा है। मैं इसे लिखने की अनुशंसा नहीं करता, लेकिन विभिन्न विशिष्ट मेट्रिक्स के लिए कार्य स्थिति प्राप्त करना किसी भी एपीआई का उपयोग करने की तुलना में बहुत तेज़ और आसान हो सकता है।

  मान लीजिए कि हमारे सभी कार्य निरर्थक नहीं हैं, लेकिन कभी-कभी उनमें गिरावट आ सकती है, और यह सामान्य है। लेकिन कुछ रुकावटें पहले से ही संदिग्ध हैं, और उनकी जाँच करना आवश्यक होगा।

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

  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 के जारी होने के पहले दस लिंक मेरे बुकमार्क से एयरफ्लो फ़ोल्डर की सामग्री हैं।

• अपाचे एयरफ़्लो दस्तावेज़ीकरण - बेशक, हमें कार्यालय से शुरुआत करनी चाहिए। दस्तावेज़ीकरण, लेकिन निर्देश कौन पढ़ता है?
• सर्वोत्तम प्रथाएं - ठीक है, कम से कम रचनाकारों की अनुशंसाएँ पढ़ें।
• एयरफ्लो यूआई - बिल्कुल शुरुआत: चित्रों में यूजर इंटरफ़ेस
• अपाचे एयरफ्लो की प्रमुख अवधारणाओं को समझना - बुनियादी अवधारणाओं का अच्छी तरह से वर्णन किया गया है, अगर (अचानक!) आपको मुझसे कुछ समझ नहीं आया।
• तियानलॉन्ग का ब्लॉग - एयरफ्लो सर्वर/क्लस्टर कैसे बनाएं, इस पर एक गाइड - एयरफ़्लो क्लस्टर स्थापित करने के लिए एक संक्षिप्त मार्गदर्शिका।
• लिफ़्ट पर अपाचे एयरफ़्लो चलाना - लगभग वही दिलचस्प लेख, शायद अधिक औपचारिकता और कम उदाहरणों को छोड़कर।
• कैसे अपाचे एयरफ़्लो सेलेरी श्रमिकों को नौकरियाँ वितरित करता है — अजवाइन के साथ मिलकर काम करने के बारे में।
• डीएजी अपाचे एयरफ्लो में सर्वोत्तम अभ्यास लिख रहा है - कार्यों की निष्क्रियता, दिनांक के बजाय आईडी द्वारा लोड करना, परिवर्तन, फ़ाइल संरचना और अन्य दिलचस्प चीज़ों के बारे में।
• अपाचे एयरफ्लो में निर्भरता का प्रबंधन - कार्यों की निर्भरता और ट्रिगर नियम, जिसका उल्लेख मैंने केवल पारित होने में किया था।
• वायुप्रवाह: जब आपका डीएजी निर्धारित समय से काफी पीछे हो - शेड्यूलर में कुछ "कार्यों को इच्छानुसार" कैसे दूर करें, खोए हुए डेटा को लोड करें और कार्यों को प्राथमिकता दें।
• अपाचे एयरफ़्लो के लिए उपयोगी SQL क्वेरीज़ - एयरफ़्लो मेटाडेटा के लिए उपयोगी SQL क्वेरीज़।
• Apache Airflow के साथ वर्कफ़्लो विकसित करना शुरू करें - कस्टम सेंसर बनाने के बारे में एक उपयोगी अनुभाग है।
• प्रेस्टो और एयरफ़्लो के साथ AWS पर Fetchr डेटा साइंस इन्फ्रा का निर्माण - डेटा साइंस के लिए AWS पर बुनियादी ढांचे के निर्माण के बारे में एक दिलचस्प संक्षिप्त टिप्पणी।
• एयरफ्लो डीएजी को डीबग करते समय जाँचने योग्य 7 सामान्य त्रुटियाँ - सामान्य गलतियाँ (जब कोई अभी भी निर्देश नहीं पढ़ता है)।
• अपाचे एयरफ्लो का उपयोग करके पासवर्ड स्टोर करें और एक्सेस करें - मुस्कुराइए कि कैसे लोग पासवर्ड संग्रहीत करने में संकोच करते हैं, हालाँकि आप केवल कनेक्शंस का उपयोग कर सकते हैं।
• पायथन और अपाचे एयरफ्लो का ज़ेन - अंतर्निहित डीएजी अग्रेषण, कार्यों में संदर्भ फेंकना, फिर से निर्भरता के बारे में, और कार्य लॉन्च को छोड़ने के बारे में भी।
• वायु प्रवाह: कम ज्ञात युक्तियाँ, तरकीबें और सर्वोत्तम प्रथाएँ - उपयोग के बारे में default arguments и params टेम्प्लेट में, साथ ही वेरिएबल्स और कनेक्शंस में।
• एयरफ्लो शेड्यूलर की प्रोफाइलिंग - योजनाकार एयरफ़्लो 2.0 के लिए कैसे तैयारी कर रहा है, इसके बारे में एक कहानी।
• डॉकर-कंपोज़ में 3 सेलेरी श्रमिकों के साथ अपाचे एयरफ्लो - हमारे क्लस्टर को तैनात करने के बारे में थोड़ा पुराना लेख docker-compose.
• एयरफ़्लो संदर्भ का उपयोग करके 4 टेम्प्लेटिंग कार्य - टेम्प्लेट और संदर्भ अग्रेषण का उपयोग करके गतिशील कार्य।
• एयरफ़्लो में त्रुटि सूचनाएँ - मेल और स्लैक द्वारा मानक और कस्टम सूचनाएं।
• एयरफ्लो कार्यशाला: बैसाखी के बिना जटिल डीएजी - ब्रांचिंग कार्य, मैक्रोज़ और एक्सकॉम।

और लेख में प्रयुक्त लिंक:

• मैक्रो संदर्भ - टेम्प्लेट में उपयोग के लिए प्लेसहोल्डर उपलब्ध हैं।
• सामान्य हानियाँ- वायुप्रवाह - डैग बनाते समय सामान्य गलतियाँ।
• पकेल/डॉकर-एयरफ्लो: डॉकर अपाचे एयरफ्लो - docker-compose प्रयोग, डिबगिंग और बहुत कुछ के लिए।
• पायथन-टेलीग्राम-बॉट/पायथन-टेलीग्राम-बॉट: हमने आपके लिए एक ऐसा रैपर बनाया है जिसे आप मना नहीं कर सकते - टेलीग्राम रेस्ट एपीआई के लिए पायथन रैपर।

स्रोत: www.habr.com