Apache Airflow៖ ធ្វើឱ្យ ETL កាន់តែងាយស្រួល

សួស្តី ខ្ញុំគឺ Dmitry Logvinenko - វិស្វករទិន្នន័យនៃនាយកដ្ឋានវិភាគនៃក្រុមក្រុមហ៊ុន Vezet ។

ខ្ញុំនឹងប្រាប់អ្នកអំពីឧបករណ៍ដ៏អស្ចារ្យសម្រាប់បង្កើតដំណើរការ ETL - Apache Airflow ។ ប៉ុន្តែ Airflow មានភាពចម្រុះ និងច្រើនយ៉ាងដែលអ្នកគួរតែពិនិត្យមើលវាឱ្យកាន់តែដិតដល់ ទោះបីជាអ្នកមិនពាក់ព័ន្ធនឹងលំហូរទិន្នន័យក៏ដោយ ប៉ុន្តែចាំបាច់ត្រូវចាប់ផ្តើមដំណើរការណាមួយជាទៀងទាត់ និងតាមដានការប្រតិបត្តិរបស់ពួកគេ។

ហើយបាទ, ខ្ញុំនឹងមិនត្រឹមតែប្រាប់, ប៉ុន្តែក៏បង្ហាញផងដែរ: កម្មវិធីនេះមានកូដជាច្រើន, រូបថតអេក្រង់និងអនុសាសន៍។

អ្វី​ដែល​អ្នក​ឃើញ​ជា​ធម្មតា​នៅ​ពេល​ដែល​អ្នក Google ពាក្យ Airflow / Wikimedia Commons

តារាងមាតិកា

• សេចក្តីណែនាំ
• ផ្នែកសំខាន់ ជាក់ស្តែង (និងទ្រឹស្តីបន្តិច)
  • ហេតុអ្វីបានជាយើង (និងអ្នក)
  • ការប្រមូលផ្តុំចង្កោម
  • គំនិតជាមូលដ្ឋាន
  • យើងបង្កើតភារកិច្ច
  • បន្តិចអំពីផ្កា
  • យើងផ្ទុកលើសទម្ងន់
  • ការតភ្ជាប់ ទំពក់ និងអថេរផ្សេងទៀត។
  • ការញែកប្រតិបត្តិករផ្ទាល់ខ្លួន
  • តើអ្នកនឹងចាក់ទេ?
  • សង្ខេប
• ផ្នែកចុងក្រោយ ឯកសារយោង និងព័ត៌មាន
  • តុងរួចដែលយើងប្រមូលបានសម្រាប់អ្នក
  • ឧបករណ៍ស្វ័យប្រវត្តិកម្មច្រើនទៀត
• សេចក្តីយោង

សេចក្តីណែនាំ

Apache Airflow គឺដូចជា Django ដែរ៖

• សរសេរជា python
• មានផ្ទាំងគ្រប់គ្រងដ៏អស្ចារ្យ
• អាចពង្រីកបានដោយគ្មានកំណត់

- ប្រសើរជាង ហើយវាត្រូវបានបង្កើតឡើងក្នុងគោលបំណងខុសគ្នាទាំងស្រុង ពោលគឺ (ដូចដែលវាត្រូវបានសរសេរមុនកាតា)៖

• កំពុងដំណើរការ និងត្រួតពិនិត្យកិច្ចការនៅលើម៉ាស៊ីនចំនួនគ្មានដែនកំណត់ (ដូចជា Celery / Kubernetes ជាច្រើន និងមនសិការរបស់អ្នកនឹងអនុញ្ញាតឱ្យអ្នក)
• ជាមួយនឹងការបង្កើតលំហូរការងារថាមវន្តពីភាពងាយស្រួលក្នុងការសរសេរ និងយល់កូដ Python
• និងសមត្ថភាពក្នុងការភ្ជាប់មូលដ្ឋានទិន្នន័យ និង APIs ជាមួយគ្នាដោយប្រើសមាសធាតុដែលត្រៀមរួចជាស្រេច និងកម្មវិធីជំនួយដែលផលិតនៅផ្ទះ (ដែលសាមញ្ញបំផុត)។

យើងប្រើ Apache Airflow ដូចនេះ៖

• យើងប្រមូលទិន្នន័យពីប្រភពផ្សេងៗ (ឧទាហរណ៍ SQL Server និង PostgreSQL ជាច្រើន APIs ជាច្រើនដែលមានកម្មវិធីវាស់វែង សូម្បីតែ 1C) នៅក្នុង DWH និង ODS (យើងមាន Vertica និង Clickhouse)។
• កម្រិតណា cronដែលចាប់ផ្តើមដំណើរការបង្រួបបង្រួមទិន្នន័យនៅលើ ODS និងត្រួតពិនិត្យការថែទាំរបស់ពួកគេផងដែរ។

រហូតមកដល់ពេលថ្មីៗនេះ តម្រូវការរបស់យើងត្រូវបានគ្របដណ្តប់ដោយម៉ាស៊ីនមេតូចមួយដែលមាន 32 cores និង 50 GB នៃ RAM ។ នៅក្នុង Airflow វាដំណើរការ៖

• более 200 ដាស (តាមពិតលំហូរការងារដែលយើងបានបំពេញកិច្ចការ)
• ជាមធ្យមនីមួយៗ 70 កិច្ចការ,
• ភាពល្អនេះចាប់ផ្តើម (ជាមធ្យមផងដែរ) ម្តងក្នុងមួយម៉ោង.

ហើយអំពីរបៀបដែលយើងពង្រីក ខ្ញុំនឹងសរសេរខាងក្រោម ប៉ុន្តែឥឡូវនេះ ចូរកំណត់បញ្ហា über ដែលយើងនឹងដោះស្រាយ៖

មានប្រភព SQL Servers ចំនួន 50 ដែលនីមួយៗមាន XNUMX databases - instances of one project រៀងគ្នាពួកគេមានរចនាសម្ព័ន្ធដូចគ្នា (ស្ទើរតែគ្រប់ទីកន្លែង mua-ha-ha) ដែលមានន័យថានីមួយៗមានតារាង Orders (ជាសំណាងល្អ តារាងដែលមាននោះ ឈ្មោះអាចត្រូវបានជំរុញទៅក្នុងអាជីវកម្មណាមួយ) ។ យើងយកទិន្នន័យដោយបន្ថែមវាលសេវាកម្ម (ម៉ាស៊ីនមេប្រភព មូលដ្ឋានទិន្នន័យប្រភព លេខសម្គាល់កិច្ចការ ETL) ហើយបោះវាដោយឥតលាក់លៀម និយាយថា Vertica ។

