Apache Airflow: ETL-ის გაადვილება

გამარჯობა, მე ვარ დიმიტრი ლოგვინენკო - კომპანია Vezet ჯგუფის ანალიტიკის დეპარტამენტის მონაცემთა ინჟინერი.

მე გეტყვით ETL პროცესების განვითარების შესანიშნავ ხელსაწყოზე - Apache Airflow. მაგრამ Airflow იმდენად მრავალმხრივი და მრავალმხრივია, რომ თქვენ უნდა დააკვირდეთ მას მაშინაც კი, თუ თქვენ არ ხართ ჩართული მონაცემთა ნაკადებში, მაგრამ გჭირდებათ პერიოდულად დაიწყოს ნებისმიერი პროცესი და აკონტროლოთ მათი შესრულება.

დიახ, მე არა მხოლოდ გეტყვით, არამედ ვაჩვენებ: პროგრამას აქვს ბევრი კოდი, ეკრანის სურათი და რეკომენდაციები.

რასაც ჩვეულებრივ ხედავთ სიტყვა Airflow / Wikimedia Commons-ში google-ში

სარჩევი

• შესავალი
• ძირითადი ნაწილი, პრაქტიკული (და ცოტა თეორიული)
  • რატომ ჩვენ (და თქვენ)
  • კლასტერის აწყობა
  • ძირითადი ცნებები
  • ჩვენ ვქმნით დავალებებს
  • ცოტა რამ ყვავილის შესახებ
  • დატვირთულს ვტვირთავთ
  • კავშირები, კაკვები და სხვა ცვლადები
  • მორგებული ოპერატორის გარჩევა
  • დალევას აპირებ?
  • შეჯამება
• ნაწილი საბოლოო, მითითება და ინფორმაცია
  • რაკი ჩვენ შევაგროვეთ თქვენთვის
  • მეტი ავტომატიზაციის ხელსაწყოები
• ლიტერატურა

შესავალი

Apache Airflow ისეთივეა, როგორც Django:

• პითონში დაწერილი
• არის შესანიშნავი ადმინისტრაციული პანელი,
• გაფართოვდება განუსაზღვრელი ვადით

- მხოლოდ უკეთესი, და ის გაკეთდა სრულიად განსხვავებული მიზნებისთვის, კერძოდ (როგორც კატას წინ წერია):

• ამოცანების გაშვება და მონიტორინგი შეუზღუდავი რაოდენობის მანქანებზე (რამდენიმე ნიახური/კუბერნეტი და თქვენი სინდისი მოგცემთ საშუალებას)
• დინამიური სამუშაო ნაკადის გენერირებით ძალიან მარტივი დასაწერი და გასაგები პითონის კოდიდან
• და ნებისმიერი მონაცემთა ბაზის და API-ების ერთმანეთთან დაკავშირების შესაძლებლობა როგორც მზა კომპონენტების, ასევე სახლში დამზადებული დანამატების გამოყენებით (რაც ძალიან მარტივია).

ჩვენ ვიყენებთ Apache Airflow ასე:

• ჩვენ ვაგროვებთ მონაცემებს სხვადასხვა წყაროებიდან (ბევრი SQL Server და PostgreSQL ინსტანციები, სხვადასხვა API აპლიკაციების მეტრიკით, თუნდაც 1C) DWH-ში და ODS-ში (გვაქვს Vertica და Clickhouse).
• რამდენად მოწინავე cron, რომელიც იწყებს მონაცემთა კონსოლიდაციის პროცესებს ODS-ზე და ასევე აკონტროლებს მათ შენარჩუნებას.

ბოლო დრომდე ჩვენს მოთხოვნილებებს ფარავდა ერთი პატარა სერვერი 32 ბირთვით და 50 GB ოპერატიული მეხსიერებით. Airflow-ში ეს მუშაობს:

• მეტი 200 დაგი (რეალურად სამუშაო ნაკადები, რომლებშიც ჩვენ დავამატეთ დავალებები),
• თითოეულში საშუალოდ 70 დავალება,
• ეს სიკეთე იწყება (ასევე საშუალოდ) საათში ერთხელ.

და იმის შესახებ, თუ როგორ გავფართოვდით, ქვემოთ დავწერ, მაგრამ ახლა განვსაზღვროთ über-პრობლემა, რომელსაც გადავჭრით:

