Apache Airflow: ETL सजिलो बनाउँदै

नमस्ते, म दिमित्री लगभिनेन्को हुँ - कम्पनीहरूको भेजेट समूहको विश्लेषण विभागको डाटा इन्जिनियर।

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

र हो, म मात्र बताउने छैन, तर पनि देखाउँछु: कार्यक्रममा धेरै कोड, स्क्रिनसट र सिफारिसहरू छन्।

तपाईले सामान्यतया के देख्नुहुन्छ जब तपाईले Airflow / Wikimedia Commons शब्द गुगल गर्नुहुन्छ

सामग्रीको तालिका

• परिचय
• मुख्य भाग, व्यावहारिक (र थोरै सैद्धान्तिक)
  • किन हामी (र तपाईं)
  • क्लस्टर जम्मा गर्दै
  • आधारभूत अवधारणाहरू
  • हामी कार्यहरू उत्पन्न गर्छौं
  • फूलको बारेमा थोरै
  • हामी अन्डरलोड लोड गर्छौं
  • जडानहरू, हुकहरू र अन्य चरहरू
  • अनुकूलन अपरेटर पार्स गर्दै
  • के तपाईं खन्याउन जाँदै हुनुहुन्छ?
  • संक्षेप हुँदै
• भाग अन्तिम, सन्दर्भ र जानकारी
  • हामीले तपाईंको लागि सङ्कलन गरेको रेक
  • थप स्वचालन उपकरणहरू
• सन्दर्भ

परिचय

Apache Airflow Django जस्तै छ:

• python मा लेखिएको
• त्यहाँ एक महान प्रशासक प्यानल छ,
• अनिश्चित कालका लागि विस्तार योग्य

- केवल राम्रो, र यो पूर्ण रूपमा फरक उद्देश्यका लागि बनाइएको थियो, अर्थात् (जस्तै यो kat अघि लेखिएको छ):

• असीमित संख्यामा मेशिनहरूमा चलिरहेको र अनुगमन कार्यहरू (जति धेरै सेलेरी / कुबर्नेट्स र तपाईंको अन्तस्करणले तपाईंलाई अनुमति दिनेछ)
• पाइथन कोड लेख्न र बुझ्न धेरै सजिलो देखि गतिशील कार्यप्रवाह उत्पादन संग
• र कुनै पनि डाटाबेस र API हरू दुवै तयार-निर्मित कम्पोनेन्टहरू र घर-निर्मित प्लगइनहरू प्रयोग गरेर एकअर्कासँग जडान गर्ने क्षमता (जुन अत्यन्त सरल छ)।

हामी यसरी Apache Airflow प्रयोग गर्छौं:

• हामी DWH र ODS (हामीसँग Vertica र Clickhouse छ) मा विभिन्न स्रोतहरू (धेरै SQL सर्भर र PostgreSQL उदाहरणहरू, अनुप्रयोग मेट्रिकहरू सहित विभिन्न API, 1C पनि) बाट डेटा सङ्कलन गर्छौं।
• कति उन्नत cron, जसले ODS मा डाटा समेकन प्रक्रियाहरू सुरु गर्दछ, र तिनीहरूको मर्मतसम्भार पनि निगरानी गर्दछ।

भर्खरै सम्म, हाम्रा आवश्यकताहरू 32 कोर र 50 GB RAM भएको एउटा सानो सर्भरद्वारा कभर गरिएको थियो। Airflow मा, यो काम गर्दछ:

• अधिक 200 डगहरू (वास्तवमा कार्यप्रवाह, जसमा हामीले कार्यहरू भर्यौं),
• प्रत्येकमा औसतमा 70 कार्यहरू,
• यो भलाइ सुरु हुन्छ (औसतमा पनि) एक घण्टा एक पटक.

र हामीले कसरी विस्तार गर्यौं भन्ने बारे, म तल लेख्नेछु, तर अब हामी über-समस्यालाई परिभाषित गरौं जुन हामी समाधान गर्नेछौं:

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

चलो जानुहोस्!

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

किन हामी (र तपाईं)