តោះទៅ!

ផ្នែកសំខាន់ ជាក់ស្តែង (និងទ្រឹស្តីបន្តិច)

ហេតុអ្វីបានជាយើង (និងអ្នក)

នៅពេលដែលដើមឈើធំហើយខ្ញុំសាមញ្ញ SQL-schik នៅក្នុងការលក់រាយរបស់រុស្ស៊ីមួយ យើងបានបោកប្រាស់ដំណើរការ ETL aka លំហូរទិន្នន័យដោយប្រើឧបករណ៍ពីរដែលមានសម្រាប់ពួកយើង៖

• មជ្ឈមណ្ឌលថាមពល Informatica - ប្រព័ន្ធរីករាលដាលខ្លាំង ផលិតភាពខ្លាំង ជាមួយនឹងផ្នែករឹងផ្ទាល់ខ្លួន កំណែផ្ទាល់ខ្លួនរបស់វា។ ខ្ញុំបានប្រើព្រះហាមឃាត់ 1% នៃសមត្ថភាពរបស់វា។ ហេតុអ្វី? ជាដំបូង ចំណុចប្រទាក់នេះ កន្លែងណាមួយពីទសវត្សរ៍ឆ្នាំ 380 បានដាក់សម្ពាធផ្លូវចិត្តលើយើង។ ទីពីរ ការបង្ខាំងនេះត្រូវបានរចនាឡើងសម្រាប់ដំណើរការដ៏ប្រណិតបំផុត ការប្រើប្រាស់ឡើងវិញនូវសមាសធាតុដ៏ខឹងសម្បារ និងល្បិចសហគ្រាសសំខាន់ៗដទៃទៀត។ អំពីអ្វីដែលវាមានតម្លៃដូចជាស្លាបរបស់ Airbus AXNUMX / ឆ្នាំយើងនឹងមិននិយាយអ្វីទាំងអស់។

  សូមប្រយ័ត្ន រូបថតអេក្រង់អាចប៉ះពាល់ដល់មនុស្សអាយុក្រោម 30 ឆ្នាំបន្តិច

  

• SQL Server Integration Server - យើងបានប្រើសមមិត្តនេះនៅក្នុងលំហូរខាងក្នុងគម្រោងរបស់យើង។ តាមពិត៖ យើងប្រើ SQL Server រួចហើយ ហើយវានឹងមិនសមហេតុផលទេក្នុងការមិនប្រើឧបករណ៍ ETL របស់វា។ អ្វីគ្រប់យ៉ាងនៅក្នុងវាគឺល្អ: ទាំងចំណុចប្រទាក់គឺស្រស់ស្អាត, និងរបាយការណ៍វឌ្ឍនភាព ... ប៉ុន្តែនេះមិនមែនជាមូលហេតុដែលយើងស្រឡាញ់ផលិតផលកម្មវិធី, អូមិនមែនសម្រាប់នេះ។ កំណែវា។ dtsx (ដែលជា XML ដែលមានថ្នាំងសាប់នៅលើរក្សាទុក) យើងអាចធ្វើបាន ប៉ុន្តែតើអ្វីទៅជាចំណុច? តើធ្វើដូចម្តេចអំពីការបង្កើតកញ្ចប់ភារកិច្ចដែលនឹងអូសតារាងរាប់រយពីម៉ាស៊ីនមេមួយទៅមួយទៀត? បាទ, អ្វីដែលមួយរយ, ម្រាមដៃសន្ទស្សន៍របស់អ្នកនឹងធ្លាក់ចុះពីម្ភៃបំណែក, ចុចលើប៊ូតុងកណ្ដុរ។ ប៉ុន្តែវាពិតជាមើលទៅទាន់សម័យជាងនេះ៖

  

យើងពិតជាបានស្វែងរកផ្លូវចេញ។ ករណីសូម្បីតែ ស្ទើរតែ បានមកដល់ម៉ាស៊ីនបង្កើតកញ្ចប់ SSIS ដែលសរសេរដោយខ្លួនឯង...

ហើយបន្ទាប់មកការងារថ្មីបានរកឃើញខ្ញុំ។ ហើយ Apache Airflow បានយកឈ្នះខ្ញុំ។

នៅពេលដែលខ្ញុំបានរកឃើញថាការពិពណ៌នាដំណើរការ ETL គឺជាកូដ Python សាមញ្ញ ខ្ញុំគ្រាន់តែមិនបានរាំដើម្បីភាពរីករាយ។ នេះជារបៀបដែលការស្ទ្រីមទិន្នន័យត្រូវបានកំណែ និងខុសគ្នា ហើយការបញ្ចូលតារាងដែលមានរចនាសម្ព័ន្ធតែមួយពីមូលដ្ឋានទិន្នន័យរាប់រយទៅក្នុងគោលដៅតែមួយបានក្លាយជាបញ្ហានៃកូដ Python នៅក្នុងអេក្រង់មួយកន្លះឬពីរ 13 "។

ការប្រមូលផ្តុំចង្កោម

ចូរយើងកុំរៀបចំសាលាមត្តេយ្យទាំងស្រុង ហើយកុំនិយាយអំពីអ្វីដែលជាក់ស្តែងទាំងស្រុងនៅទីនេះ ដូចជាការដំឡើង Airflow មូលដ្ឋានទិន្នន័យដែលអ្នកបានជ្រើសរើស Celery និងករណីផ្សេងទៀតដែលបានពិពណ៌នានៅក្នុងចត។

ដើម្បីឲ្យយើងអាចចាប់ផ្តើមការពិសោធន៍ភ្លាមៗ ខ្ញុំបានគូសវាស docker-compose.yml ដែលក្នុងនោះ៖