არსებობს სამი წყარო SQL სერვერი, თითოეულს აქვს 50 მონაცემთა ბაზა - ერთი პროექტის მაგალითი, შესაბამისად, მათ აქვთ იგივე სტრუქტურა (თითქმის ყველგან, mua-ha-ha), რაც ნიშნავს, რომ თითოეულს აქვს Orders ცხრილი (საბედნიეროდ, ცხრილი ამ სახელი შეიძლება ნებისმიერ ბიზნესში იყოს ჩართული). ჩვენ ვიღებთ მონაცემებს სერვისის ველების დამატებით (წყაროს სერვერი, წყაროს მონაცემთა ბაზა, ETL ამოცანის ID) და გულუბრყვილოდ ვყრით მათ, ვთქვათ, Vertica-ში.

მოდით წავიდეთ!

ძირითადი ნაწილი, პრაქტიკული (და ცოტა თეორიული)

რატომ ჩვენ (და თქვენ)

როცა ხეები დიდი იყო და მე უბრალო SQLერთ რუსულ საცალო ვაჭრობაში, ჩვენ მოვატყუეთ ETL პროცესები, ანუ მონაცემთა ნაკადები, ჩვენთვის ხელმისაწვდომი ორი ინსტრუმენტის გამოყენებით:

• Informatica Power Center - უკიდურესად გავრცელებული სისტემა, უკიდურესად პროდუქტიული, საკუთარი აპარატურით, საკუთარი ვერსიით. ღმერთმა ქნას მისი შესაძლებლობების 1% გამოვიყენე. რატომ? ისე, პირველ რიგში, ეს ინტერფეისი, სადღაც 380-იანი წლებიდან, გონებრივად ზეწოლას ახდენს ჩვენზე. მეორეც, ეს კონტრაქცია შექმნილია უკიდურესად ლამაზი პროცესებისთვის, კომპონენტების აღშფოთებული ხელახალი გამოყენებისთვის და სხვა ძალიან მნიშვნელოვანი საწარმოს ხრიკებისთვის. იმის შესახებ, რომ ღირს, ისევე როგორც Airbus AXNUMX ფრთა / წელიწადში, ჩვენ არაფერს ვიტყვით.

  ფრთხილად, სკრინშოტმა შეიძლება ოდნავ დააზარალოს 30 წლამდე ადამიანები

  

• SQL Server ინტეგრაციის სერვერი - ეს ამხანაგი გამოვიყენეთ ჩვენს შიდაპროექტულ ნაკადებში. სინამდვილეში: ჩვენ უკვე ვიყენებთ SQL Server-ს და რატომღაც არაგონივრული იქნება არ გამოვიყენოთ მისი ETL ინსტრუმენტები. მასში ყველაფერი კარგია: ინტერფეისიც ლამაზია და პროგრესის ანგარიშები... მაგრამ ეს არ არის მიზეზი, რის გამოც ჩვენ გვიყვარს პროგრამული პროდუქტები, ოჰ, არა ამისთვის. მისი ვერსია dtsx (რომელიც არის XML კვანძებით შერწყმული შენახვაზე) შეგვიძლია, მაგრამ რა აზრი აქვს? რას იტყვით დავალების პაკეტის შექმნაზე, რომელიც გადაიტანს ასობით ცხრილს ერთი სერვერიდან მეორეზე? დიახ, რა ასია, თქვენი საჩვენებელი თითი ოცი ცალიდან ჩამოვარდება, მაუსის ღილაკზე დაჭერით. მაგრამ ის ნამდვილად უფრო მოდურად გამოიყურება:

  

ჩვენ, რა თქმა უნდა, ვეძებდით გამოსავალს. საქმე კი почти მივიდა თვითნაწერი SSIS პაკეტის გენერატორთან ...

…და მერე ახალი სამსახური მიპოვა. და Apache Airflow-მა გამასწრო მასზე.

როდესაც გავარკვიე, რომ ETL პროცესის აღწერილობები მარტივი პითონის კოდია, უბრალოდ არ ვცეკვავდი სიხარულისგან. ასე ხდებოდა მონაცემთა ნაკადების ვერსიები და განსხვავებები და ასობით მონაცემთა ბაზიდან ერთიანი სტრუქტურის მქონე ცხრილების ერთ სამიზნეში გადატანა პითონის კოდის საკითხი გახდა ერთნახევარ ან ორ 13 დიუმიან ეკრანზე.

კლასტერის აწყობა