जब रुखहरू ठूला थिए र म साधारण थिएँ SQL-एक रूसी खुद्रा मा schik, हामीले हामीलाई उपलब्ध दुई उपकरणहरू प्रयोग गरेर ETL प्रक्रियाहरू उर्फ ​​डेटा प्रवाह घोटाला गर्यौं:

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

  सावधान, स्क्रिनसटले ३० वर्षभन्दा कम उमेरका मानिसहरूलाई अलिकति चोट पुर्याउन सक्छ

  

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

  

हामीले पक्कै पनि बाटो खोज्यौं। मामला पनि लगभग स्व-लिखित SSIS प्याकेज जेनरेटरमा आयो ...

... र त्यसपछि मलाई नयाँ काम भेटियो। र अपाचे एयरफ्लोले मलाई यसमा ओभरटेक गर्यो।

जब मैले पत्ता लगाएँ कि ETL प्रक्रिया विवरणहरू साधारण पाइथन कोड हुन्, मैले आनन्दको लागि नाचिन। यसरी डाटा स्ट्रिमहरू संस्करण र भिन्न थिए, र सयौं डाटाबेसहरूबाट एउटै लक्ष्यमा एउटै संरचनासहितको तालिकाहरू डेढ वा दुई 13” स्क्रिनहरूमा पाइथन कोडको कुरा भयो।

क्लस्टर जम्मा गर्दै

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

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

• वास्तवमा उठौं हावा प्रवाह: अनुसूचीकर्ता, वेबसर्भर। सेलरी कार्यहरू अनुगमन गर्न फूल पनि त्यहाँ कताई हुनेछ (किनकि यो पहिले नै भित्र धकेलिएको छ। apache/airflow:1.10.10-python3.7तर हामीलाई कुनै आपत्ति छैन)
• PostgreSQL, जसमा Airflow ले आफ्नो सेवा जानकारी (शेड्युलर डेटा, कार्यान्वयन तथ्याङ्क, आदि) लेख्नेछ, र Celery ले सम्पन्न कार्यहरू चिन्ह लगाउनेछ;
• Redis, जसले सेलरीको लागि टास्क ब्रोकरको रूपमा काम गर्नेछ;
• सेलेरी कामदार, जुन कार्यहरूको प्रत्यक्ष कार्यान्वयनमा संलग्न हुनेछ।
• फोल्डरमा ./dags हामी 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

टिप्पणीहरू:

• रचना को विधानसभा मा, म धेरै हदसम्म प्रसिद्ध छवि मा निर्भर थिए puckel/docker- airflow - यसलाई जाँच गर्न निश्चित हुनुहोस्। सायद तपाईलाई तपाईको जीवनमा अरु केहि चाहिदैन।
• सबै एयरफ्लो सेटिङहरू मार्फत मात्र उपलब्ध छैनन् airflow.cfg, तर वातावरण चर मार्फत पनि (विकासकर्ताहरूलाई धन्यवाद), जुन मैले दुर्भावनापूर्ण रूपमा फाइदा उठाएँ।
• स्वाभाविक रूपमा, यो उत्पादन-तयार छैन: मैले जानाजानी कन्टेनरहरूमा मुटुको धड्कनहरू राखिन, मैले सुरक्षाको साथ चिन्ता गरेन। तर मैले हाम्रा प्रयोगकर्ताहरूको लागि न्यूनतम उपयुक्त गरें।
• नोट गर्नुहोस् कि:
  • डेग फोल्डर अनुसूचक र कार्यकर्ता दुवैको लागि पहुँचयोग्य हुनुपर्छ।
  • यो सबै तेस्रो-पक्ष पुस्तकालयहरूमा लागू हुन्छ - तिनीहरू सबै एक अनुसूचक र कामदारहरू संग मेशिन मा स्थापित हुनुपर्छ।

खैर, अब यो सरल छ:

$ docker-compose up --scale worker=3

सबै कुरा उठे पछि, तपाइँ वेब इन्टरफेसहरू हेर्न सक्नुहुन्छ:

• हावा प्रवाह: http://127.0.0.1:8080/admin/
• फूल: http://127.0.0.1:5555/dashboard

आधारभूत अवधारणाहरू

यदि तपाईंले यी सबै "डागहरू" मा केही बुझ्नुभएको छैन भने, यहाँ छोटो शब्दकोश छ:

• अनुसूची - एयरफ्लोमा सबैभन्दा महत्त्वपूर्ण अंकल, रोबोटले कडा मेहनत गर्छ भन्ने नियन्त्रण गर्दै, व्यक्ति होइन: तालिका अनुगमन गर्दछ, ड्यागहरू अपडेट गर्दछ, कार्यहरू सुरू गर्दछ।

  सामान्यतया, पुरानो संस्करणहरूमा, उसलाई मेमोरीमा समस्या थियो (होइन, एम्नेसिया होइन, तर लीक) र विरासत प्यारामिटर पनि कन्फिगमा रह्यो। run_duration - यसको पुन: सुरु अन्तराल। तर अहिले सबै ठीक छ।

• DAG (उर्फ "डेग") - "निर्देशित एसाइक्लिक ग्राफ", तर यस्तो परिभाषाले थोरै मानिसहरूलाई बताउनेछ, तर वास्तवमा यो एकअर्कासँग अन्तरक्रिया गर्ने कार्यहरूका लागि कन्टेनर हो (तल हेर्नुहोस्) वा SSIS मा प्याकेजको एनालग र Informatica मा Workflow। ।

  Dags को अतिरिक्त, त्यहाँ अझै पनि subdags हुन सक्छ, तर हामी सम्भवतः तिनीहरूमा पुग्न सक्दैनौं।

• 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.

फूलको बारेमा थोरै

जब कामदारहरूले हाम्रो प्यासिफायरहरू पिटिरहेका छन्, आउनुहोस्, हामीलाई केहि देखाउन सक्ने अर्को उपकरण सम्झौं - फूल।

कार्यकर्ता नोडहरूमा सारांश जानकारीको साथ पहिलो पृष्ठ:

काममा गएका कार्यहरूको साथ सबैभन्दा तीव्र पृष्ठ:

हाम्रो दलाल को स्थिति संग सबै भन्दा बोरिंग पृष्ठ:

उज्यालो पृष्ठ कार्य स्थिति ग्राफ र तिनीहरूको कार्यान्वयन समय संग छ:

हामी अन्डरलोड लोड गर्छौं

त्यसोभए, सबै कार्यहरू काम गरिसकेका छन्, तपाईं घाइतेहरूलाई टाढा लैजान सक्नुहुन्छ।

र त्यहाँ धेरै घाइते थिए - एक कारण वा अर्को लागि। Airflow को सही प्रयोगको मामलामा, यी धेरै वर्गहरूले डेटा निश्चित रूपमा आइपुगेन भनेर संकेत गर्दछ।

तपाईंले लग हेर्नु पर्छ र पतित कार्य उदाहरणहरू पुन: सुरु गर्न आवश्यक छ।

कुनै पनि वर्गमा क्लिक गरेर, हामी हाम्रो लागि उपलब्ध कार्यहरू देख्नेछौं:

तपाईं लिन सक्नुहुन्छ र पतन खाली गर्न सक्नुहुन्छ। त्यो हो, हामी बिर्सन्छौं कि त्यहाँ केहि असफल भएको छ, र उही उदाहरण कार्य अनुसूचकमा जान्छ।

यो स्पष्ट छ कि सबै रातो वर्गहरु संग माउस संग यो गर्न धेरै मानवीय छैन - यो हामीले Airflow बाट अपेक्षा के होइन। स्वाभाविक रूपमा, हामीसँग सामूहिक विनाशका हतियारहरू छन्: Browse/Task Instances

एकैचोटि सबै कुरा चयन गरौं र शून्यमा रिसेट गरौं, सही वस्तुमा क्लिक गर्नुहोस्:

सफा गरिसकेपछि, हाम्रा ट्याक्सीहरू यस्तो देखिन्छन् (तिनीहरू पहिले नै अनुसूचितकर्ताले तिनीहरूलाई तालिका बनाउनको लागि पर्खिरहेका छन्):

जडानहरू, हुकहरू र अन्य चरहरू