• ចូរយើងលើកយ៉ាងពិតប្រាកដ លំហូរ​ខ្យល់៖ កម្មវិធីកំណត់ពេល, Webserver ។ ផ្កាក៏នឹងវិលទៅទីនោះ ដើម្បីតាមដានកិច្ចការ Celery (ព្រោះវាត្រូវបានរុញចូលរួចហើយ apache/airflow:1.10.10-python3.7ប៉ុន្តែយើងមិនប្រកាន់ទេ)
• PostgreSQLដែលក្នុងនោះ Airflow នឹងសរសេរព័ត៌មានសេវាកម្មរបស់វា (ទិន្នន័យកម្មវិធីកំណត់ពេល ស្ថិតិប្រតិបត្តិ។ល។) ហើយ Celery នឹងសម្គាល់កិច្ចការដែលបានបញ្ចប់។
• Redisដែលនឹងដើរតួជាឈ្មួញកណ្តាលកិច្ចការសម្រាប់ Celery;
• កម្មករ 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.cfgប៉ុន្តែតាមរយៈអថេរបរិស្ថាន (អរគុណចំពោះអ្នកអភិវឌ្ឍន៍) ដែលខ្ញុំបានទាញយកប្រយោជន៍ដោយព្យាបាទ។
• តាមធម្មជាតិ វាមិនមែនជាផលិតកម្មដែលត្រៀមរួចជាស្រេចទេ៖ ខ្ញុំមិនមានចេតនាមិនដាក់បេះដូងនៅលើធុងទេ ខ្ញុំមិនរំខានសន្តិសុខទេ។ ប៉ុន្តែខ្ញុំបានធ្វើអប្បបរមាសមរម្យសម្រាប់អ្នកពិសោធន៍របស់យើង។
• ចំណាំ​ថា:
  • ថត Dag ត្រូវតែអាចចូលប្រើបានទាំងអ្នកកំណត់ពេល និងកម្មករ។
  • ដូចគ្នានេះដែរអនុវត្តចំពោះបណ្ណាល័យភាគីទីបីទាំងអស់ - ពួកគេទាំងអស់ត្រូវតែត្រូវបានដំឡើងនៅលើម៉ាស៊ីនដែលមានកម្មវិធីកំណត់ពេល និងកម្មករ។

ឥឡូវនេះវាសាមញ្ញហើយ៖

$ docker-compose up --scale worker=3

បន្ទាប់ពីអ្វីៗកើនឡើង អ្នកអាចមើលចំណុចប្រទាក់គេហទំព័រ៖

• លំហូរ​ខ្យល់: http://127.0.0.1:8080/admin/
• ផ្កា៖ http://127.0.0.1:5555/dashboard

គំនិតជាមូលដ្ឋាន

ប្រសិនបើអ្នកមិនយល់អ្វីទាំងអស់នៅក្នុង "dags" ទាំងនេះទេនោះ នេះគឺជាវចនានុក្រមខ្លីមួយ៖

• កម្មវិធីកំណត់ពេលវេលា - ពូសំខាន់បំផុតនៅក្នុង Airflow ដែលគ្រប់គ្រងថាមនុស្សយន្តធ្វើការយ៉ាងលំបាក ហើយមិនមែនជាមនុស្សទេ៖ តាមដានកាលវិភាគ ធ្វើបច្ចុប្បន្នភាព dags ចាប់ផ្តើមកិច្ចការ។

  ជាទូទៅនៅក្នុងកំណែចាស់ គាត់មានបញ្ហាជាមួយនឹងការចងចាំ (ទេ មិនមែនការភ្លេចភ្លាំងទេ ប៉ុន្តែការលេចធ្លាយ) ហើយប៉ារ៉ាម៉ែត្រកេរ្តិ៍ដំណែលនៅតែមាននៅក្នុងការកំណត់។ run_duration - ចន្លោះពេលចាប់ផ្តើមឡើងវិញរបស់វា។ ប៉ុន្តែឥឡូវនេះអ្វីៗគឺល្អ។

• DAG (aka "dag") - "ក្រាហ្វដែលដឹកនាំដោយ acyclic" ប៉ុន្តែនិយមន័យបែបនេះនឹងប្រាប់មនុស្សតិចតួច ប៉ុន្តែតាមពិតវាគឺជាកុងតឺន័រសម្រាប់កិច្ចការដែលទាក់ទងគ្នាទៅវិញទៅមក (សូមមើលខាងក្រោម) ឬ analogue នៃកញ្ចប់នៅក្នុង SSIS និងលំហូរការងារនៅក្នុង Informatica .

  បន្ថែមពីលើ dags អាចនៅតែមាន subdags ប៉ុន្តែយើងទំនងជានឹងមិនទទួលបានពួកវាទេ។

• DAG រត់ - ដើមដាចដែលត្រូវបានចាត់តាំងដោយខ្លួនវាផ្ទាល់ execution_date. Dagrans នៃ dag ដូចគ្នាអាចដំណើរការស្របគ្នា (ប្រសិនបើអ្នកបានធ្វើឱ្យកិច្ចការរបស់អ្នកមិនមានកម្លាំង ពិតណាស់)។
• ប្រតិបត្តិករ គឺជាបំណែកនៃកូដដែលទទួលខុសត្រូវក្នុងការអនុវត្តសកម្មភាពជាក់លាក់មួយ។ មានប្រតិបត្តិករបីប្រភេទ៖
  • សកម្មភាពដូចជាសំណព្វរបស់យើង។ PythonOperatorដែលអាចប្រតិបត្តិកូដ Python ណាមួយ (ត្រឹមត្រូវ) ។
  • ការផ្ទេរប្រាក់ដែលដឹកជញ្ជូនទិន្នន័យពីកន្លែងមួយទៅកន្លែងមួយ និយាយថា MsSqlToHiveTransfer;
  • អង្គញាណ ម៉្យាងវិញទៀត វានឹងអនុញ្ញាតឱ្យអ្នកមានប្រតិកម្ម ឬបន្ថយល្បឿននៃការប្រតិបត្តិបន្ថែមទៀតរហូតដល់ព្រឹត្តិការណ៍មួយកើតឡើង។ HttpSensor អាចទាញចំណុចបញ្ចប់ដែលបានបញ្ជាក់ ហើយនៅពេលដែលការឆ្លើយតបដែលចង់បានកំពុងរង់ចាំ ចាប់ផ្តើមការផ្ទេរ GoogleCloudStorageToS3Operator. ចិត្តដែលចង់ដឹងចង់ឃើញនឹងសួរថា “ហេតុអ្វី? យ៉ាងណាមិញ អ្នកអាចធ្វើពាក្យដដែលៗនៅក្នុងប្រតិបត្តិករ!” ហើយបន្ទាប់មកដើម្បីកុំឱ្យស្ទះអាងនៃភារកិច្ចជាមួយប្រតិបត្តិករដែលផ្អាក។ ឧបករណ៍ចាប់សញ្ញាចាប់ផ្តើម ពិនិត្យ និងងាប់ មុនពេលការប៉ុនប៉ងលើកក្រោយ។