ნუ მოვაწყობთ მთლიანად საბავშვო ბაღს და არ ვისაუბროთ აქ სრულიად ცხად რაღაცეებზე, როგორიცაა Airflow-ის, თქვენ მიერ არჩეული მონაცემთა ბაზის, Celery-ის და დოკებში აღწერილი სხვა შემთხვევების დაყენება.

რათა დაუყოვნებლივ დავიწყოთ ექსპერიმენტები, დავხატე ესკიზი docker-compose.yml რომელშიც:

• რეალურად ავწიოთ ჰაერის: განრიგი, ვებსერვერი. ყვავილი ასევე ტრიალებს იქ ნიახურის ამოცანების მონიტორინგისთვის (რადგან ის უკვე ჩასმულია apache/airflow:1.10.10-python3.7, მაგრამ ჩვენ წინააღმდეგი არ ვართ)
• PostgreSQL, რომელშიც Airflow ჩაწერს თავის სერვისის ინფორმაციას (განრიგის მონაცემებს, შესრულების სტატისტიკას და ა.შ.), ხოლო Celery მონიშნავს დასრულებულ დავალებებს;
• Redis, რომელიც შეასრულებს სამუშაო ბროკერს Celery-სთვის;
• ნიახურის მუშა, რომელიც დავალებების უშუალო შესრულებაში იქნება დაკავებული.
• საქაღალდეში ./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, არამედ გარემოს ცვლადების საშუალებით (დეველოპერების წყალობით), რომლითაც მე მავნედ ვისარგებლე.
• ბუნებრივია, ის არ არის მზად წარმოებისთვის: მე განზრახ არ დავაყენე გულისცემა კონტეინერებზე, არ შევაწუხე უსაფრთხოება. მაგრამ მე გავაკეთე ჩვენი ექსპერიმენტატორებისთვის შესაფერისი მინიმალური.
• Ჩაინიშნე:
  • dag საქაღალდე ხელმისაწვდომი უნდა იყოს როგორც დამგეგმავისთვის, ასევე მუშაკებისთვის.
  • იგივე ეხება მესამე მხარის ყველა ბიბლიოთეკას - ისინი ყველა უნდა იყოს დაინსტალირებული მანქანებზე, რომლებსაც აქვთ გრაფიკი და მუშები.

ისე, ახლა ეს მარტივია:

$ docker-compose up --scale worker=3

მას შემდეგ რაც ყველაფერი გაიზრდება, შეგიძლიათ გადახედოთ ვებ ინტერფეისებს:

• ჰაერი: http://127.0.0.1:8080/admin/
• ყვავილი: http://127.0.0.1:5555/dashboard

ძირითადი ცნებები

თუ თქვენ ვერაფერი გაიგეთ ყველა ამ "დაგში", მაშინ აქ არის მოკლე ლექსიკონი:

• Scheduler - ყველაზე მნიშვნელოვანი ბიძა Airflow-ში, რომელიც აკონტროლებს, რომ რობოტები ბევრს მუშაობენ და არა ადამიანი: აკონტროლებს განრიგს, განაახლებს დავალებებს, იწყებს დავალებებს.

  ზოგადად, ძველ ვერსიებში მას ჰქონდა პრობლემები მეხსიერებასთან (არა, არა ამნეზია, არამედ გაჟონვა) და მემკვიდრეობის პარამეტრი კონფიგურაციაშიც კი დარჩა run_duration - მისი გადატვირთვის ინტერვალი. მაგრამ ახლა ყველაფერი კარგადაა.

• DAG (aka "dag") - "მიმართული აციკლური გრაფიკი", მაგრამ ასეთი განმარტება რამდენიმე ადამიანს ეტყვის, მაგრამ სინამდვილეში ეს არის კონტეინერი ამოცანებისთვის, რომლებიც ურთიერთქმედებენ ერთმანეთთან (იხ. ქვემოთ) ან პაკეტის ანალოგი SSIS-ში და Workflow Informatica-ში. .

  დაღების გარდა, შეიძლება არსებობდეს ქვედაგები, მაგრამ, სავარაუდოდ, მათ ვერ მივაღწევთ.

• DAG Run - ინიციალიზებული 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)

მოდით გავარკვიოთ:

• პირველ რიგში, ჩვენ შემოვიტანთ საჭირო ლიბებს და კიდევ რაღაც;
• sql_server_ds - რამდენად List[namedtuple[str, str]] კავშირების სახელებით Airflow Connections-დან და მონაცემთა ბაზებით, საიდანაც ავიღებთ ჩვენს ფირფიტას;
• dag - ჩვენი დაგის განცხადება, რომელიც აუცილებლად უნდა იყოს globals()თორემ Airflow ვერ იპოვის. დუგმა ასევე უნდა თქვას:
  • რა ქვია orders - შემდეგ ეს სახელი გამოჩნდება ვებ ინტერფეისში,
  • რომ ის იმუშავებს რვა ივლისის შუაღამედან,
  • და ის უნდა იმუშაოს, დაახლოებით ყოველ 6 საათში (მკაცრი ბიჭებისთვის აქ ნაცვლად timedelta() დასაშვებია cron- ხაზი 0 0 0/6 ? * * *, ნაკლებად მაგარისთვის - გამოთქმა მოსწონს @daily);
• workflow() მთავარ საქმეს გააკეთებს, მაგრამ არა ახლა. ამ დროისთვის, ჩვენ უბრალოდ გადავიტანთ ჩვენს კონტექსტს ჟურნალში.
• ახლა კი დავალებების შექმნის მარტივი მაგია:
  • ჩვენ გავდივართ ჩვენს წყაროებში;
  • ინიციალიზაცია PythonOperator, რომელიც შეასრულებს ჩვენს დუმს workflow(). არ დაგავიწყდეთ დავალების უნიკალური (დაგის ფარგლებში) სახელის მითითება და თავად დაგის მიბმა. დროშა provide_context თავის მხრივ, დამატებით არგუმენტებს ჩაასხამს ფუნქციაში, რომელსაც ჩვენ ყურადღებით ვაგროვებთ **context.

ჯერ-ჯერობით სულ ესაა. რაც მივიღეთ:

• ახალი დარტყმა ვებ ინტერფეისში,
• ასი ნახევარი დავალება, რომელიც შესრულდება პარალელურად (თუ ამის საშუალებას იძლევა Airflow, Celery პარამეტრები და სერვერის მოცულობა).

ისე, თითქმის მივხვდი.

ვინ დააინსტალირებს დამოკიდებულებებს?

ამ ყველაფრის გასამარტივებლად, მე ჩავდექი 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 - არაფერი გვიშლის ხელს შეგვექმნა საკუთარი ოპერატორები, რითაც ვისარგებლეთ Unblocked-ზე შეტყობინებების გასაგზავნად პატარა შეფუთვით. (ამ ოპერატორზე დაწვრილებით ქვემოთ ვისაუბრებთ);
• default_args={} - dag-ს შეუძლია ერთი და იგივე არგუმენტების გავრცელება მის ყველა ოპერატორზე;
• to='{{ var.value.all_the_kings_men }}' - ველი to ჩვენ არ გვექნება მყარი კოდირებული, მაგრამ დინამიურად გენერირებული Jinja-ს გამოყენებით და ცვლადი ელ.ფოსტის სიით, რომელიც მე ფრთხილად ჩავდე Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS - ოპერატორის გაშვების პირობა. ჩვენს შემთხვევაში, წერილი უფროსებისკენ მიფრინავს მხოლოდ იმ შემთხვევაში, თუ ყველა დამოკიდებულება დამუშავდა წარმატებით;
• tg_bot_conn_id='tg_main' - არგუმენტები conn_id მივიღოთ კავშირის ID-ები, რომლებშიც ჩვენ ვქმნით Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED - Telegram-ში შეტყობინებები გაფრინდება მხოლოდ იმ შემთხვევაში, თუ არსებობს ამოცანები;
• task_concurrency=1 - ჩვენ ვკრძალავთ ერთი ამოცანის რამდენიმე ამოცანის ერთდროულ გაშვებას. წინააღმდეგ შემთხვევაში, ჩვენ მივიღებთ რამდენიმე ერთდროულად გაშვებას VerticaOperator (ერთ მაგიდას უყურებს);
• report_update >> [email, tg] - ყველა VerticaOperator თანხვედრა წერილებისა და შეტყობინებების გაგზავნისას, როგორიცაა:
  

  მაგრამ რადგან შემტყობინებელ ოპერატორებს აქვთ გაშვების განსხვავებული პირობები, მხოლოდ ერთი იმუშავებს. ხეების ხედში ყველაფერი ნაკლებად ვიზუალურად გამოიყურება:
  

რამდენიმე სიტყვას ვიტყვი ამის შესახებ მაკროები და მათი მეგობრები - ცვლადები.