यो अर्को DAG हेर्ने समय हो, 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(), जसले हामीलाई नामबाट जडानहरू प्राप्त गर्दछ, दिनेछ अनियमित धेरै नामहरूबाट (यो राउन्ड रोबिन बनाउन बढी तार्किक हुनेछ, तर यसलाई एयरफ्लो विकासकर्ताहरूको विवेकमा छोडौं)।

चरहरू र जडानहरू पक्कै पनि राम्रो उपकरणहरू हुन्, तर ब्यालेन्स गुमाउनु हुँदैन: तपाईंको प्रवाहको कुन भागहरू तपाईंले कोडमा भण्डारण गर्नुहुन्छ, र कुन भागहरू तपाईंले भण्डारणको लागि Airflow लाई दिनुहुन्छ। एकातिर, यो चाँडै मान परिवर्तन गर्न सुविधाजनक हुन सक्छ, उदाहरणका लागि, मेलिङ बक्स, 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)

यहाँ, Airflow मा सबै चीज जस्तै, सबै धेरै सरल छ:

• बाट वंशानुगत BaseOperator, जसले केहि एयरफ्लो-विशिष्ट चीजहरू लागू गर्दछ (तपाईंको फुर्सदमा हेर्नुहोस्)
• घोषित क्षेत्रहरू template_fields, जसमा Jinja ले प्रक्रिया गर्न म्याक्रोहरू खोज्नेछ।
• को लागि सही तर्कहरू मिलाइयो __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 फिल्ड हो), जसमा मैले (मेरो आफ्नै निर्देशन अनुसार!) Telegram bot टोकन राखें: {"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 Server बाट Vertica मा डाटा स्थानान्तरण गर्ने वाचा गरेको थियो, र त्यसपछि उसले यसलाई लियो र विषय, स्काउन्डरेलबाट हट्यो!

यो क्रूरता जानाजानी थियो, मैले तपाइँको लागि केहि शब्दावली बुझाउनु थियो। अब तपाईं अगाडि बढ्न सक्नुहुन्छ।

हाम्रो योजना यस्तो थियो:

1. गर
2. कार्यहरू उत्पन्न गर्नुहोस्
3. हेरौ सबै कति सुन्दर छ
4. भर्नका लागि सत्र नम्बरहरू असाइन गर्नुहोस्
5. SQL सर्भरबाट डाटा प्राप्त गर्नुहोस्
6. Vertica मा डाटा राख्नुहोस्
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

त्यहाँ हामी उठाउँछौं:

• Vertica होस्टको रूपमा dwh सबैभन्दा पूर्वनिर्धारित सेटिङहरूसँग,
• SQL सर्भर को तीन उदाहरणहरू,
• हामी पछिल्लोमा डाटाबेसहरू भर्छौं केही डाटा (कुनै पनि अवस्थामा नहेर्नुहोस् mssql_init.py!)

हामीले पछिल्लो पटक भन्दा थोरै जटिल आदेशको मद्दतले सबै राम्रो सुरुवात गर्छौं:

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

के हाम्रो चमत्कार randomizer उत्पन्न, तपाईं वस्तु प्रयोग गर्न सक्नुहुन्छ 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)

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 Airflow लाई बताउने छ कि त्यहाँ कुनै त्रुटिहरू छैनन्, तर हामी कार्य छोड्छौं। इन्टरफेसमा हरियो वा रातो वर्ग हुनेछैन, तर गुलाबी।

हाम्रो डाटा टस गरौं धेरै स्तम्भहरू:

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

अर्थात्:

• हामीले आदेश लिएको डाटाबेस,
• हाम्रो बाढी सत्रको ID (यो फरक हुनेछ हरेक कार्यको लागि),
• स्रोत र अर्डर ID बाट ह्यास - ताकि अन्तिम डाटाबेसमा (जहाँ सबै कुरा एउटै टेबलमा राखिएको छ) हामीसँग एक अद्वितीय अर्डर आईडी छ।

अन्तिम चरण बाँकी छ: सबै कुरा Vertica मा खन्याउनुहोस्। र, अनौठो कुरा, यो गर्ने सबैभन्दा शानदार र प्रभावकारी तरिकाहरू मध्ये एक 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)