• កិច្ចការ។ - ប្រតិបត្តិករដែលបានប្រកាស ដោយមិនគិតពីប្រភេទ និងភ្ជាប់ជាមួយដាក ត្រូវបានតម្លើងឋានៈភារកិច្ច។
• ឧទាហរណ៍ភារកិច្ច - នៅពេលដែលអ្នករៀបចំផែនការទូទៅបានសម្រេចចិត្តថាវាដល់ពេលដែលត្រូវបញ្ជូនភារកិច្ចទៅក្នុងសមរភូមិលើអ្នកសំដែង (នៅនឹងកន្លែង ប្រសិនបើយើងប្រើ LocalExecutor ឬទៅថ្នាំងពីចម្ងាយក្នុងករណី CeleryExecutor) វាផ្តល់បរិបទដល់ពួកគេ (ឧ. សំណុំនៃអថេរ - ប៉ារ៉ាម៉ែត្រប្រតិបត្តិ) ពង្រីកពាក្យបញ្ជា ឬគំរូសំណួរ ហើយបញ្ចូលពួកវា។

យើងបង្កើតភារកិច្ច

ជាដំបូង ចូរយើងគូសបញ្ជាក់អំពីគ្រោងការណ៍ទូទៅនៃ doug របស់យើង ហើយបន្ទាប់មកយើងនឹងចូលទៅក្នុងព័ត៌មានលម្អិតកាន់តែច្រើនឡើងៗ ពីព្រោះយើងអនុវត្តដំណោះស្រាយដែលមិនសំខាន់មួយចំនួន។

ដូច្នេះ ក្នុងទម្រង់ដ៏សាមញ្ញបំផុតរបស់វា ដាវបែបនេះនឹងមើលទៅដូចនេះ៖

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]] ជាមួយនឹងឈ្មោះនៃការតភ្ជាប់ពី Airflow Connections និងមូលដ្ឋានទិន្នន័យដែលយើងនឹងយកចានរបស់យើង;
• dag - សេចក្តីប្រកាសអំពីដាករបស់យើង ដែលត្រូវតែចាំបាច់នៅក្នុង globals()បើមិនដូច្នេះទេ Airflow នឹងមិនស្វែងរកវាទេ។ Doug ក៏ត្រូវនិយាយថា៖
  • តើ​គាត់​ឈ្មោះ​អី orders - បន្ទាប់មកឈ្មោះនេះនឹងបង្ហាញនៅក្នុងចំណុចប្រទាក់បណ្ដាញ
  • ថាគាត់នឹងធ្វើការចាប់ពីពាក់កណ្តាលអធ្រាត្រថ្ងៃទីប្រាំបីខែកក្កដា។
  • ហើយវាគួរដំណើរការប្រហែលរៀងរាល់ 6 ម៉ោងម្តង (សម្រាប់បុរសរឹងរូសនៅទីនេះជំនួសឱ្យ timedelta() អាចទទួលយកបាន cron- បន្ទាត់ 0 0 0/6 ? * * *, សម្រាប់ត្រជាក់តិច - កន្សោមដូចជា @daily);
• workflow() នឹងធ្វើការងារសំខាន់ ប៉ុន្តែមិនមែនឥឡូវនេះទេ។ សម្រាប់ពេលនេះ យើងនឹងទម្លាក់បរិបទរបស់យើងទៅក្នុងកំណត់ហេតុ។
• ហើយឥឡូវនេះវេទមន្តសាមញ្ញនៃការបង្កើតភារកិច្ច:
  • យើងដំណើរការតាមរយៈប្រភពរបស់យើង;
  • ចាប់ផ្តើម PythonOperatorដែលនឹងប្រតិបត្តិអត់ចេះសោះរបស់យើង។ workflow(). កុំភ្លេចបញ្ជាក់ឈ្មោះតែមួយគត់ (នៅក្នុងដាក) នៃកិច្ចការ ហើយចងដាកដោយខ្លួនឯង។ ទង់ជាតិ provide_context នៅក្នុងវេននឹងចាក់អាគុយម៉ង់បន្ថែមទៅក្នុងមុខងារដែលយើងនឹងប្រមូលដោយប្រុងប្រយ័ត្នដោយប្រើ **context.

សម្រាប់ពេលនេះ នោះហើយជាអ្វីទាំងអស់។ អ្វីដែលយើងទទួលបាន៖

• dag ថ្មីនៅក្នុងចំណុចប្រទាក់បណ្ដាញ,
• កិច្ចការមួយនិងកន្លះរយដែលនឹងត្រូវប្រតិបត្តិស្របគ្នា (ប្រសិនបើ Airflow, ការកំណត់ Celery និងសមត្ថភាពម៉ាស៊ីនមេអនុញ្ញាត)។

ជិតបានហើយ។

តើអ្នកណានឹងដំឡើងភាពអាស្រ័យ?

ដើម្បី​សម្រួល​រឿង​ទាំង​អស់​នេះ ខ្ញុំ​បាន​ចូល docker-compose.yml ដំណើរការ requirements.txt នៅលើថ្នាំងទាំងអស់។

ឥឡូវ​បាត់​ហើយ៖

ការ៉េពណ៌ប្រផេះគឺជាកិច្ចការដែលដំណើរការដោយកម្មវិធីកំណត់ពេល។

យើង​រង់ចាំ​បន្តិច ការងារ​ត្រូវ​បាន​ចាប់​យក​ដោយ​កម្មករ៖

បៃតង ពិតណាស់បានបញ្ចប់ការងាររបស់ពួកគេដោយជោគជ័យ។ ហង្សក្រហមមិនសូវជោគជ័យទេ។