მაკრო არის Jinja ჩანაცვლება, რომელსაც შეუძლია შეცვალოს სხვადასხვა სასარგებლო ინფორმაცია ოპერატორის არგუმენტებში. მაგალითად, ასე:

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. საუკეთესო ნაწილი არის ის, რომ კონტექსტური ცვლადები მიმაგრებულია კონკრეტული ამოცანის მაგალითზე (კვადრატი ხეების ხედში) და გადატვირთვისას, ჩანაცვლების ველები გაფართოვდება იმავე მნიშვნელობებზე.

მინიჭებული მნიშვნელობების ნახვა შესაძლებელია Rendered ღილაკის გამოყენებით თითოეულ დავალების ინსტანციაზე. ასეა დავალება წერილის გაგზავნით:

ასე რომ, დავალების შესრულებისას შეტყობინების გაგზავნა:

უახლესი ხელმისაწვდომი ვერსიისთვის ჩაშენებული მაკროების სრული სია ხელმისაწვდომია აქ: მაკრო მინიშნება

უფრო მეტიც, დანამატების დახმარებით ჩვენ შეგვიძლია გამოვაცხადოთ საკუთარი მაკროები, მაგრამ ეს სხვა ამბავია.

გარდა წინასწარ განსაზღვრული ნივთებისა, ჩვენ შეგვიძლია შევცვალოთ ჩვენი ცვლადების მნიშვნელობები (ეს უკვე გამოვიყენე ზემოთ მოცემულ კოდში). შევქმნათ შიგნით 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) - ფაქტია, რომ ტიპების სია დამაგრებულია Airflow-ის მოდელებში და არ შეიძლება გაფართოვდეს საწყის კოდებში მოხვედრის გარეშე (თუ უცებ არ მოვძებნე რამე, შემასწორეთ), მაგრამ არაფერი შეგვიშლის კრედიტების მიღებაში მხოლოდ სახელი.

თქვენ ასევე შეგიძლიათ გააკეთოთ რამდენიმე კავშირი იმავე სახელით: ამ შემთხვევაში, მეთოდი BaseHook.get_connection(), რომელიც გვაძლევს კავშირებს სახელით, მოგვცემს შემთხვევითი რამდენიმე სახელიდან (უფრო ლოგიკური იქნებოდა Round Robin-ის გაკეთება, მაგრამ მოდით ეს Airflow დეველოპერების სინდისზე დავტოვოთ).

ცვლადები და კავშირები, რა თქმა უნდა, მაგარი ინსტრუმენტებია, მაგრამ მნიშვნელოვანია, რომ არ დაკარგოთ ბალანსი: თქვენი ნაკადების რომელ ნაწილებს ინახავთ თავად კოდში და რომელ ნაწილებს აძლევთ Airflow-ს შესანახად. ერთის მხრივ, მოსახერხებელია მნიშვნელობის სწრაფად შეცვლა, მაგალითად, საფოსტო ყუთი, ინტერფეისის საშუალებით. მეორეს მხრივ, ეს კვლავ მაუსის დაწკაპუნებაზე დაბრუნებაა, რომლისგანაც ჩვენ (მე) გვინდოდა თავის დაღწევა.

კავშირებთან მუშაობა ერთ-ერთი ამოცანაა კაკვები. ზოგადად, საჰაერო ნაკადის კაკვები არის მესამე მხარის სერვისებთან და ბიბლიოთეკებთან დაკავშირების წერტილები. Მაგალითად, 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(), რომელსაც 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 ველი), რომელშიც მე (ჩემი ინსტრუქციის მიხედვით!) ვდებ Telegram-ის ბოტის ჟეტონს: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• მე ვქმნი მაგალითს ჩვენი TelegramBot, მისცეს მას კონკრეტული ნიშანი.

Სულ ეს არის. თქვენ შეგიძლიათ მიიღოთ კლიენტი კაუჭიდან გამოყენებით TelegramBotHook().clent ან TelegramBotHook().get_conn().

და ფაილის მეორე ნაწილი, რომელშიც ვაკეთებ მიკროშეფუთვას Telegram 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-იდან ვერტიკაში მონაცემების გადატანას დააპირა და მერე აიღო და თემიდან გადავიდა ნაძირალა!

ეს სისასტიკე იყო მიზანმიმართული, მე უბრალოდ მომიწია თქვენთვის რაიმე ტერმინოლოგიის გაშიფვრა. ახლა შეგიძლია უფრო შორს წახვიდე.