1. हामी एक विशेष रिसीभर बनाउँदैछौं StringIO.
2. pandas कृपया हाम्रो राख्नेछ DataFrame फारममा CSV-लाइनहरू।
3. हुकको साथ हाम्रो मनपर्ने Vertica मा जडान खोलौं।
4. र अब सहयोग संग 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 र माउस र म एयरफ्लोसँग ... र त्यसपछि हामी मर्मतसम्भारको सहजता पनि तुलना गर्नेछौं ... वाह, मलाई लाग्छ कि तपाईं सहमत हुनुहुनेछ कि म तिनीहरूलाई सबै मोर्चाहरूमा हराउनेछु!

यदि अलि बढी गम्भीर रूपमा, त्यसपछि Apache Airflow - कार्यक्रम कोडको रूपमा प्रक्रियाहरू वर्णन गरेर - मेरो काम गर्यो। धेरै थप सहज र रमाइलो।

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

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

हामीले तपाईंको लागि सङ्कलन गरेको रेक

• start_date। हो, यो पहिले नै स्थानीय मेम हो। डगको मुख्य तर्क मार्फत start_date सबै पास। संक्षेपमा, यदि तपाइँ निर्दिष्ट गर्नुहुन्छ start_date वर्तमान मिति, र schedule_interval - एक दिन, त्यसपछि DAG भोलि सुरु हुनेछ।

  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 मेटाडाटा विनिमयको लागि (मैले भने मेटाडाटा!) dag कार्यहरू बीच।
• मेल दुरुपयोग। खैर, म के भन्न सक्छु? अलर्टहरू पतन गरिएका कार्यहरूको पुनरावृत्तिको लागि सेट अप गरिएको थियो। अब मेरो काम Gmail मा Airflow बाट 90k इमेलहरू छन्, र वेब मेल थूथनले एक पटकमा 100 भन्दा बढी उठाउन र मेटाउन अस्वीकार गर्दछ।

थप समस्याहरू: Apache Airflow Pitfails

थप स्वचालन उपकरणहरू

हाम्रो हातले होइन हाम्रो टाउकोले अझ बढी काम गर्नको लागि, एयरफ्लोले हाम्रो लागि यो तयार गरेको छ:

• REST API - उहाँसँग अझै पनि प्रयोगात्मक स्थिति छ, जसले उहाँलाई काम गर्नबाट रोक्दैन। यसको साथ, तपाईंले ड्यागहरू र कार्यहरू बारे जानकारी मात्र प्राप्त गर्न सक्नुहुन्न, तर डेग रोक्न/सुरु गर्न, DAG रन वा पूल सिर्जना गर्न सक्नुहुन्छ।
• 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)

• Airflow मेटाडेटाबेस जडान गर्दै। म यसमा लेख्न सिफारिस गर्दिन, तर विभिन्न विशिष्ट मेट्रिक्सका लागि कार्य राज्यहरू प्राप्त गर्नु कुनै पनि API मार्फत भन्दा धेरै छिटो र सजिलो हुन सक्छ।

  भनौं कि हाम्रा सबै कार्यहरू निर्दोष छैनन्, तर तिनीहरू कहिलेकाहीं खस्न सक्छन्, र यो सामान्य हो। तर केही अवरोधहरू पहिले नै शंकास्पद छन्, र यो जाँच गर्न आवश्यक हुनेछ।

  SQL सावधान!

  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 को जारीबाट पहिलो दस लिङ्कहरू मेरो बुकमार्कहरूबाट Airflow फोल्डरको सामग्री हुन्।

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

र लेखमा प्रयोग गरिएका लिङ्कहरू:

• म्याक्रो सन्दर्भ - टेम्प्लेटहरूमा प्रयोगको लागि प्लेसहोल्डरहरू उपलब्ध छन्।
• सामान्य पिटफल्स - एयरफ्लो - ड्यागहरू सिर्जना गर्दा सामान्य गल्तीहरू।
• puckel/docker-airflow: Docker Apache Airflow - docker-compose प्रयोग, डिबगिङ र थपको लागि।
• python-telegram-bot/python-telegram-bot: हामीले तपाईलाई अस्वीकार गर्न नसक्ने र्यापर बनाएका छौं - टेलिग्राम REST API को लागि पाइथन र्यापर।

स्रोत: www.habr.com