ដោយវិធីនេះមិនមានថតនៅលើផលិតផលរបស់យើងទេ។ ./dagsមិនមានការធ្វើសមកាលកម្មរវាងម៉ាស៊ីនទេ - 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 ទទួលយកលេខសម្គាល់ការតភ្ជាប់ដែលយើងបង្កើត Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED - សារក្នុង Telegram នឹងហោះហើរទៅឆ្ងាយ លុះត្រាតែមានកិច្ចការធ្លាក់។
• task_concurrency=1 - យើងហាមឃាត់ការបើកដំណើរការក្នុងពេលដំណាលគ្នានៃកិច្ចការជាច្រើននៃកិច្ចការមួយ។ បើមិនដូច្នេះទេ យើងនឹងទទួលបានការបើកដំណើរការក្នុងពេលដំណាលគ្នាជាច្រើន។ VerticaOperator (មើលតារាងមួយ);
• report_update >> [email, tg] - ទាំងអស់ VerticaOperator រួម​បញ្ចូល​គ្នា​ក្នុង​ការ​ផ្ញើ​សំបុត្រ និង​សារ​ដូច​នេះ៖
  

  ប៉ុន្តែចាប់តាំងពីប្រតិបត្តិករជូនដំណឹងមានលក្ខខណ្ឌនៃការបើកដំណើរការផ្សេងគ្នា មានតែមួយប៉ុណ្ណោះដែលនឹងដំណើរការ។ នៅក្នុងទិដ្ឋភាពមែកធាង អ្វីៗមើលទៅហាក់ដូចជាមិនសូវច្បាស់បន្តិច៖
  

ខ្ញុំនឹងនិយាយពាក្យពីរបីអំពី ម៉ាក្រូ និងមិត្តភក្តិរបស់ពួកគេ - អថេរ.

Macros គឺជាកន្លែងដាក់ 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. ផ្នែកដ៏ល្អបំផុតគឺថាអថេរបរិបទត្រូវបានភ្ជាប់ទៅនឹងឧទាហរណ៍កិច្ចការជាក់លាក់មួយ (ការ៉េក្នុងទិដ្ឋភាពមែកធាង) ហើយនៅពេលចាប់ផ្តើមឡើងវិញ កន្លែងដាក់នឹងពង្រីកទៅតម្លៃដូចគ្នា។

តម្លៃដែលបានកំណត់អាចត្រូវបានមើលដោយប្រើប៊ូតុងបង្ហាញនៅលើឧទាហរណ៍ភារកិច្ចនីមួយៗ។ នេះជារបៀបដែលភារកិច្ចផ្ញើសំបុត្រ៖

ហើយដូច្នេះនៅឯភារកិច្ចជាមួយនឹងការផ្ញើសារមួយ:

បញ្ជីពេញលេញនៃម៉ាក្រូដែលភ្ជាប់មកជាមួយសម្រាប់កំណែចុងក្រោយបំផុតអាចរកបាននៅទីនេះ៖ សេចក្តីយោងម៉ាក្រូ

លើសពីនេះទៅទៀត ដោយមានជំនួយពីកម្មវិធីជំនួយ យើងអាចប្រកាសម៉ាក្រូផ្ទាល់ខ្លួនរបស់យើង ប៉ុន្តែនោះជារឿងមួយទៀត។

បន្ថែមពីលើវត្ថុដែលបានកំណត់ជាមុន យើងអាចជំនួសតម្លៃនៃអថេររបស់យើង (ខ្ញុំបានប្រើវារួចហើយនៅក្នុងកូដខាងលើ)។ ចូរយើងបង្កើតនៅក្នុង 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 ហើយមិនអាចពង្រីកដោយមិនចាំបាច់ចូលទៅក្នុងកូដប្រភពទេ (ប្រសិនបើភ្លាមៗនោះខ្ញុំមិនបាន Google អ្វីមួយ សូមកែតម្រូវខ្ញុំ) ប៉ុន្តែគ្មានអ្វីនឹងរារាំងយើងពីការទទួលបានក្រេឌីតដោយគ្រាន់តែ ឈ្មោះ។

អ្នកក៏អាចធ្វើការតភ្ជាប់ជាច្រើនដែលមានឈ្មោះដូចគ្នា: ក្នុងករណីនេះវិធីសាស្ត្រ BaseHook.get_connection()ដែលធ្វើឱ្យយើងភ្ជាប់ទំនាក់ទំនងតាមឈ្មោះនឹងផ្តល់ឱ្យ ចៃដន្យ ពីឈ្មោះជាច្រើន (វានឹងសមហេតុផលជាងក្នុងការបង្កើត Round Robin ប៉ុន្តែសូមទុកវានៅលើមនសិការរបស់អ្នកអភិវឌ្ឍន៍ Airflow)។

Variables និង Connections គឺជាឧបករណ៍ដ៏ត្រជាក់ ប៉ុន្តែវាសំខាន់ណាស់ដែលមិនត្រូវបាត់បង់សមតុល្យ៖ តើផ្នែកណាមួយនៃលំហូររបស់អ្នកដែលអ្នករក្សាទុកនៅក្នុងកូដខ្លួនឯង និងផ្នែកណាមួយដែលអ្នកផ្តល់ឱ្យ Airflow សម្រាប់ផ្ទុក។ ម្យ៉ាងវិញទៀត វាអាចមានភាពងាយស្រួលក្នុងការផ្លាស់ប្តូរតម្លៃយ៉ាងឆាប់រហ័ស ឧទាហរណ៍ ប្រអប់សំបុត្រតាមរយៈ UI ។ ម៉្យាងវិញទៀត នេះនៅតែជាការត្រលប់ទៅការចុចកណ្ដុរ ដែលយើង (ខ្ញុំ) ចង់កម្ចាត់។

ការធ្វើការជាមួយការតភ្ជាប់គឺជាកិច្ចការមួយ។ ទំពក់. ជាទូទៅ ទំពក់ Airflow គឺជាចំណុចសម្រាប់ភ្ជាប់វាទៅសេវាកម្មភាគីទីបី និងបណ្ណាល័យ។ ឧ. JiraHook នឹងបើកម៉ាស៊ីនភ្ញៀវសម្រាប់ពួកយើងដើម្បីធ្វើអន្តរកម្មជាមួយ Jira (អ្នកអាចផ្លាស់ទីភារកិច្ចទៅមក) និងដោយមានជំនួយពី 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: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• ខ្ញុំបង្កើតឧទាហរណ៍របស់យើង។ TelegramBotផ្តល់ឱ្យវានូវសញ្ញាសម្ងាត់ជាក់លាក់មួយ។

អស់ហើយ។ អ្នកអាចទទួលបានអតិថិជនពីទំពក់ដោយប្រើ TelegramBotHook().clent ឬ TelegramBotHook().get_conn().

ហើយផ្នែកទីពីរនៃឯកសារដែលខ្ញុំបង្កើត microwrapper សម្រាប់ 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 ទៅ Vertica ហើយបន្ទាប់មកគាត់បានយកវាចេញ ហើយផ្លាស់ប្តូរប្រធានបទនេះ គឺអ្នកមើលងាយ!