ჩვენი გეგმა ასეთი იყო:

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:

მთავარია ეს არ აჩვენო ანალიტიკოსებს

დამუშავება 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. ჰუკის დახმარებით ვიღებთ Airflow-დან pymssql- დაკავშირება
2. მოდით შევცვალოთ მოთხოვნაში შეზღუდვა თარიღის სახით - ის ჩააგდებს ფუნქციას შაბლონის ძრავის მიერ.
3. კვებავს ჩვენს მოთხოვნას pandasვინ მოგვიღებს DataFrame - მომავალში გამოგვადგება.

მე ვიყენებ ჩანაცვლებას {dt} მოთხოვნის პარამეტრის ნაცვლად %s არა იმიტომ, რომ ბოროტი პინოქიო ვარ, არამედ იმიტომ pandas ვერ უმკლავდება pymssql და სრიალებს უკანასკნელს params: Listთუმცა მას ნამდვილად სურს tuple.
ასევე გაითვალისწინეთ, რომ დეველოპერი pymssql გადაწყვიტა აღარ დაეჭირა მისთვის მხარი და დროა გადახვიდე pyodbc.

ვნახოთ, რითი ავსებდა Airflow-მა ჩვენი ფუნქციების არგუმენტები:

თუ მონაცემები არ არის, მაშინ გაგრძელებას აზრი არ აქვს. მაგრამ ასევე უცნაურია შევსების წარმატებულად მიჩნევა. მაგრამ ეს არ არის შეცდომა. ა-ა-აჰ რა ვქნა?! და აი რა:

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 - ისე, რომ საბოლოო მონაცემთა ბაზაში (სადაც ყველაფერი ერთ ცხრილშია ჩასმული) გვქონდეს უნიკალური შეკვეთის 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)

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

შეჯამება

- კარგი, - თქვა პატარა თაგვმა, - არა, ახლა
დარწმუნებული ხარ, რომ მე ვარ ყველაზე საშინელი ცხოველი ტყეში?

ჯულია დონალდსონი, გრიფალო

ვფიქრობ, მე და ჩემს კოლეგებს რომ გვქონდეს კონკურსი: ვინ სწრაფად შექმნის და დაიწყებს ETL პროცესს ნულიდან: ისინი თავიანთი SSIS-ით და მაუსით და მე Airflow-ით... და შემდეგ ჩვენ ასევე შევადარებთ მოვლის სიმარტივეს... ვაიმე, მგონი დამეთანხმებით, რომ ყველა ფრონტზე დავამარცხებ!

თუ ცოტა უფრო სერიოზულად, მაშინ Apache Airflow - პროცესების აღწერით პროგრამის კოდის სახით - შეასრულა ჩემი საქმე ბევრი უფრო კომფორტული და სასიამოვნო.

მისი შეუზღუდავი გაფართოება, როგორც დანამატების, ისე მასშტაბურობისადმი მიდრეკილების თვალსაზრისით, გაძლევთ შესაძლებლობას გამოიყენოთ 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, რაც ყველაზე ხშირად მიუთითებს იმაზე, რომ დაგავიწყდათ დაგების ოპერატორთან დაკავშირება.

• ყველა ერთ მანქანაზე. დიახ, და ბაზები (თავად Airflow და ჩვენი საფარი), და ვებ სერვერი, და გრაფიკი და მუშები. და იმუშავა კიდეც. მაგრამ დროთა განმავლობაში, სერვისებისთვის დავალებების რაოდენობა გაიზარდა და როდესაც PostgreSQL-მა დაიწყო ინდექსზე რეაგირება 20 ms-ის ნაცვლად 5 წამში, ჩვენ ავიღეთ იგი და წავიყვანეთ.
• ლოკალური აღმასრულებელი. დიახ, ჩვენ ისევ მასზე ვსხედვართ და უკვე უფსკრულის პირას მივედით. LocalExecutor აქამდე საკმარისი იყო ჩვენთვის, მაგრამ ახლა დროა გავაფართოვოთ მინიმუმ ერთი მუშაკი და ჩვენ ბევრი შრომა მოგვიწევს CeleryExecutor-ში გადასასვლელად. და იმის გათვალისწინებით, რომ თქვენ შეგიძლიათ მასთან მუშაობა ერთ აპარატზე, არაფერი გიშლით ხელს, გამოიყენოთ Celery თუნდაც სერვერზე, რომელიც "რა თქმა უნდა, არასოდეს შევა წარმოებაში, გულწრფელად!"
• გამოუყენებლობა ჩაშენებული ხელსაწყოები:
  • კავშირი მომსახურების სერთიფიკატების შესანახად,
  • SLA მენატრება უპასუხოს დავალებებს, რომლებიც დროულად არ შესრულდა,
  • xcom მეტამონაცემების გაცვლისთვის (მე ვთქვი მეტამონაცემები!) dag ამოცანებს შორის.