អំពើ​ឃោរឃៅ​នេះ​គឺ​មាន​ចេតនា ខ្ញុំ​គ្រាន់​តែ​បក​ស្រាយ​ពាក្យ​មួយ​ចំនួន​សម្រាប់​អ្នក។ ឥឡូវនេះអ្នកអាចទៅបន្ថែមទៀត។

ផែនការរបស់យើងគឺនេះ៖

1. ធ្វើដាច
2. បង្កើតកិច្ចការ
3. មើល​ថា​ស្អាត​ប៉ុណ្ណា​ទៅ​
4. កំណត់លេខវគ្គដើម្បីបំពេញ
5. ទទួលបានទិន្នន័យពី SQL Server
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 Server,
• យើងបំពេញមូលដ្ឋានទិន្នន័យនៅពេលក្រោយជាមួយនឹងទិន្នន័យមួយចំនួន (ក្នុងករណីណាក៏ដោយកុំពិនិត្យមើល 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 មិនមែនដោយសារតែខ្ញុំជា Pinocchio អាក្រក់នោះទេ ប៉ុន្តែដោយសារតែ 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']])

ឈ្មោះ៖

• មូលដ្ឋានទិន្នន័យដែលយើងបានបញ្ជាទិញ
• លេខសម្គាល់នៃវគ្គទឹកជំនន់របស់យើង (វានឹងខុសគ្នា សម្រាប់រាល់កិច្ចការ),
• លេខសម្គាល់ពីប្រភព និងលំដាប់លេខសម្គាល់ - ដូច្នេះនៅក្នុងមូលដ្ឋានទិន្នន័យចុងក្រោយ (ដែលអ្វីៗត្រូវបានចាក់ចូលទៅក្នុងតារាងតែមួយ) យើងមានលេខសម្គាល់លំដាប់តែមួយគត់។

ជំហានចុងក្រោយនៅតែមាន: ចាក់អ្វីគ្រប់យ៉ាងទៅក្នុង 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

សង្ខេប

- មែនហើយ - កូនកណ្តុរតូចបាននិយាយថា - មែនទេឥឡូវនេះ
តើអ្នកជឿថាខ្ញុំជាសត្វដែលគួរឱ្យខ្លាចបំផុតនៅក្នុងព្រៃ?

Julia Donaldson, The Gruffalo

ខ្ញុំគិតថាប្រសិនបើសហសេវិករបស់ខ្ញុំ និងខ្ញុំមានការប្រកួតប្រជែងមួយ៖ តើអ្នកណានឹងបង្កើត និងចាប់ផ្តើមដំណើរការ ETL យ៉ាងឆាប់រហ័សពីដំបូង៖ ពួកគេជាមួយនឹង SSIS និងកណ្តុររបស់ពួកគេ និងខ្ញុំជាមួយនឹង Airflow ... ហើយបន្ទាប់មកយើងក៏នឹងប្រៀបធៀបភាពងាយស្រួលនៃការថែទាំផងដែរ ... អីយ៉ា ខ្ញុំ​គិត​ថា​អ្នក​នឹង​យល់ព្រម​ថា​ខ្ញុំ​នឹង​វាយ​ពួក​គេ​នៅ​លើ​មុខ​ទាំង​អស់!

ប្រសិនបើធ្ងន់ធ្ងរជាងនេះបន្តិច នោះ Apache Airflow - ដោយពណ៌នាអំពីដំណើរការក្នុងទម្រង់ជាកូដកម្មវិធី - បានធ្វើការងាររបស់ខ្ញុំ ច្រើន កាន់តែមានផាសុកភាព និងរីករាយ។

ការពង្រីកគ្មានដែនកំណត់របស់វា ទាំងនៅក្នុងលក្ខខណ្ឌនៃកម្មវិធីជំនួយ និងទំនោរទៅរកការធ្វើមាត្រដ្ឋាន ផ្តល់ឱ្យអ្នកនូវឱកាសដើម្បីប្រើប្រាស់ Airflow នៅស្ទើរតែគ្រប់ផ្នែក៖ សូម្បីតែនៅក្នុងវដ្តពេញលេញនៃការប្រមូល រៀបចំ និងដំណើរការទិន្នន័យ សូម្បីតែនៅក្នុងការបាញ់បង្ហោះរ៉ុក្កែត (ទៅកាន់ភពព្រះអង្គារ នៃ វគ្គសិក្សា) ។

ផ្នែកចុងក្រោយ ឯកសារយោង និងព័ត៌មាន

តុងរួចដែលយើងប្រមូលបានសម្រាប់អ្នក

• start_date. បាទ នេះ​ជា meme ក្នុងស្រុក​រួច​ហើយ។ តាមរយៈអាគុយម៉ង់ចម្បងរបស់ Doug start_date ឆ្លងកាត់ទាំងអស់។ សង្ខេបប្រសិនបើអ្នកបញ្ជាក់នៅក្នុង start_date កាលបរិច្ឆេទបច្ចុប្បន្ន និង schedule_interval - ថ្ងៃមួយបន្ទាប់មក DAG នឹងចាប់ផ្តើមនៅថ្ងៃស្អែកមិនមុនទេ។

  start_date = datetime(2020, 7, 7, 0, 1, 2)

  ហើយលែងមានបញ្ហាទៀតហើយ។

  មាន​កំហុស​ពេល​រត់​មួយ​ទៀត​ដែល​ទាក់ទង​នឹង​វា៖ Task is missing the start_date parameterដែលភាគច្រើនបង្ហាញថាអ្នកភ្លេចចងភ្ជាប់ជាមួយប្រតិបត្តិករ dag ។

• ទាំងអស់នៅលើម៉ាស៊ីនតែមួយ។ បាទ/ចាស និងមូលដ្ឋាន (Airflow ខ្លួនវាផ្ទាល់ និងថ្នាំកូតរបស់យើង) និងម៉ាស៊ីនមេគេហទំព័រ និងអ្នកកំណត់ពេល និងកម្មករ។ ហើយវាថែមទាំងដំណើរការទៀតផង។ ប៉ុន្តែយូរ ៗ ទៅចំនួននៃភារកិច្ចសម្រាប់សេវាកម្មបានកើនឡើងហើយនៅពេលដែល PostgreSQL ចាប់ផ្តើមឆ្លើយតបទៅនឹងសន្ទស្សន៍ក្នុងរយៈពេល 20 វិនាទីជំនួសឱ្យ 5 ms យើងបានយកវាទៅឆ្ងាយ។
• ប្រតិបត្តិករក្នុងស្រុក។ បាទ យើង​នៅ​តែ​អង្គុយ​លើ​វា ហើយ​យើង​បាន​មក​ដល់​មាត់​ជ្រៅ​ហើយ។ LocalExecutor គឺគ្រប់គ្រាន់សម្រាប់យើងរហូតមកដល់ពេលនេះ ប៉ុន្តែឥឡូវនេះវាដល់ពេលដែលត្រូវពង្រីកជាមួយបុគ្គលិកយ៉ាងហោចណាស់ម្នាក់ ហើយយើងនឹងត្រូវខិតខំប្រឹងប្រែងដើម្បីផ្លាស់ទីទៅ CeleryExecutor ។ ហើយនៅក្នុងទិដ្ឋភាពនៃការពិតដែលថាអ្នកអាចធ្វើការជាមួយវានៅលើម៉ាស៊ីនតែមួយគ្មានអ្វីរារាំងអ្នកពីការប្រើប្រាស់ Celery សូម្បីតែនៅលើម៉ាស៊ីនមេដែល "ជាការពិតណាស់នឹងមិនចូលទៅក្នុងការផលិតដោយស្មោះត្រង់!"
• មិនប្រើ ឧបករណ៍ដែលភ្ជាប់មកជាមួយ:
  • ការតភ្ជាប់ ដើម្បីរក្សាទុកព័ត៌មានបញ្ជាក់សេវាកម្ម,
  • SLA នឹក ដើម្បីឆ្លើយតបទៅនឹងកិច្ចការដែលមិនបានសម្រេចទាន់ពេលវេលា។
  • xcom សម្រាប់ការផ្លាស់ប្តូរទិន្នន័យមេតា (ខ្ញុំបាននិយាយ មេតាទិន្នន័យ!) រវាងការងារ dag ។
• ការបំពានសំបុត្រ។ អញ្ចឹងតើខ្ញុំអាចនិយាយអ្វីបាន? ការជូន​ដំណឹង​ត្រូវ​បាន​បង្កើត​ឡើង​សម្រាប់​ការ​ធ្វើ​ដដែលៗ​នៃ​កិច្ចការ​ដែល​បាន​ធ្លាក់​ចុះ។ ឥឡូវនេះ Gmail ការងាររបស់ខ្ញុំមានអ៊ីមែល > 90k ពី Airflow ហើយ web mail muzzle បដិសេធមិនទទួលយក និងលុបច្រើនជាង 100 ក្នុងពេលតែមួយ។

គ្រោះថ្នាក់ច្រើនទៀត៖ Apache Airflow Pitfails

ឧបករណ៍ស្វ័យប្រវត្តិកម្មច្រើនទៀត

ដើម្បីឱ្យយើងធ្វើការកាន់តែច្រើនដោយប្រើក្បាលរបស់យើង មិនមែនដោយដៃរបស់យើង Airflow បានរៀបចំសម្រាប់យើងនូវចំណុចនេះ៖

• REST API - គាត់នៅតែមានស្ថានភាពនៃបទពិសោធន៍ដែលមិនរារាំងគាត់ពីការងារ។ ជាមួយវា អ្នកមិនត្រឹមតែអាចទទួលបានព័ត៌មានអំពី dags និងកិច្ចការប៉ុណ្ណោះទេ ប៉ុន្តែថែមទាំងបញ្ឈប់/ចាប់ផ្តើម dag បង្កើត DAG Run ឬអាង។
• CLI - ឧបករណ៍ជាច្រើនអាចរកបានតាមរយៈបន្ទាត់ពាក្យបញ្ជាដែលមិនត្រឹមតែមិនងាយស្រួលប្រើតាមរយៈ WebUI ប៉ុណ្ណោះទេ ប៉ុន្តែជាទូទៅគឺអវត្តមាន។ ឧទាហរណ៍:
  • backfill ត្រូវការដើម្បីចាប់ផ្តើមឧទាហរណ៍កិច្ចការឡើងវិញ។
    ជាឧទាហរណ៍ អ្នកវិភាគបានមក ហើយនិយាយថា៖ «ហើយសមមិត្ត មានការសមហេតុសមផលក្នុងទិន្នន័យចាប់ពីថ្ងៃទី 1 ដល់ថ្ងៃទី 13 ខែមករា! ជួសជុល កែ ជួសជុល ជួសជុល! ហើយអ្នកគឺជាចំណង់ចំណូលចិត្តបែបនេះ:

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

  • សេវាមូលដ្ឋាន៖ initdb, resetdb, upgradedb, checkdb.
  • runដែលអនុញ្ញាតឱ្យអ្នកដំណើរការកិច្ចការឧទាហរណ៍មួយ ហើយថែមទាំងដាក់ពិន្ទុលើភាពអាស្រ័យទាំងអស់។ លើសពីនេះទៀតអ្នកអាចដំណើរការវាតាមរយៈ LocalExecutorទោះបីជាអ្នកមានចង្កោម Celery ក៏ដោយ។
  • ធ្វើរឿងដូចគ្នា។ testមានតែនៅក្នុងមូលដ្ឋានមិនសរសេរអ្វីទាំងអស់។
  • connections អនុញ្ញាតឱ្យបង្កើតការតភ្ជាប់ដ៏ធំពីសែល។
• python 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)