• ფოსტის ბოროტად გამოყენება. აბა, რა ვთქვა? დაყენებული იყო სიგნალიზაცია დაცემულ დავალების ყველა გამეორებისთვის. ახლა ჩემს სამუშაო Gmail-ს აქვს >90 ათასი ელ.წერილი Airflow-დან და ვებ ფოსტის მჭიდი უარს ამბობს ერთდროულად 100-ზე მეტის აღებაზე და წაშლაზე.

მეტი ხარვეზები: Apache Airflow Pitfails

მეტი ავტომატიზაციის ხელსაწყოები

იმისთვის, რომ უფრო მეტად ვიმუშაოთ თავით და არა ხელებით, Airflow-მ მოგვამზადა ეს:

• REST API - ჯერ კიდევ აქვს ექსპერიმენტატორის სტატუსი, რაც არ უშლის ხელს მუშაობაში. მასთან ერთად, თქვენ შეგიძლიათ არა მხოლოდ მიიღოთ ინფორმაცია dags და ამოცანების შესახებ, არამედ შეაჩეროთ/დაიწყოთ dag, შექმნათ DAG Run ან pool.
• CLI - მრავალი ინსტრუმენტი ხელმისაწვდომია ბრძანების სტრიქონში, რომლებიც არა მხოლოდ მოუხერხებელია WebUI-ის საშუალებით გამოსაყენებლად, არამედ საერთოდ არ არსებობს. Მაგალითად:
  • backfill საჭიროა ამოცანების შემთხვევების გადატვირთვა.
    მაგალითად, მოვიდნენ ანალიტიკოსები და თქვეს: „შენ კი, ამხანაგო, 1-დან 13 იანვრამდე მონაცემებში სისულელე გაქვს! გაასწორე, გაასწორე, გაასწორე, გაასწორე!” და შენ ხარ ასეთი ჰობი:

    airflow backfill -s '2020-01-01' -e '2020-01-13' orders

  • ბაზის სერვისი: initdb, resetdb, upgradedb, checkdb.
  • run, რომელიც საშუალებას გაძლევთ შეასრულოთ ერთი ინსტანციის დავალება და გაიტანოთ ყველა დამოკიდებულების ქულები. უფრო მეტიც, შეგიძლიათ მისი გაშვება LocalExecutor, მაშინაც კი, თუ თქვენ გაქვთ ნიახურის მტევანი.
  • თითქმის იგივეს აკეთებს test, მხოლოდ ბაზებშიც არაფერს წერს.
  • connections ჭურვიდან კავშირების მასობრივი შექმნის საშუალებას იძლევა.
• პითონის API - ურთიერთქმედების საკმაოდ მძიმე გზა, რომელიც განკუთვნილია დანამატებისთვის და არა მასში პატარა ხელებით. მაგრამ ვინ გვიშლის ხელს /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 Documentation - რა თქმა უნდა, ოფისიდან უნდა დავიწყოთ. დოკუმენტაცია, მაგრამ ვინ კითხულობს ინსტრუქციებს?
• საუკეთესო პრაქტიკა - კარგი, მაინც წაიკითხეთ შემქმნელების რეკომენდაციები.
• Airflow UI - თავიდანვე: მომხმარებლის ინტერფეისი სურათებში
• Apache Airflow-ის ძირითადი კონცეფციების გაგება - ძირითადი ცნებები კარგად არის აღწერილი, თუ (მოულოდნელად!) ჩემგან რაღაც ვერ გაიგე.
• Tianlong-ის ბლოგი — გზამკვლევი, თუ როგორ უნდა ავაშენოთ საჰაერო ნაკადის სერვერი/კლასტერი - მოკლე სახელმძღვანელო ჰაერის ნაკადის კლასტერის დასაყენებლად.
• გაშვებული Apache Airflow Lyft-ზე - თითქმის იგივე საინტერესო სტატია, გარდა, ალბათ, მეტი ფორმალიზმი და ნაკლები მაგალითები.
• როგორ ანაწილებს Apache Airflow სამუშაოებს ნიახურის მუშაკებზე — სელერისთან ერთად მუშაობის შესახებ.
• DAG წერის საუკეთესო პრაქტიკა Apache Airflow-ში - ამოცანების უიმედოობის შესახებ, თარიღის ნაცვლად ID-ით ჩატვირთვა, ტრანსფორმაცია, ფაილის სტრუქტურა და სხვა საინტერესო რამ.
• Apache Airflow-ში დამოკიდებულების მართვა - ამოცანების დამოკიდებულებები და Trigger Rule, რომელიც მე მხოლოდ წარსულში აღვნიშნე.
• ჰაერის ნაკადი: როდესაც თქვენი DAG შორს არის განრიგს - როგორ გადავლახოთ გარკვეული „სამუშაოები, როგორც ეს იყო დაგეგმილი“ განრიგში, ჩატვირთოთ დაკარგული მონაცემები და დავადგინოთ პრიორიტეტები.
• სასარგებლო SQL მოთხოვნები Apache Airflow-ისთვის - სასარგებლო SQL მოთხოვნები Airflow მეტამონაცემებისთვის.
• დაიწყეთ სამუშაო ნაკადების შემუშავება Apache Airflow-ით - არის სასარგებლო განყოფილება პერსონალური სენსორის შექმნის შესახებ.
• Fetchr Data Science Infra-ს აგება AWS-ზე Presto-სა და Airflow-ით — საინტერესო მოკლე ჩანაწერი მონაცემთა მეცნიერებისთვის AWS-ზე ინფრასტრუქტურის აშენების შესახებ.
• 7 ჩვეულებრივი შეცდომა, რომელიც უნდა შეამოწმოთ ჰაერის ნაკადის DAG-ების გამართვისას - გავრცელებული შეცდომები (როდესაც ვინმე ჯერ კიდევ არ კითხულობს ინსტრუქციებს).
• შეინახეთ და შედით პაროლზე Apache Airflow-ის გამოყენებით - გაიღიმეთ, როგორ ართმევენ ხალხს პაროლების შენახვას, თუმცა შეგიძლიათ უბრალოდ გამოიყენოთ Connections.
• პითონის ზენი და აპაჩის ჰაერის ნაკადი - იმპლიციტური DAG გადამისამართება, კონტექსტის გადაგდება ფუნქციებში, ისევ დამოკიდებულებების შესახებ და ასევე ამოცანების გაშვების გამოტოვების შესახებ.
• ჰაერის ნაკადი: ნაკლებად ცნობილი რჩევები, ხრიკები და საუკეთესო პრაქტიკა - გამოყენების შესახებ default arguments и params შაბლონებში, ასევე ცვლადებში და კავშირებში.
• ჰაერის ნაკადის განრიგის პროფილირება - ამბავი იმის შესახებ, თუ როგორ ემზადება დამგეგმავი Airflow 2.0-ისთვის.
• Apache Airflow 3 ნიახური მუშაკით დოკერ-კომპოზიციაში - ოდნავ მოძველებული სტატია ჩვენი კლასტერის განლაგების შესახებ docker-compose.
• 4 შაბლონური ამოცანები ჰაერის ნაკადის კონტექსტის გამოყენებით - დინამიური ამოცანები შაბლონებისა და კონტექსტის გადამისამართების გამოყენებით.
• შეცდომის შეტყობინებები Airflow-ში - სტანდარტული და მორგებული შეტყობინებები ფოსტით და Slack-ით.
• საჰაერო ნაკადის სახელოსნო: კომპლექსური DAG-ები ყავარჯნების გარეშე - განშტოების ამოცანები, მაკროები და XCom.

და სტატიაში გამოყენებული ბმულები:

• მაკრო მინიშნება - ჩანაცვლების ველები ხელმისაწვდომია შაბლონებში გამოსაყენებლად.
• საერთო პრობლემები - ჰაერის ნაკადი — ხშირი შეცდომები დაფების შექმნისას.
• puckel/docker-airflow: Docker Apache Airflow - docker-compose ექსპერიმენტებისთვის, გამართვისთვის და სხვა.
• python-telegram-bot/python-telegram-bot: ჩვენ გაგიკეთეთ შეფუთვა, რომელზეც უარს ვერ იტყვით - პითონის შეფუთვა Telegram REST API-სთვის.

წყარო: www.habr.com