• កំពុងភ្ជាប់ទៅមូលដ្ឋានទិន្នន័យមេតាលំហូរខ្យល់។ ខ្ញុំ​មិន​ណែនាំ​ឲ្យ​សរសេរ​ទៅ​វា​ទេ ប៉ុន្តែ​ការ​ទទួល​បាន​ស្ថានភាព​កិច្ចការ​សម្រាប់​ម៉ែត្រ​ជាក់លាក់​ផ្សេងៗ​អាច​លឿន​ជាង​និង​ងាយស្រួល​ជាង​ការ​ប្រើ APIs ណាមួយ។

  ឧបមាថា កិច្ចការទាំងអស់របស់យើងមិនអស់កំលាំងទេ ប៉ុន្តែពេលខ្លះវាអាចធ្លាក់ចុះ ហើយនេះជារឿងធម្មតា។ ប៉ុន្តែការស្ទះមួយចំនួនមានការសង្ស័យរួចហើយ ហើយចាំបាច់ត្រូវពិនិត្យ។

  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 — មគ្គុទ្ទេសក៍ស្តីពីរបៀបបង្កើតម៉ាស៊ីនមេលំហូរខ្យល់ - មគ្គុទ្ទេសក៍ខ្លីៗសម្រាប់ការដំឡើង Airflow cluster ។
• ដំណើរការ Apache Airflow នៅ Lyft - ស្ទើរតែអត្ថបទដែលគួរឱ្យចាប់អារម្មណ៍ដូចគ្នា លើកលែងតែ ប្រហែលជាផ្លូវការ និងឧទាហរណ៍តិចជាង។
• របៀបដែល Apache Airflow ចែកចាយការងារលើកម្មករ Celery - អំពីការធ្វើការជាមួយ Celery ។
• DAG សរសេរការអនុវត្តល្អបំផុតនៅក្នុង Apache Airflow - អំពីភាពគ្មានសមត្ថភាពនៃភារកិច្ច ការផ្ទុកដោយលេខសម្គាល់ជំនួសឱ្យកាលបរិច្ឆេទ ការផ្លាស់ប្តូរ រចនាសម្ព័ន្ធឯកសារ និងអ្វីដែលគួរឱ្យចាប់អារម្មណ៍ផ្សេងទៀត។
• ការគ្រប់គ្រងភាពអាស្រ័យនៅក្នុង Apache Airflow - ភាពអាស្រ័យនៃភារកិច្ច និងច្បាប់គន្លឹះ ដែលខ្ញុំបានលើកឡើងតែក្នុងការឆ្លងកាត់ប៉ុណ្ណោះ។
• លំហូរខ្យល់៖ នៅពេលដែល DAG របស់អ្នកនៅឆ្ងាយពីកាលវិភាគ - របៀបយកឈ្នះលើ "ការងារដូចបំណង" មួយចំនួននៅក្នុងកម្មវិធីកំណត់ពេល ផ្ទុកទិន្នន័យដែលបាត់បង់ និងកំណត់អាទិភាពកិច្ចការ។
• សំណួរ SQL មានប្រយោជន៍សម្រាប់ Apache Airflow - សំណួរ SQL មានប្រយោជន៍ចំពោះទិន្នន័យមេតាលំហូរខ្យល់។
• ចាប់ផ្តើមបង្កើតលំហូរការងារជាមួយ Apache Airflow - មានផ្នែកមានប្រយោជន៍អំពីការបង្កើតឧបករណ៍ចាប់សញ្ញាផ្ទាល់ខ្លួន។
• ការកសាង Fetchr Data Science Infra នៅលើ AWS ជាមួយ Presto និង Airflow - កំណត់ចំណាំខ្លីគួរឱ្យចាប់អារម្មណ៍អំពីការកសាងហេដ្ឋារចនាសម្ព័ន្ធនៅលើ AWS សម្រាប់វិទ្យាសាស្ត្រទិន្នន័យ។
• កំហុសទូទៅចំនួន 7 ដើម្បីពិនិត្យមើលនៅពេលបំបាត់កំហុស DAGs លំហូរខ្យល់ - កំហុសទូទៅ (នៅពេលដែលនរណាម្នាក់នៅតែមិនអានការណែនាំ) ។
• រក្សាទុក និងចូលប្រើពាក្យសម្ងាត់ដោយប្រើ Apache Airflow - ញញឹមពីរបៀបដែលមនុស្សចាប់យកលេខសម្ងាត់ ទោះបីអ្នកគ្រាន់តែអាចប្រើ Connections ក៏ដោយ។
• Zen នៃ Python និង Apache Airflow - ការបញ្ជូនបន្ត DAG ដោយប្រយោល ការបោះចោលបរិបទនៅក្នុងមុខងារ ជាថ្មីម្តងទៀតអំពីភាពអាស្រ័យ និងអំពីការរំលងការចាប់ផ្ដើមកិច្ចការ។
• លំហូរខ្យល់៖ គន្លឹះខ្លីៗ ល្បិច និងការអនុវត្តល្អបំផុត - អំពីការប្រើប្រាស់ default arguments и params នៅក្នុងគំរូ ក៏ដូចជាអថេរ និងការតភ្ជាប់។
• កំណត់​តម្លៃ​កម្មវិធី​កំណត់​ពេល​លំហូរ​ខ្យល់ - រឿងអំពីរបៀបដែលអ្នករៀបចំផែនការកំពុងរៀបចំសម្រាប់ Airflow 2.0 ។
• Apache Airflow ជាមួយបុគ្គលិក Celery 3 នាក់ក្នុង docker-compose - អត្ថបទហួសសម័យបន្តិចអំពីការពង្រាយចង្កោមរបស់យើងនៅក្នុង docker-compose.
• 4 ការងារគំរូដោយប្រើបរិបទលំហូរខ្យល់ - កិច្ចការថាមវន្តដោយប្រើគំរូ និងការបញ្ជូនបន្តបរិបទ។
• ការជូនដំណឹងអំពីបញ្ហានៅក្នុងលំហូរខ្យល់ - ការជូនដំណឹងតាមស្តង់ដារ និងផ្ទាល់ខ្លួនតាមសំបុត្រ និង Slack ។
• សិក្ខាសាលាលំហូរខ្យល់៖ DAGs ស្មុគស្មាញដោយគ្មានឈើច្រត់ - កិច្ចការសាខា ម៉ាក្រូ និង XCom ។

និងតំណភ្ជាប់ដែលប្រើក្នុងអត្ថបទ៖

• សេចក្តីយោងម៉ាក្រូ - កន្លែងដាក់មានសម្រាប់ប្រើក្នុងគំរូ។
• ឧបសគ្គទូទៅ - លំហូរខ្យល់ - កំហុសទូទៅនៅពេលបង្កើត dags ។
• puckel/docker-airflow៖ Docker Apache Airflow - docker-compose សម្រាប់ការពិសោធន៍ ការបំបាត់កំហុស និងច្រើនទៀត។
• python-telegram-bot/python-telegram-bot៖ យើង​បាន​បង្កើត​អ្នក​ជា​អ្នក​រុំ​ដែល​អ្នក​មិន​អាច​បដិសេធ - Python wrapper សម្រាប់ Telegram REST API ។

ប្រភព: www.habr.com