Apache Airflow: ເຮັດໃຫ້ ETL ງ່າຍຂຶ້ນ

ສະບາຍດີ, ຂ້ອຍແມ່ນ Dmitry Logvinenko - ວິສະວະກອນຂໍ້ມູນຂອງພະແນກການວິເຄາະຂອງກຸ່ມບໍລິສັດ Vezet.

ຂ້າພະເຈົ້າຈະບອກທ່ານກ່ຽວກັບເຄື່ອງມືທີ່ປະເສີດສໍາລັບການພັດທະນາຂະບວນການ ETL - Apache Airflow. ແຕ່ Airflow ແມ່ນມີຄວາມຫລາກຫລາຍແລະມີຫຼາຍຮູບແບບທີ່ທ່ານຄວນພິຈາລະນາເບິ່ງມັນຢ່າງໃກ້ຊິດເຖິງແມ່ນວ່າທ່ານຈະບໍ່ມີສ່ວນຮ່ວມໃນການໄຫລຂອງຂໍ້ມູນ, ແຕ່ຕ້ອງມີການເປີດຕົວຂະບວນການຕ່າງໆເປັນໄລຍະແລະຕິດຕາມການປະຕິບັດຂອງພວກເຂົາ.

ແລະແມ່ນແລ້ວ, ຂ້າພະເຈົ້າຈະບໍ່ພຽງແຕ່ບອກ, ແຕ່ຍັງສະແດງໃຫ້ເຫັນ: ໂຄງການມີຫຼາຍລະຫັດ, screenshots ແລະຄໍາແນະນໍາ.

ສິ່ງທີ່ທ່ານມັກຈະເຫັນໃນເວລາທີ່ທ່ານ google ຄໍາວ່າ Airflow / Wikimedia Commons

ຕາຕະລາງເນື້ອຫາ

• ການນໍາສະເຫນີ
• ພາກສ່ວນຕົ້ນຕໍ, ການປະຕິບັດ (ແລະທິດສະດີເລັກນ້ອຍ)
  • ເປັນຫຍັງພວກເຮົາ (ແລະເຈົ້າ)
  • ການປະກອບກຸ່ມ
  • ແນວຄວາມຄິດພື້ນຖານ
  • ພວກເຮົາສ້າງວຽກງານ
  • ເລັກນ້ອຍກ່ຽວກັບດອກໄມ້
  • ພວກເຮົາໂຫຼດ underloaded
  • ການເຊື່ອມຕໍ່, hooks ແລະຕົວແປອື່ນໆ
  • ກຳລັງວິເຄາະຕົວດຳເນີນການແບບກຳນົດເອງ
  • ເຈົ້າຈະຖອກບໍ?
  • Summing up
• ສ່ວນສຸດທ້າຍ, ຂໍ້ມູນອ້າງອີງ ແລະຂໍ້ມູນ
  • rake ທີ່​ພວກ​ເຮົາ​ໄດ້​ເກັບ​ກໍາ​ສໍາ​ລັບ​ທ່ານ​
  • ເຄື່ອງມືອັດຕະໂນມັດເພີ່ມເຕີມ
• ເອກະສານ

ການນໍາສະເຫນີ

Apache Airflow ແມ່ນຄືກັນກັບ Django:

• ຂຽນໃນ python
• ມີຄະນະບໍລິຫານງານທີ່ດີ,
• ສາມາດຂະຫຍາຍໄດ້ຢ່າງບໍ່ມີກຳນົດ

- ພຽງ​ແຕ່​ດີກ​ວ່າ​, ແລະ​ມັນ​ໄດ້​ຖືກ​ສ້າງ​ຂຶ້ນ​ສໍາ​ລັບ​ຈຸດ​ປະ​ສົງ​ທີ່​ແຕກ​ຕ່າງ​ກັນ​ຫມົດ​, ຄື (ດັ່ງ​ທີ່​ມັນ​ໄດ້​ຖືກ​ຂຽນ​ໄວ້​ກ່ອນ kat​)​:

• ແລ່ນ​ແລະ​ຕິດ​ຕາມ​ກວດ​ກາ​ວຽກ​ງານ​ໃນ​ຈໍາ​ນວນ​ບໍ່​ຈໍາ​ກັດ​ຂອງ​ເຄື່ອງ​ຈັກ (ເປັນ Celery / Kubernetes ຫຼາຍ​ແລະ​ຈິດ​ສໍາ​ນຶກ​ຂອງ​ທ່ານ​ຈະ​ອະ​ນຸ​ຍາດ​ໃຫ້​ທ່ານ​)
• ດ້ວຍການສ້າງຂະບວນການເຮັດວຽກແບບເຄື່ອນໄຫວຈາກການຂຽນແລະເຂົ້າໃຈລະຫັດ Python ງ່າຍຫຼາຍ
• ແລະຄວາມສາມາດໃນການເຊື່ອມຕໍ່ຖານຂໍ້ມູນແລະ APIs ກັບກັນແລະກັນໂດຍໃຊ້ອົງປະກອບທີ່ກຽມພ້ອມແລະ plugins ທີ່ເຮັດໃນເຮືອນ (ເຊິ່ງງ່າຍດາຍທີ່ສຸດ).

ພວກເຮົາໃຊ້ Apache Airflow ແບບນີ້:

• ພວກເຮົາເກັບກໍາຂໍ້ມູນຈາກແຫຼ່ງຕ່າງໆ (ຫຼາຍໆ SQL Server ແລະ PostgreSQL instances, APIs ຕ່າງໆທີ່ມີ metrics ຄໍາຮ້ອງສະຫມັກ, ເຖິງແມ່ນວ່າ 1C) ໃນ DWH ແລະ ODS (ພວກເຮົາມີ Vertica ແລະ Clickhouse).
• ກ້າວຫນ້າແນວໃດ cron, ເຊິ່ງເລີ່ມຕົ້ນຂະບວນການລວມຂໍ້ມູນຢູ່ໃນ ODS, ແລະຍັງຕິດຕາມການບໍາລຸງຮັກສາຂອງພວກເຂົາ.

ຈົນກ່ວາບໍ່ດົນມານີ້, ຄວາມຕ້ອງການຂອງພວກເຮົາໄດ້ຖືກຄຸ້ມຄອງໂດຍເຄື່ອງແມ່ຂ່າຍຂະຫນາດນ້ອຍຫນຶ່ງທີ່ມີ 32 cores ແລະ 50 GB ຂອງ RAM. ໃນ Airflow, ນີ້ເຮັດວຽກ:

• ຫຼາຍ 200 ດາດ (ຕົວຈິງແລ້ວ workflows, ທີ່ພວກເຮົາ stuffed ວຽກງານ),
• ໃນ​ແຕ່​ລະ​ໂດຍ​ສະ​ເລ່ຍ​ 70 ວຽກງານ,
• ຄວາມດີນີ້ເລີ່ມຕົ້ນ (ໂດຍສະເລ່ຍ) ຫນຶ່ງຊົ່ວໂມງ.

ແລະກ່ຽວກັບວິທີທີ່ພວກເຮົາຂະຫຍາຍ, ຂ້ອຍຈະຂຽນຂ້າງລຸ່ມນີ້, ແຕ່ຕອນນີ້ໃຫ້ກໍານົດບັນຫາ über ທີ່ພວກເຮົາຈະແກ້ໄຂ:

ມີສາມແຫຼ່ງ SQL Servers, ແຕ່ລະຄົນມີ 50 ຖານຂໍ້ມູນ - ຕົວຢ່າງຂອງໂຄງການຫນຶ່ງ, ຕາມລໍາດັບ, ພວກເຂົາມີໂຄງສ້າງດຽວກັນ (ເກືອບທຸກບ່ອນ, mua-ha-ha), ຊຶ່ງຫມາຍຄວາມວ່າແຕ່ລະຄົນມີຕາຕະລາງຄໍາສັ່ງ (ໂຊກດີ, ຕາຕະລາງທີ່ມີນັ້ນ. ຊື່ສາມາດກົດເຂົ້າໄປໃນທຸລະກິດໃດກໍ່ຕາມ). ພວກເຮົາເອົາຂໍ້ມູນໂດຍການເພີ່ມຊ່ອງບໍລິການ (ເຄື່ອງແມ່ຂ່າຍແຫຼ່ງ, ຖານຂໍ້ມູນແຫຼ່ງ, ETL task ID) ແລະຖິ້ມພວກມັນໃສ່, ເວົ້າ, Vertica.

ໃຫ້ໄປ!

ພາກສ່ວນຕົ້ນຕໍ, ການປະຕິບັດ (ແລະທິດສະດີເລັກນ້ອຍ)

ເປັນຫຍັງພວກເຮົາ (ແລະເຈົ້າ)

ໃນເວລາທີ່ຕົ້ນໄມ້ໃຫຍ່ແລະຂ້າພະເຈົ້າງ່າຍດາຍ SQL-schik ໃນ​ການ​ຄ້າ​ຂາຍ​ຍ່ອຍ​ຂອງ​ລັດ​ເຊຍ​ຫນຶ່ງ​, ພວກ​ເຮົາ​ຫລອກ​ລວງ​ຂະ​ບວນ​ການ ETL aka ການ​ໄຫຼ​ຂອງ​ຂໍ້​ມູນ​ໂດຍ​ນໍາ​ໃຊ້​ສອງ​ເຄື່ອງ​ມື​ທີ່​ມີ​ໃຫ້​ພວກ​ເຮົາ​:

• ສູນພະລັງງານ Informatica - ເປັນ​ລະ​ບົບ​ການ​ແຜ່​ກະ​ຈາຍ​ທີ່​ສຸດ​, ຜະ​ລິດ​ຕະ​ພັນ​ທີ່​ສຸດ​, ມີ​ຮາດ​ແວ​ຂອງ​ຕົນ​ເອງ​, ສະ​ບັບ​ຂອງ​ຕົນ​ເອງ​. ຂ້າພະເຈົ້າໄດ້ໃຊ້ພຣະເຈົ້າຫ້າມ 1% ຂອງຄວາມສາມາດຂອງມັນ. ເປັນຫຍັງ? ດີ, ກ່ອນອື່ນ ໝົດ, ການໂຕ້ຕອບນີ້, ຢູ່ບ່ອນໃດບ່ອນ ໜຶ່ງ ຈາກຊຸມປີ 380, ຄວາມກົດດັນທາງຈິດໃຈຕໍ່ພວກເຮົາ. ອັນທີສອງ, contraption ນີ້ໄດ້ຖືກອອກແບບສໍາລັບຂະບວນການ fancy ທີ່ສຸດ, ອົງປະກອບ furious reuse ແລະ tricks ທີ່ສໍາຄັນຫຼາຍວິສາຫະກິດອື່ນໆ. ກ່ຽວກັບຄວາມຈິງທີ່ວ່າມັນມີຄ່າໃຊ້ຈ່າຍ, ເຊັ່ນປີກຂອງ Airbus AXNUMX / ປີ, ພວກເຮົາຈະບໍ່ເວົ້າຫຍັງ.

  ລະວັງ, ພາບໜ້າຈໍອາດເຮັດໃຫ້ຄົນອາຍຸຕໍ່າກວ່າ 30 ປີບາດເຈັບເລັກນ້ອຍ

  

• SQL Server Integration Server - ພວກ​ເຮົາ​ໄດ້​ນໍາ​ໃຊ້ comrade ນີ້​ໃນ​ການ​ໄຫຼ​ເຂົ້າ​ຂອງ​ໂຄງ​ການ​ຂອງ​ພວກ​ເຮົາ​. ແລ້ວ, ໃນຄວາມເປັນຈິງ: ພວກເຮົາໃຊ້ SQL Server ແລ້ວ, ແລະມັນອາດຈະບໍ່ສົມເຫດສົມຜົນທີ່ຈະບໍ່ໃຊ້ເຄື່ອງມື ETL ຂອງມັນ. ທຸກສິ່ງທຸກຢ່າງຢູ່ໃນມັນດີ: ທັງການໂຕ້ຕອບແມ່ນສວຍງາມ, ແລະລາຍງານຄວາມຄືບຫນ້າ ... ແຕ່ນີ້ບໍ່ແມ່ນເຫດຜົນທີ່ພວກເຮົາຮັກຜະລິດຕະພັນຊອບແວ, ໂອ້, ບໍ່ແມ່ນສໍາລັບເລື່ອງນີ້. ລຸ້ນມັນ dtsx (ເຊິ່ງແມ່ນ XML ກັບ nodes shuffled on save) ພວກເຮົາສາມາດ, ແຕ່ສິ່ງທີ່ເປັນຈຸດ? ເຮັດແນວໃດກ່ຽວກັບການສ້າງຊຸດວຽກທີ່ຈະລາກຫຼາຍຮ້ອຍຕາຕະລາງຈາກເຄື່ອງແມ່ຂ່າຍຫນຶ່ງໄປຫາອື່ນ? ແມ່ນແລ້ວ, ສິ່ງທີ່ເປັນຮ້ອຍ, ນິ້ວມືດັດສະນີຂອງທ່ານຈະຫຼຸດລົງຈາກຊາວຕ່ອນ, ຄລິກໃສ່ປຸ່ມຫນູ. ແຕ່ແນ່ນອນມັນເບິ່ງຄົນອັບເດດ: ຫຼາຍ:

  

ພວກເຮົາແນ່ນອນຊອກຫາວິທີທາງອອກ. ກໍລະນີ ເກືອບ ມາຮອດເຄື່ອງຜະລິດຊຸດ SSIS ທີ່ຂຽນດ້ວຍຕົນເອງ ...

… ແລ້ວວຽກໃໝ່ກໍພົບຂ້ອຍ. ແລະ Apache Airflow overtook ຂ້າພະເຈົ້າກ່ຽວກັບມັນ.

ເມື່ອຂ້ອຍພົບວ່າຄໍາອະທິບາຍກ່ຽວກັບຂະບວນການ ETL ແມ່ນລະຫັດ Python ງ່າຍດາຍ, ຂ້ອຍບໍ່ໄດ້ເຕັ້ນລໍາສໍາລັບຄວາມສຸກ. ນີ້ແມ່ນວິທີການຖ່າຍທອດຂໍ້ມູນແລະຄວາມແຕກຕ່າງ, ແລະການຖອກເທຕາຕະລາງທີ່ມີໂຄງສ້າງດຽວຈາກຫຼາຍຮ້ອຍຖານຂໍ້ມູນເຂົ້າໄປໃນເປົ້າຫມາຍດຽວໄດ້ກາຍເປັນເລື່ອງຂອງລະຫັດ Python ໃນຫນຶ່ງແລະເຄິ່ງຫນຶ່ງຫຼືສອງຫນ້າຈໍ 13 ".

ການປະກອບກຸ່ມ

ບໍ່ໃຫ້ຈັດແຈງໂຮງຮຽນອະນຸບານຢ່າງສົມບູນ, ແລະບໍ່ໄດ້ເວົ້າກ່ຽວກັບສິ່ງທີ່ຊັດເຈນຢູ່ທີ່ນີ້, ເຊັ່ນ: ການຕິດຕັ້ງ Airflow, ຖານຂໍ້ມູນທີ່ທ່ານເລືອກ, Celery ແລະກໍລະນີອື່ນໆທີ່ອະທິບາຍໄວ້ໃນ docks.

ເພື່ອໃຫ້ພວກເຮົາສາມາດເລີ່ມຕົ້ນການທົດລອງໄດ້ທັນທີ, ຂ້າພະເຈົ້າໄດ້ແຕ້ມຮູບ docker-compose.yml ໃນນັ້ນ:

• ໃຫ້ຂອງຕົວຈິງຍົກສູງ ກະແສອາກາດ: Scheduler, Webserver. ດອກໄມ້ຍັງຈະຖືກຫມຸນຢູ່ທີ່ນັ້ນເພື່ອຕິດຕາມວຽກງານຂອງ Celery (ເນື່ອງຈາກວ່າມັນໄດ້ຖືກ pushed ແລ້ວ apache/airflow:1.10.10-python3.7, ແຕ່​ພວກ​ເຮົາ​ບໍ່​ສົນ​ໃຈ​)
• PostgreSQL, ໃນທີ່ Airflow ຈະຂຽນຂໍ້ມູນການບໍລິການຂອງມັນ (ຂໍ້ມູນຕາຕະລາງ, ສະຖິຕິການປະຕິບັດ, ແລະອື່ນໆ), ແລະ Celery ຈະຫມາຍວຽກງານທີ່ສໍາເລັດ;
• Redis, ເຊິ່ງຈະເຮັດຫນ້າທີ່ເປັນນາຍຫນ້າສໍາລັບ Celery;
• ພະນັກງານ Celery, ເຊິ່ງຈະໄດ້ຮັບການດໍາເນີນການໂດຍກົງຂອງວຽກງານ.
• ໄປທີ່ໂຟນເດີ ./dags ພວກເຮົາຈະເພີ່ມໄຟລ໌ຂອງພວກເຮົາດ້ວຍຄໍາອະທິບາຍຂອງ dags. ພວກມັນຈະຖືກເກັບຂຶ້ນເທິງແມງວັນ, ສະນັ້ນບໍ່ຈຳເປັນທີ່ຈະຈູດກອງທັງໝົດຫຼັງຈາກຈາມແຕ່ລະຄັ້ງ.

ໃນບາງບ່ອນ, ລະຫັດໃນຕົວຢ່າງບໍ່ໄດ້ສະແດງໃຫ້ເຫັນຢ່າງສົມບູນ (ເພື່ອບໍ່ໃຫ້ clutter ຂໍ້ຄວາມ), ແຕ່ບາງບ່ອນມັນຖືກດັດແກ້ໃນຂະບວນການ. ຕົວຢ່າງລະຫັດການເຮັດວຽກທີ່ສົມບູນສາມາດພົບໄດ້ຢູ່ໃນບ່ອນເກັບມ້ຽນ 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 ທັງຫມົດແມ່ນມີຢູ່ບໍ່ພຽງແຕ່ຜ່ານ airflow.cfg, ແຕ່ຍັງຜ່ານຕົວແປສະພາບແວດລ້ອມ (ຂໍຂອບໃຈກັບຜູ້ພັດທະນາ), ເຊິ່ງຂ້ອຍໄດ້ໃຊ້ປະໂຫຍດຈາກອັນຕະລາຍ.
• ຕາມທໍາມະຊາດ, ມັນບໍ່ແມ່ນການຜະລິດທີ່ກຽມພ້ອມ: ຂ້າພະເຈົ້າເຈດຕະນາບໍ່ໄດ້ໃສ່ການເຕັ້ນຂອງຫົວໃຈໃສ່ຖັງ, ຂ້າພະເຈົ້າບໍ່ໄດ້ລົບກວນຄວາມປອດໄພ. ແຕ່ຂ້ອຍໄດ້ເຮັດຕໍາ່ສຸດທີ່ເຫມາະສົມສໍາລັບນັກທົດລອງຂອງພວກເຮົາ.
• ໃຫ້ສັງເກດວ່າ:
  • ໂຟນເດີ dag ຕ້ອງສາມາດເຂົ້າເຖິງໄດ້ທັງຜູ້ກຳນົດເວລາ ແລະຄົນງານ.
  • ດຽວກັນໃຊ້ກັບຫ້ອງສະຫມຸດພາກສ່ວນທີສາມທັງຫມົດ - ພວກເຂົາທັງຫມົດຕ້ອງໄດ້ຮັບການຕິດຕັ້ງຢູ່ໃນເຄື່ອງຈັກທີ່ມີຕາຕະລາງແລະພະນັກງານ.

ແລ້ວ, ດຽວນີ້ມັນງ່າຍດາຍ:

$ docker-compose up --scale worker=3

ຫຼັງຈາກທີ່ທຸກສິ່ງທຸກຢ່າງເພີ່ມຂຶ້ນ, ທ່ານສາມາດເບິ່ງໃນການໂຕ້ຕອບຂອງເວັບໄຊຕ໌:

• ການບິນ: http://127.0.0.1:8080/admin/
• ດອກໄມ້: http://127.0.0.1:5555/dashboard

ແນວຄວາມຄິດພື້ນຖານ

ຖ້າທ່ານບໍ່ເຂົ້າໃຈສິ່ງໃດໃນ "dags", ນີ້ແມ່ນວັດຈະນານຸກົມສັ້ນ:

• Scheduler - ລຸງທີ່ສໍາຄັນທີ່ສຸດໃນ Airflow, ການຄວບຄຸມທີ່ຫຸ່ນຍົນເຮັດວຽກຫນັກ, ແລະບໍ່ແມ່ນບຸກຄົນ: ຕິດຕາມກວດກາຕາຕະລາງ, ປັບປຸງ dags, ເປີດຕົວວຽກງານ.

  ໂດຍທົ່ວໄປແລ້ວ, ໃນຮຸ່ນເກົ່າ, ລາວມີບັນຫາກັບຄວາມຊົງຈໍາ (ບໍ່, ບໍ່ແມ່ນຄວາມຈໍາ, ແຕ່ການຮົ່ວໄຫຼ) ແລະຕົວກໍານົດການມໍລະດົກຍັງຢູ່ໃນການຕັ້ງຄ່າ. run_duration — ໄລ​ຍະ​ການ​ເລີ່ມ​ຕົ້ນ​ໃຫມ່​ຂອງ​ຕົນ​. ແຕ່ດຽວນີ້ທຸກຢ່າງດີແລ້ວ.

• DAG (aka "dug") - "ກາຟ acyclic ຊີ້", ແຕ່ຄໍານິຍາມດັ່ງກ່າວຈະບອກຄົນຈໍານວນຫນ້ອຍຫນຶ່ງ, ແຕ່ໃນຄວາມເປັນຈິງມັນເປັນຖັງສໍາລັບວຽກງານທີ່ພົວພັນກັບກັນແລະກັນ (ເບິ່ງຂ້າງລຸ່ມນີ້) ຫຼືການປຽບທຽບຂອງ Package ໃນ SSIS ແລະ Workflow ໃນ Informatica .

  ນອກເຫນືອຈາກ dags, ມັນອາດຈະຍັງມີ subdags, ແຕ່ພວກເຮົາສ່ວນຫຼາຍອາດຈະບໍ່ເຂົ້າຫາພວກມັນ.

• ແລ່ນ DAG - ເລີ່ມຕົ້ນ dag, ເຊິ່ງມອບຫມາຍຂອງຕົນເອງ execution_date. Dagrans ຂອງ dag ດຽວກັນສາມາດເຮັດວຽກຂະຫນານກັນ (ຖ້າທ່ານເຮັດໃຫ້ວຽກງານຂອງທ່ານບໍ່ມີຄວາມເຂັ້ມແຂງ, ແນ່ນອນ).
• Operator ແມ່ນຊິ້ນສ່ວນຂອງລະຫັດທີ່ຮັບຜິດຊອບສໍາລັບການດໍາເນີນການສະເພາະໃດຫນຶ່ງ. ມີສາມປະເພດຂອງຜູ້ປະກອບການ:
  • ການປະຕິບັດຄື favorite ຂອງພວກເຮົາ PythonOperator, ເຊິ່ງສາມາດປະຕິບັດລະຫັດ Python ໃດໆ (ທີ່ຖືກຕ້ອງ);
  • ການໂອນ, ເຊິ່ງການຂົນສົ່ງຂໍ້ມູນຈາກສະຖານທີ່ໄປຫາສະຖານທີ່, ເວົ້າວ່າ, MsSqlToHiveTransfer;
  • sensor ໃນທາງກົງກັນຂ້າມ, ມັນຈະຊ່ວຍໃຫ້ທ່ານສາມາດຕອບໂຕ້ຫຼືຊ້າລົງການປະຕິບັດຕໍ່ໄປຂອງ dag ຈົນກ່ວາເຫດການເກີດຂື້ນ. HttpSensor ສາມາດດຶງຈຸດສິ້ນສຸດທີ່ລະບຸໄວ້, ແລະເມື່ອຄໍາຕອບທີ່ຕ້ອງການລໍຖ້າ, ເລີ່ມຕົ້ນການໂອນ GoogleCloudStorageToS3Operator. ຈິດໃຈທີ່ສົງໄສຈະຖາມວ່າ: “ເປັນຫຍັງ? ຫຼັງຈາກທີ່ທັງຫມົດ, ທ່ານສາມາດເຮັດຊ້ໍາໄດ້ໃນຕົວປະຕິບັດການ!” ແລະຫຼັງຈາກນັ້ນ, ໃນຄໍາສັ່ງທີ່ຈະບໍ່ clog ສະນຸກເກີຂອງວຽກງານທີ່ມີຜູ້ປະຕິບັດການໂຈະ. ເຊັນເຊີເລີ່ມຕົ້ນ, ກວດສອບແລະຕາຍກ່ອນທີ່ຈະພະຍາຍາມຕໍ່ໄປ.
• Task - ຜູ້ປະຕິບັດການປະກາດ, ບໍ່ວ່າຈະເປັນປະເພດໃດ, ແລະຕິດກັບດາກໄດ້ຖືກເລື່ອນຊັ້ນໃນຫນ້າວຽກ.
• ຕົວຢ່າງໜ້າວຽກ - ເມື່ອຜູ້ວາງແຜນທົ່ວໄປຕັດສິນໃຈວ່າມັນແມ່ນເວລາທີ່ຈະສົ່ງວຽກງານເຂົ້າໄປໃນການສູ້ຮົບກັບນັກສະແດງ - ພະນັກງານ (ຢູ່ໃນຈຸດ, ຖ້າພວກເຮົາໃຊ້ LocalExecutor ຫຼືກັບ node ຫ່າງໄກສອກຫຼີກໃນກໍລະນີຂອງ CeleryExecutor), ມັນກໍານົດບໍລິບົດໃຫ້ພວກເຂົາ (i. e., ຊຸດຂອງຕົວແປ - ພາລາມິເຕີການປະຕິບັດ), ຂະຫຍາຍຄໍາສັ່ງຫຼືແບບສອບຖາມ, ແລະປະກອບໃຫ້ເຂົາເຈົ້າ.

ພວກເຮົາສ້າງວຽກງານ

ທໍາອິດ, ໃຫ້ອະທິບາຍໂຄງການທົ່ວໄປຂອງ doug ຂອງພວກເຮົາ, ແລະຫຼັງຈາກນັ້ນພວກເຮົາຈະເຂົ້າໄປໃນລາຍລະອຽດເພີ່ມເຕີມແລະຫຼາຍ, ເພາະວ່າພວກເຮົານໍາໃຊ້ບາງວິທີແກ້ໄຂທີ່ບໍ່ແມ່ນ trivial.

ດັ່ງນັ້ນ, ໃນຮູບແບບທີ່ງ່າຍດາຍທີ່ສຸດ, ດັ່ງກ່າວຈະມີລັກສະນະນີ້:

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 - ການປະກາດຂອງ doug ຂອງພວກເຮົາ, ເຊິ່ງຈໍາເປັນຕ້ອງມີຢູ່ໃນ globals()ຖ້າບໍ່ດັ່ງນັ້ນ Airflow ຈະບໍ່ພົບມັນ. Doug ຍັງຕ້ອງການເວົ້າວ່າ:
  • ລາວຊື່ຫຍັງ orders - ຫຼັງ​ຈາກ​ນັ້ນ​ຊື່​ນີ້​ຈະ​ປາ​ກົດ​ຢູ່​ໃນ​ການ​ໂຕ້​ຕອບ​ຂອງ​ເວັບ​ໄຊ​ຕ​໌​,
  • ວ່າລາວຈະເຮັດວຽກຕັ້ງແຕ່ທ່ຽງຄືນຂອງວັນທີແປດຂອງເດືອນກໍລະກົດ,
  • ແລະມັນຄວນຈະດໍາເນີນການ, ປະມານທຸກໆ 6 ຊົ່ວໂມງ (ສໍາລັບ guys tough ທີ່ນີ້ແທນທີ່ຈະເປັນ timedelta() ຍອມຮັບໄດ້ cron- ສາຍ 0 0 0/6 ? * * *, ສໍາລັບເຢັນຫນ້ອຍ - ການສະແດງອອກຄື @daily);
• workflow() ຈະເຮັດວຽກຕົ້ນຕໍ, ແຕ່ບໍ່ແມ່ນໃນປັດຈຸບັນ. ສໍາລັບໃນປັດຈຸບັນ, ພວກເຮົາຈະຖິ້ມບໍລິບົດຂອງພວກເຮົາເຂົ້າໄປໃນບັນທຶກ.
• ແລະໃນປັດຈຸບັນ magic ງ່າຍດາຍຂອງການສ້າງວຽກງານ:
  • ພວກເຮົາດໍາເນີນການໂດຍຜ່ານແຫຼ່ງຂອງພວກເຮົາ;
  • ເລີ່ມຕົ້ນ PythonOperator, ເຊິ່ງຈະປະຕິບັດ dummy ຂອງພວກເຮົາ workflow(). ຢ່າ​ລືມ​ລະ​ບຸ​ຊື່​ທີ່​ເປັນ​ເອ​ກະ​ລັກ (ພາຍ​ໃນ​ດາ​ກ​) ຂອງ​ວຽກ​ງານ​ແລະ​ຜູກ​ພັນ​ຂອງ​ຕົນ​ເອງ​. ທຸງ provide_context ໃນທາງກັບກັນ, ຈະຖອກເທການໂຕ້ຖຽງເພີ່ມເຕີມເຂົ້າໄປໃນຫນ້າທີ່, ເຊິ່ງພວກເຮົາຈະເກັບກໍາຢ່າງລະມັດລະວັງໂດຍໃຊ້ **context.

ສໍາລັບໃນປັດຈຸບັນ, ນັ້ນແມ່ນທັງຫມົດ. ສິ່ງທີ່ພວກເຮົາໄດ້ຮັບ:

• dag ໃຫມ່ໃນການໂຕ້ຕອບເວັບໄຊຕ໌,
• ຫນຶ່ງແລະເຄິ່ງຫນຶ່ງເປັນຮ້ອຍຫນ້າວຽກທີ່ຈະຖືກປະຕິບັດໃນຂະຫນານ (ຖ້າ Airflow, ການຕັ້ງຄ່າ Celery ແລະຄວາມສາມາດຂອງເຄື່ອງແມ່ຂ່າຍອະນຸຍາດໃຫ້ມັນ).

ດີ, ເກືອບໄດ້ຮັບມັນ.

ໃຜຈະຕິດຕັ້ງການເພິ່ງພາອາໄສ?

ເພື່ອເຮັດໃຫ້ສິ່ງທັງຫມົດນີ້ງ່າຍດາຍ, ຂ້າພະເຈົ້າໄດ້ screwed ໃນ docker-compose.yml ການປຸງແຕ່ງ requirements.txt ໃນທຸກ nodes.

ໃນປັດຈຸບັນມັນຫມົດໄປ:

ສີ່ຫຼ່ຽມ​ສີ​ຂີ້​ເຖົ່າ​ແມ່ນ​ຕົວ​ຢ່າງ​ວຽກ​ງານ​ປະ​ມວນ​ຜົນ​ໂດຍ​ການ​ຕັ້ງ​ເວ​ລາ​.

ພວກ​ເຮົາ​ລໍ​ຖ້າ​ອີກ​ໜ້ອຍ​ໜຶ່ງ, ວຽກ​ງານ​ທີ່​ກຳ​ມະ​ກອນ​ໄດ້​ຖືກ​ຈັບ​ກຸມ:

ສີຂຽວ, ແນ່ນອນ, ໄດ້ສໍາເລັດວຽກງານຂອງເຂົາເຈົ້າ. Reds ບໍ່ປະສົບຜົນສໍາເລັດຫຼາຍ.

ໂດຍວິທີທາງການ, ບໍ່ມີໂຟນເດີກ່ຽວກັບຜະລິດຕະພັນຂອງພວກເຮົາ ./dags, ບໍ່ມີການ synchronization ລະຫວ່າງເຄື່ອງຈັກ - dags ທັງຫມົດນອນຢູ່ໃນ git ຢູ່ໃນ Gitlab ຂອງພວກເຮົາ, ແລະ Gitlab CI ແຈກຢາຍການອັບເດດໃຫ້ກັບເຄື່ອງຈັກເມື່ອລວມເຂົ້າກັນ master.

ເລັກນ້ອຍກ່ຽວກັບດອກໄມ້

ໃນຂະນະທີ່ຄົນງານກຳລັງຕີເຄື່ອງປັ້ນດິນເຜົາຂອງພວກເຮົາ, ໃຫ້ເຮົາຈື່ຈຳເຄື່ອງມືໜຶ່ງທີ່ສາມາດສະແດງໃຫ້ເຮົາເຫັນໄດ້ - ດອກໄມ້.

ຫນ້າທໍາອິດທີ່ມີຂໍ້ມູນສະຫຼຸບກ່ຽວກັບຂໍ້ຂອງພະນັກງານ:

ໜ້າວຽກທີ່ເຂັ້ມງວດທີ່ສຸດກັບວຽກທີ່ໄປເຮັດວຽກ:

ຫນ້າເບື່ອທີ່ສຸດກັບສະຖານະຂອງນາຍຫນ້າຂອງພວກເຮົາ:

ຫນ້າທີ່ສົດໃສທີ່ສຸດແມ່ນກັບກາຟສະຖານະຫນ້າວຽກແລະເວລາປະຕິບັດຂອງພວກເຂົາ:

ພວກເຮົາໂຫຼດ underloaded

ດັ່ງ​ນັ້ນ​, ວຽກ​ງານ​ທັງ​ຫມົດ​ໄດ້​ຮັບ​ຜົນ​, ທ່ານ​ສາ​ມາດ​ປະ​ຕິ​ບັດ​ການ​ບາດ​ເຈັບ​ໄດ້​.

ແລະ​ມີ​ຜູ້​ບາດ​ເຈັບ​ຫຼາຍ​ຄົນ—ດ້ວຍ​ເຫດ​ຜົນ​ອັນ​ໜຶ່ງ​ຫຼື​ອັນ​ອື່ນ. ໃນກໍລະນີຂອງການນໍາໃຊ້ທີ່ຖືກຕ້ອງຂອງ Airflow, ສີ່ຫຼ່ຽມເຫຼົ່ານີ້ຊີ້ໃຫ້ເຫັນວ່າຂໍ້ມູນບໍ່ໄດ້ມາຮອດແນ່ນອນ.

ທ່ານຈໍາເປັນຕ້ອງເບິ່ງບັນທຶກແລະ restart ກໍລະນີທີ່ຫຼຸດລົງ.

ໂດຍ​ການ​ຄລິກ​ໃສ່​ສີ່​ຫລ່ຽມ​ໃດ​ຫນຶ່ງ​, ພວກ​ເຮົາ​ຈະ​ເຫັນ​ການ​ປະ​ຕິ​ບັດ​ທີ່​ມີ​ໃຫ້​ພວກ​ເຮົາ​:

ທ່ານ​ສາ​ມາດ​ໃຊ້​ເວ​ລາ​ແລະ​ເຮັດ​ໃຫ້​ລ້າງ​ການ​ຫຼຸດ​ລົງ​ໄດ້​. ນັ້ນແມ່ນ, ພວກເຮົາລືມວ່າບາງສິ່ງບາງຢ່າງລົ້ມເຫລວຢູ່ທີ່ນັ້ນ, ແລະວຽກງານຕົວຢ່າງດຽວກັນຈະໄປຫາຕົວກໍານົດເວລາ.

ມັນເປັນທີ່ຊັດເຈນວ່າການເຮັດແບບນີ້ກັບຫນູກັບສີ່ຫລ່ຽມສີແດງທັງຫມົດແມ່ນບໍ່ເປັນມະນຸດຫຼາຍ - ນີ້ບໍ່ແມ່ນສິ່ງທີ່ພວກເຮົາຄາດຫວັງຈາກ Airflow. ຕາມທໍາມະຊາດ, ພວກເຮົາມີອາວຸດທໍາລາຍມະຫາຊົນ: Browse/Task Instances

ມາເລືອກທຸກຢ່າງພ້ອມໆກັນ ແລະຣີເຊັດເປັນສູນ, ຄລິກລາຍການທີ່ຖືກຕ້ອງ:

ຫຼັງ​ຈາກ​ການ​ທໍາ​ຄວາມ​ສະ​ອາດ​, ລົດ​ແທັກ​ຊີ​ຂອງ​ພວກ​ເຮົາ​ມີ​ລັກ​ສະ​ນະ​ນີ້ (ພວກ​ເຂົາ​ເຈົ້າ​ໄດ້​ລໍ​ຖ້າ​ສໍາ​ລັບ​ການ​ຈັດ​ຕາ​ຕະ​ລາງ​ການ​ໃຫ້​ເຂົາ​ເຈົ້າ​)​:

ການເຊື່ອມຕໍ່, hooks ແລະຕົວແປອື່ນໆ

ມັນເປັນເວລາທີ່ຈະເບິ່ງ 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]

ທຸກຄົນເຄີຍເຮັດການອັບເດດລາຍງານບໍ? ນີ້ແມ່ນຂອງນາງອີກເທື່ອຫນຶ່ງ: ມີບັນຊີລາຍຊື່ຂອງແຫຼ່ງຂໍ້ມູນຈາກບ່ອນທີ່ໄດ້ຮັບຂໍ້ມູນ; ມີບັນຊີລາຍຊື່ທີ່ຈະເອົາ; ຢ່າລືມ honk ໃນເວລາທີ່ທຸກສິ່ງທຸກຢ່າງເກີດຂຶ້ນຫຼືແຕກ (ດີ, ນີ້ບໍ່ແມ່ນກ່ຽວກັບພວກເຮົາ, ບໍ່).

ໃຫ້ຜ່ານໄຟລ໌ອີກເທື່ອຫນຶ່ງແລະເບິ່ງສິ່ງທີ່ບໍ່ຊັດເຈນໃຫມ່:

• from commons.operators import TelegramBotSendMessage - ບໍ່ມີຫຍັງກີດຂວາງພວກເຮົາຈາກການເຮັດໃຫ້ຜູ້ປະກອບການຂອງພວກເຮົາເອງ, ເຊິ່ງພວກເຮົາໄດ້ໃຊ້ປະໂຫຍດຈາກການສ້າງ wrapper ຂະຫນາດນ້ອຍສໍາລັບການສົ່ງຂໍ້ຄວາມໄປຫາ Unblocked. (ພວກເຮົາຈະສົນທະນາເພີ່ມເຕີມກ່ຽວກັບຜູ້ປະກອບການນີ້ຂ້າງລຸ່ມນີ້);
• default_args={} - dag ສາມາດແຈກຢາຍການໂຕ້ຖຽງດຽວກັນກັບຜູ້ປະຕິບັດການທັງຫມົດຂອງມັນ;
• to='{{ var.value.all_the_kings_men }}' - ພາກສະໜາມ to ພວກເຮົາຈະບໍ່ມີ hardcoded, ແຕ່ສ້າງແບບເຄື່ອນໄຫວໂດຍໃຊ້ 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 ມາ​ຮ່ວມ​ກັນ​ໃນ​ການ​ສົ່ງ​ຈົດ​ຫມາຍ​ແລະ​ຂໍ້​ຄວາມ​, ເຊັ່ນ​ນີ້​:
  

  ແຕ່ເນື່ອງຈາກຜູ້ປະຕິບັດການແຈ້ງການມີເງື່ອນໄຂການເປີດຕົວທີ່ແຕກຕ່າງກັນ, ພຽງແຕ່ຫນຶ່ງຈະເຮັດວຽກ. ໃນ Tree View, ທຸກສິ່ງທຸກຢ່າງເບິ່ງຫນ້ອຍລົງເລັກນ້ອຍ:
  

ຂ້ອຍຈະເວົ້າສອງສາມຄໍາກ່ຽວກັບ ມະຫາພາກ ແລະ​ຫມູ່​ເພື່ອນ​ຂອງ​ເຂົາ​ເຈົ້າ - ຕົວແປ.

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. ສ່ວນທີ່ດີທີ່ສຸດແມ່ນວ່າຕົວແປບໍລິບົດຖືກຕິດໃສ່ກັບຕົວຢ່າງວຽກສະເພາະ (ສີ່ຫຼ່ຽມໃນ Tree View), ແລະເມື່ອເລີ່ມຕົ້ນໃຫມ່, ຕົວຍຶດຈະຂະຫຍາຍໄປສູ່ຄ່າດຽວກັນ.

ຄ່າທີ່ໄດ້ຮັບມອບໝາຍສາມາດເບິ່ງໄດ້ໂດຍໃຊ້ປຸ່ມ Rendered ໃນແຕ່ລະໜ້າວຽກ. ນີ້ແມ່ນວິທີການສົ່ງຈົດຫມາຍ:

ແລະດັ່ງນັ້ນຢູ່ໃນວຽກງານທີ່ມີການສົ່ງຂໍ້ຄວາມ:

ບັນຊີລາຍຊື່ຄົບຖ້ວນຂອງ macro ທີ່ມີໃນຕົວສໍາລັບຮຸ່ນຫຼ້າສຸດທີ່ມີຢູ່ແມ່ນຢູ່ທີ່ນີ້: ການອ້າງອີງມະຫາພາກ

ຍິ່ງໄປກວ່ານັ້ນ, ດ້ວຍການຊ່ວຍເຫຼືອຂອງ plugins, ພວກເຮົາສາມາດປະກາດ macro ຂອງພວກເຮົາເອງ, ແຕ່ນັ້ນແມ່ນເລື່ອງອື່ນ.

ນອກເຫນືອຈາກສິ່ງທີ່ກໍານົດໄວ້ລ່ວງຫນ້າ, ພວກເຮົາສາມາດທົດແທນຄ່າຂອງຕົວແປຂອງພວກເຮົາ (ຂ້ອຍໄດ້ໃຊ້ມັນຢູ່ໃນລະຫັດຂ້າງເທິງ). ມາສ້າງໃນ Admin/Variables ສອງສາມຢ່າງ:

ທຸກ​ສິ່ງ​ທຸກ​ຢ່າງ​ທີ່​ທ່ານ​ສາ​ມາດ​ນໍາ​ໃຊ້​:

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

ຄ່າສາມາດເປັນ Scalar, ຫຼືມັນສາມາດເປັນ JSON. ໃນກໍລະນີຂອງ JSON:

bot_config

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

ພຽງແຕ່ໃຊ້ເສັ້ນທາງໄປຫາກະແຈທີ່ຕ້ອງການ: {{ var.json.bot_config.bot.token }}.

ຂ້າ​ພະ​ເຈົ້າ​ຮູ້​ຫນັງ​ສື​ຈະ​ເວົ້າ​ຫນຶ່ງ​ຄໍາ​ແລະ​ສະ​ແດງ​ໃຫ້​ເຫັນ screenshot ຫນຶ່ງ​ກ່ຽວ​ກັບ​ການ​ ການເຊື່ອມຕໍ່. ທຸກສິ່ງທຸກຢ່າງແມ່ນພື້ນຖານຢູ່ທີ່ນີ້: ໃນຫນ້າ Admin/Connections ພວກເຮົາສ້າງການເຊື່ອມຕໍ່, ເພີ່ມການເຂົ້າສູ່ລະບົບ / ລະຫັດຜ່ານຂອງພວກເຮົາແລະຕົວກໍານົດການສະເພາະຫຼາຍຢູ່ທີ່ນັ້ນ. ແບບນີ້:

ລະຫັດຜ່ານສາມາດຖືກເຂົ້າລະຫັດ (ຢ່າງລະອຽດກວ່າຄ່າເລີ່ມຕົ້ນ), ຫຼືທ່ານສາມາດອອກຈາກປະເພດການເຊື່ອມຕໍ່ (ດັ່ງທີ່ຂ້ອຍໄດ້ເຮັດສໍາລັບ tg_main) - ຄວາມຈິງແມ່ນວ່າບັນຊີລາຍຊື່ຂອງປະເພດແມ່ນ hardwired ໃນຮູບແບບ Airflow ແລະບໍ່ສາມາດຂະຫຍາຍໄດ້ໂດຍບໍ່ມີການເຂົ້າໄປໃນລະຫັດແຫຼ່ງ (ຖ້າຫາກວ່າທັນທີທັນໃດຂ້າພະເຈົ້າບໍ່ໄດ້ google ບາງສິ່ງບາງຢ່າງ, ກະລຸນາແກ້ໄຂຂ້າພະເຈົ້າ), ແຕ່ບໍ່ມີຫຍັງຈະຢຸດພວກເຮົາຈາກການໄດ້ຮັບສິນເຊື່ອພຽງແຕ່ໂດຍການ. ຊື່.

ນອກນັ້ນທ່ານຍັງສາມາດເຮັດການເຊື່ອມຕໍ່ຈໍານວນຫນຶ່ງທີ່ມີຊື່ດຽວກັນ: ໃນກໍລະນີນີ້, ວິທີການ BaseHook.get_connection(), ເຊິ່ງເຮັດໃຫ້ພວກເຮົາເຊື່ອມຕໍ່ໂດຍຊື່, ຈະໃຫ້ ສຸ່ມ ຈາກຫຼາຍໆ namesakes (ມັນຈະມີເຫດຜົນຫຼາຍກວ່າທີ່ຈະເຮັດໃຫ້ Round Robin, ແຕ່ໃຫ້ມັນຢູ່ໃນຈິດໃຈຂອງຜູ້ພັດທະນາ Airflow).

ຕົວແປແລະການເຊື່ອມຕໍ່ແນ່ນອນແມ່ນເຄື່ອງມືທີ່ເຢັນ, ແຕ່ມັນສໍາຄັນທີ່ຈະບໍ່ສູນເສຍຄວາມສົມດູນ: ສ່ວນໃດຂອງການໄຫຼຂອງເຈົ້າທີ່ທ່ານເກັບໄວ້ໃນລະຫັດຕົວມັນເອງ, ແລະສ່ວນໃດທີ່ທ່ານໃຫ້ Airflow ສໍາລັບການເກັບຮັກສາ. ໃນດ້ານຫນຶ່ງ, ມັນສາມາດສະດວກໃນການປ່ຽນແປງມູນຄ່າຢ່າງໄວວາ, ຕົວຢ່າງ, ກ່ອງຈົດຫມາຍ, ຜ່ານ UI. ໃນທາງກົງກັນຂ້າມ, ນີ້ຍັງເປັນການກັບຄືນໄປຫາການຄລິກຫນູ, ຈາກທີ່ພວກເຮົາ (ຂ້ອຍ) ຕ້ອງການກໍາຈັດ.

ການເຮັດວຽກກັບການເຊື່ອມຕໍ່ແມ່ນຫນຶ່ງໃນວຽກງານ hooks. ໂດຍທົ່ວໄປ, Airflow hooks ແມ່ນຈຸດສໍາລັບການເຊື່ອມຕໍ່ມັນກັບການບໍລິການພາກສ່ວນທີສາມແລະຫ້ອງສະຫມຸດ. ຕົວຢ່າງ: 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, ເຊິ່ງປະຕິບັດບາງສິ່ງທີ່ສະເພາະ Airflow (ເບິ່ງການພັກຜ່ອນຂອງທ່ານ)
• ທົ່ງນາທີ່ປະກາດ template_fields, ໃນທີ່ Jinja ຈະຊອກຫາມະຫາພາກເພື່ອປະມວນຜົນ.
• ຈັດແຈງການໂຕ້ຖຽງທີ່ຖືກຕ້ອງສໍາລັບ __init__(), ຕັ້ງຄ່າເລີ່ມຕົ້ນໃນບ່ອນທີ່ມີຄວາມຈໍາເປັນ.
• ພວກເຮົາບໍ່ໄດ້ລືມກ່ຽວກັບການເລີ່ມຕົ້ນຂອງບັນພະບຸລຸດ.
• ເປີດ hook ທີ່ສອດຄ້ອງກັນ TelegramBotHookໄດ້ຮັບວັດຖຸຂອງລູກຄ້າຈາກມັນ.
• ວິທີການ overridden (ກໍານົດຄືນໃຫມ່). BaseOperator.execute(), ທີ່ Airfow ຈະ twitch ເມື່ອເວລາທີ່ຈະເປີດຕົວຜູ້ປະຕິບັດງານ - ໃນມັນພວກເຮົາຈະປະຕິບັດການປະຕິບັດຕົ້ນຕໍ, ລືມເຂົ້າສູ່ລະບົບ. (ພວກເຮົາເຂົ້າສູ່ລະບົບ, ໂດຍວິທີທາງການ, ເຂົ້າໄປ stdout и stderr - ການ​ໄຫຼ​ວຽນ​ຂອງ​ອາ​ກາດ​ຈະ​ສະ​ກັດ​ທຸກ​ສິ່ງ​ທຸກ​ຢ່າງ​, ຫໍ່​ມັນ​ງາມ​, decompose ມັນ​ໃນ​ທີ່​ຈໍາ​ເປັນ​.

ໃຫ້ເບິ່ງສິ່ງທີ່ພວກເຮົາມີ commons/hooks.py. ສ່ວນທໍາອິດຂອງໄຟລ໌, ກັບ hook ຕົວຂອງມັນເອງ:

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;
• overriding ວິທີການມາດຕະຖານ: ຂ້ອຍຈໍາກັດຕົວເອງ get_conn(), ໃນທີ່ຂ້ອຍໄດ້ຮັບຕົວກໍານົດການເຊື່ອມຕໍ່ໂດຍຊື່ແລະພຽງແຕ່ໄດ້ຮັບສ່ວນ extra (ນີ້ແມ່ນຊ່ອງ JSON), ເຊິ່ງຂ້ອຍ (ຕາມຄໍາແນະນໍາຂອງຂ້ອຍເອງ!) ໃສ່ Telegram bot token: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• ຂ້ອຍສ້າງຕົວຢ່າງຂອງພວກເຮົາ TelegramBot, ໃຫ້ມັນເປັນ token ສະເພາະ.

ຫມົດ​ເທົ່າ​ນີ້. ທ່ານສາມາດໄດ້ຮັບລູກຄ້າຈາກ hook ໂດຍໃຊ້ 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 - ໃນ plugin, ໃສ່ໃນ repository ສາທາລະນະ, ແລະໃຫ້ມັນກັບ Open Source.

ໃນຂະນະທີ່ພວກເຮົາສຶກສາທັງຫມົດນີ້, ການປັບປຸງບົດລາຍງານຂອງພວກເຮົາໄດ້ລົ້ມເຫລວຢ່າງສໍາເລັດຜົນແລະສົ່ງຂໍ້ຄວາມຂໍ້ຜິດພາດໃຫ້ຂ້ອຍໃນຊ່ອງທາງ. ຂ້ອຍ​ຈະ​ກວດ​ເບິ່ງ​ວ່າ​ຜິດ​ບໍ...

ບາງສິ່ງບາງຢ່າງແຕກຢູ່ໃນ doge ຂອງພວກເຮົາ! ມັນບໍ່ແມ່ນສິ່ງທີ່ພວກເຮົາຄາດຫວັງ? ຢ່າງ​ແນ່​ນອນ!

ເຈົ້າຈະຖອກບໍ?

ເຈົ້າຮູ້ສຶກວ່າຂ້ອຍພາດບາງສິ່ງບາງຢ່າງບໍ? ມັນເບິ່ງຄືວ່າລາວສັນຍາວ່າຈະໂອນຂໍ້ມູນຈາກ 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

ສິ່ງທີ່ມະຫັດສະຈັນ randomizer ຂອງພວກເຮົາສ້າງ, ທ່ານສາມາດນໍາໃຊ້ລາຍການ Data Profiling/Ad Hoc Query:

ສິ່ງທີ່ສໍາຄັນແມ່ນບໍ່ສະແດງໃຫ້ນັກວິເຄາະ

ລະອຽດກ່ຽວກັບ ເຊດຊັນ ETL ຂ້າພະເຈົ້າຈະບໍ່, ທຸກສິ່ງທຸກຢ່າງແມ່ນເລື່ອງເລັກໆນ້ອຍໆຢູ່ທີ່ນັ້ນ: ພວກເຮົາສ້າງພື້ນຖານ, ມີເຄື່ອງຫມາຍໃນມັນ, ພວກເຮົາຫໍ່ທຸກສິ່ງທຸກຢ່າງດ້ວຍຕົວຈັດການສະພາບການ, ແລະຕອນນີ້ພວກເຮົາເຮັດສິ່ງນີ້:

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

    # Load worflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

from sys import stderr

class Session:
    """ETL workflow session

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

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

        self._task_name = task_name
        self._id = None

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

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

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

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

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

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

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

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

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

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

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

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

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

ເວລາມາຮອດແລ້ວ ເກັບກໍາຂໍ້ມູນຂອງພວກເຮົາ ຈາກຫນຶ່ງແລະເຄິ່ງຫນຶ່ງຮ້ອຍຕາຕະລາງຂອງພວກເຮົາ. ໃຫ້ເຮັດສິ່ງນີ້ດ້ວຍການຊ່ວຍເຫຼືອຂອງສາຍ unpretentious ຫຼາຍ:

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. ດ້ວຍການຊ່ວຍເຫຼືອຂອງ hook ທີ່ພວກເຮົາໄດ້ຮັບຈາກ 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']])

ຄື:

• ຖານຂໍ້ມູນທີ່ພວກເຮົາເອົາຄໍາສັ່ງ,
• ID ຂອງກອງປະຊຸມນໍ້າຖ້ວມຂອງພວກເຮົາ (ມັນຈະແຕກຕ່າງກັນ ສໍາລັບທຸກໆວຽກງານ),
• A hash ຈາກແຫຼ່ງແລະຄໍາສັ່ງ ID - ດັ່ງນັ້ນໃນຖານຂໍ້ມູນສຸດທ້າຍ (ບ່ອນທີ່ທຸກສິ່ງທຸກຢ່າງແມ່ນ poured ເຂົ້າໄປໃນຕາຕະລາງຫນຶ່ງ) ພວກເຮົາມີ ID ຄໍາສັ່ງເປັນເອກະລັກ.

ຂັ້ນຕອນສຸດທ້າຍຍັງຄົງຢູ່: ເອົາທຸກສິ່ງທຸກຢ່າງເຂົ້າໄປໃນ Vertica. ແລະ, oddly ພຽງພໍ, ຫນຶ່ງໃນວິທີທີ່ຫນ້າປະທັບໃຈທີ່ສຸດແລະປະສິດທິພາບທີ່ຈະເຮັດນີ້ແມ່ນຜ່ານ 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 ທີ່ພວກເຮົາມັກດ້ວຍ hook.
4. ແລະໃນປັດຈຸບັນດ້ວຍການຊ່ວຍເຫຼືອ copy() ສົ່ງຂໍ້ມູນຂອງພວກເຮົາໂດຍກົງຫາ Vertika!

ພວກ​ເຮົາ​ຈະ​ເອົາ​ຈາກ​ຄົນ​ຂັບ​ວ່າ​ມີ​ແຖວ​ໃດ​ທີ່​ເຕັມ​ໄປ​, ແລະ​ບອກ​ຜູ້​ຈັດ​ການ​ກອງ​ປະ​ຊຸມ​ວ່າ​ທຸກ​ສິ່ງ​ທຸກ​ຢ່າງ​ແມ່ນ OK​:

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() ຂ້ອຍສ້າງ schema ຖານຂໍ້ມູນແລະຕາຕະລາງ (ຖ້າພວກເຂົາບໍ່ມີຢູ່ແລ້ວ, ແນ່ນອນ). ສິ່ງທີ່ ສຳ ຄັນແມ່ນການຈັດແຈງການເພິ່ງພາອາໄສຢ່າງຖືກຕ້ອງ:

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

Summing up

- ດີ, - ຫນູນ້ອຍເວົ້າວ່າ, - ບໍ່ແມ່ນມັນ, ໃນປັດຈຸບັນ
ເຈົ້າໝັ້ນໃຈບໍວ່າຂ້ອຍເປັນສັດທີ່ຂີ້ຮ້າຍທີ່ສຸດໃນປ່າ?

Julia Donaldson, The Gruffalo

ຂ້ອຍຄິດວ່າຖ້າເພື່ອນຮ່ວມງານຂອງຂ້ອຍແລະຂ້ອຍມີການແຂ່ງຂັນ: ຜູ້ທີ່ຈະສ້າງແລະເປີດຕົວຂະບວນການ ETL ຢ່າງໄວວາຈາກຈຸດເລີ່ມຕົ້ນ: ພວກເຂົາກັບ SSIS ແລະຫນູແລະຂ້ອຍກັບ Airflow ... ແລະຫຼັງຈາກນັ້ນພວກເຮົາຍັງຈະປຽບທຽບຄວາມງ່າຍຂອງການບໍາລຸງຮັກສາ ... ວ້າວ, ຂ້ອຍຄິດວ່າເຈົ້າຈະເຫັນດີວ່າຂ້ອຍຈະຕີພວກເຂົາໃນທຸກດ້ານ!

ຖ້າຫາກວ່າເລັກນ້ອຍຫຼາຍ, ຫຼັງຈາກນັ້ນ Apache Airflow - ໂດຍການອະທິບາຍຂະບວນການໃນຮູບແບບຂອງລະຫັດໂຄງການ - ໄດ້ເຮັດວຽກຂອງຂ້ອຍ. ຫຼາຍ ສະດວກສະບາຍ ແລະມີຄວາມສຸກຫຼາຍຂຶ້ນ.

ການຂະຫຍາຍທີ່ບໍ່ຈໍາກັດຂອງມັນ, ທັງໃນແງ່ຂອງ plug-ins ແລະ predisposition ກັບ scalability, ເຮັດໃຫ້ທ່ານມີໂອກາດທີ່ຈະນໍາໃຊ້ Airflow ໃນເກືອບທຸກພື້ນທີ່: ເຖິງແມ່ນວ່າໃນວົງຈອນເຕັມຂອງການເກັບກໍາ, ການກະກຽມແລະການປະມວນຜົນຂໍ້ມູນ, ເຖິງແມ່ນວ່າໃນການເປີດຕົວຈະລວດ (ໄປ Mars, ຂອງ. ຫຼັກສູດ).

ສ່ວນສຸດທ້າຍ, ຂໍ້ມູນອ້າງອີງ ແລະຂໍ້ມູນ

rake ທີ່​ພວກ​ເຮົາ​ໄດ້​ເກັບ​ກໍາ​ສໍາ​ລັບ​ທ່ານ​

• start_date. ແມ່ນແລ້ວ, ນີ້ແມ່ນ meme ທ້ອງຖິ່ນແລ້ວ. ການໂຕ້ຖຽງຫຼັກຂອງ Via Doug start_date ຜ່ານທັງໝົດ. ໂດຍຫຍໍ້, ຖ້າທ່ານລະບຸໃນ start_date ວັນທີປະຈຸບັນ, ແລະ schedule_interval - ມື້ຫນຶ່ງ, ຫຼັງຈາກນັ້ນ DAG ຈະເລີ່ມຕົ້ນໃນມື້ອື່ນບໍ່ມີກ່ອນຫນ້ານີ້.

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

  ແລະບໍ່ມີບັນຫາຫຼາຍ.

  ມີຄວາມຜິດພາດ runtime ອື່ນທີ່ກ່ຽວຂ້ອງກັບມັນ: Task is missing the start_date parameter, ເຊິ່ງສ່ວນຫຼາຍມັກຈະຊີ້ໃຫ້ເຫັນວ່າເຈົ້າລືມຜູກມັດກັບຕົວປະຕິບັດການ dag.

• ທັງໝົດໃນເຄື່ອງດຽວ. ແມ່ນແລ້ວ, ແລະພື້ນຖານ (Airflow ຕົວຂອງມັນເອງແລະການເຄືອບຂອງພວກເຮົາ), ແລະເຄື່ອງແມ່ຂ່າຍເວັບໄຊຕ໌, ແລະຕາຕະລາງ, ແລະພະນັກງານ. ແລະເຖິງແມ່ນວ່າມັນເຮັດວຽກ. ແຕ່ໃນໄລຍະເວລາ, ຈໍານວນຫນ້າວຽກສໍາລັບການບໍລິການເພີ່ມຂຶ້ນ, ແລະເມື່ອ PostgreSQL ເລີ່ມຕອບສະຫນອງກັບດັດສະນີໃນ 20 s ແທນທີ່ຈະເປັນ 5 ms, ພວກເຮົາເອົາມັນໄປ.
• LocalExecutor. ແມ່ນແລ້ວ, ພວກເຮົາຍັງນັ່ງຢູ່ເທິງມັນ, ແລະພວກເຮົາໄດ້ມາຮອດຂອບຂອງເຫວເລິກແລ້ວ. LocalExecutor ໄດ້ພຽງພໍສໍາລັບພວກເຮົາຈົນເຖິງປະຈຸບັນ, ແຕ່ຕອນນີ້ມັນເຖິງເວລາທີ່ຈະຂະຫຍາຍກັບພະນັກງານຢ່າງຫນ້ອຍຫນຶ່ງຄົນ, ແລະພວກເຮົາຈະຕ້ອງເຮັດວຽກຫນັກເພື່ອຍ້າຍໄປ CeleryExecutor. ແລະໃນທັດສະນະຂອງຄວາມຈິງທີ່ວ່າທ່ານສາມາດເຮັດວຽກກັບມັນຢູ່ໃນເຄື່ອງດຽວ, ບໍ່ມີຫຍັງຢຸດເຈົ້າຈາກການໃຊ້ Celery ເຖິງແມ່ນວ່າຢູ່ໃນເຄື່ອງແມ່ຂ່າຍ, ເຊິ່ງ "ແນ່ນອນ, ຈະບໍ່ເຂົ້າໄປໃນການຜະລິດ, ດ້ວຍຄວາມຊື່ສັດ!"
• ບໍ່ໃຊ້ ເຄື່ອງ​ມື​ໃນ​ຕົວ​:
  • ການເຊື່ອມຕໍ່ ເພື່ອ​ເກັບ​ຮັກ​ສາ​ໃບ​ຢັ້ງ​ຢືນ​ການ​ບໍ​ລິ​ການ​,
  • SLA Misses ຕອບ​ສະ​ຫນອງ​ວຽກ​ງານ​ທີ່​ບໍ່​ໄດ້​ທັນ​ການ​,
  • xcom ສໍາລັບການແລກປ່ຽນ metadata (ຂ້າພະເຈົ້າເວົ້າ ເມຕາຂໍ້ມູນ!) ລະຫວ່າງວຽກງານ dag.
• ການລ່ວງລະເມີດທາງໄປສະນີ. ດີ, ຂ້ອຍສາມາດເວົ້າຫຍັງໄດ້? ການເຕືອນໄດ້ຖືກສ້າງຕັ້ງຂຶ້ນສໍາລັບການຄ້າງຄືນທັງຫມົດຂອງວຽກງານທີ່ຫຼຸດລົງ. ດຽວນີ້ Gmail ວຽກຂອງຂ້ອຍມີ > 90k ອີເມວຈາກ Airflow, ແລະ web mail muzzle ປະຕິເສດທີ່ຈະເອົາແລະລຶບຫຼາຍກວ່າ 100 ສະບັບຕໍ່ຄັ້ງ.

ຂຸມ​ຫຼາຍ​: Apache Airflow Pitfails

ເຄື່ອງມືອັດຕະໂນມັດເພີ່ມເຕີມ

ເພື່ອໃຫ້ພວກເຮົາເຮັດວຽກຫຼາຍຂຶ້ນດ້ວຍຫົວຂອງພວກເຮົາແລະບໍ່ແມ່ນດ້ວຍມືຂອງພວກເຮົາ, Airflow ໄດ້ກະກຽມສໍາລັບພວກເຮົານີ້:

• REST API - ລາວຍັງມີສະຖານະຂອງ Experimental, ເຊິ່ງບໍ່ໄດ້ປ້ອງກັນບໍ່ໃຫ້ລາວເຮັດວຽກ. ດ້ວຍມັນ, ທ່ານບໍ່ພຽງແຕ່ສາມາດໄດ້ຮັບຂໍ້ມູນກ່ຽວກັບ dags ແລະວຽກງານ, ແຕ່ຍັງຢຸດ / ເລີ່ມຕົ້ນ dag, ສ້າງ DAG Run ຫຼືສະນຸກເກີ.
• CLI - ເຄື່ອງມືຈໍານວນຫຼາຍສາມາດໃຊ້ໄດ້ຜ່ານເສັ້ນຄໍາສັ່ງທີ່ບໍ່ພຽງແຕ່ບໍ່ສະດວກໃນການນໍາໃຊ້ຜ່ານ WebUI, ແຕ່ໂດຍທົ່ວໄປແລ້ວແມ່ນບໍ່ມີ. ຍົກ​ຕົວ​ຢ່າງ:
  • backfill ຕ້ອງການເພື່ອປິດເປີດໜ້າວຽກຄືນໃໝ່.
    ຕົວຢ່າງ, ນັກວິເຄາະມາແລະເວົ້າວ່າ: "ແລະທ່ານ, comrade, ມີຄວາມບໍ່ມີເຫດຜົນໃນຂໍ້ມູນຈາກວັນທີ 1 ຫາ 13 ມັງກອນ! ແກ້ໄຂ, ແກ້ໄຂ, ແກ້ໄຂ, ແກ້ໄຂມັນ!" ແລະທ່ານເປັນ hob ດັ່ງກ່າວ:

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

  • ບໍລິການພື້ນຖານ: initdb, resetdb, upgradedb, checkdb.
  • run, ເຊິ່ງອະນຸຍາດໃຫ້ທ່ານສາມາດດໍາເນີນວຽກງານຕົວຢ່າງຫນຶ່ງ, ແລະແມ້ກະທັ້ງຄະແນນກ່ຽວກັບການຂຶ້ນກັບທັງຫມົດ. ຍິ່ງໄປກວ່ານັ້ນ, ທ່ານສາມາດດໍາເນີນການຜ່ານ LocalExecutor, ເຖິງແມ່ນວ່າທ່ານມີກຸ່ມ Celery.
  • ເຮັດສິ່ງດຽວກັນຫຼາຍ test, ພຽງແຕ່ຍັງຢູ່ໃນຖານຂຽນບໍ່ມີຫຍັງ.
  • connections ອະນຸຍາດໃຫ້ການສ້າງມະຫາຊົນຂອງການເຊື່ອມຕໍ່ຈາກແກະ.
• python api - ເປັນວິທີທີ່ຍາກຫຼາຍຂອງການໂຕ້ຕອບ, ເຊິ່ງມີຈຸດປະສົງສໍາລັບ plugins, ແລະບໍ່ swarming ໃນມັນດ້ວຍມືພຽງເລັກນ້ອຍ. ແຕ່​ໃຜ​ຈະ​ຢຸດ​ພວກ​ເຮົາ​ຈາກ​ການ​ໄປ /home/airflow/dags, ແລ່ນ ipython ແລະເລີ່ມ messing ປະມານ? ທ່ານສາມາດ, ສໍາລັບການຍົກຕົວຢ່າງ, ສົ່ງອອກການເຊື່ອມຕໍ່ທັງຫມົດທີ່ມີລະຫັດດັ່ງຕໍ່ໄປນີ້:

  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. ຂ້ອຍບໍ່ແນະນໍາໃຫ້ຂຽນມັນ, ແຕ່ການໄດ້ຮັບລັດວຽກສໍາລັບ metrics ສະເພາະຕ່າງໆສາມາດໄວແລະງ່າຍກວ່າຜ່ານ 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 ຈາກ bookmarks ຂອງຂ້ອຍ.

• ເອກະສານ Apache Airflow - ແນ່ນອນ, ພວກເຮົາຕ້ອງເລີ່ມຕົ້ນດ້ວຍຫ້ອງການ. ເອກະສານ, ແຕ່ວ່າໃຜອ່ານຄໍາແນະນໍາ?
• ການ​ປະ​ຕິ​ບັດ​ທີ່​ດີ​ທີ່​ສຸດ - ດີ, ຢ່າງຫນ້ອຍອ່ານຄໍາແນະນໍາຈາກຜູ້ສ້າງ.
• UI Airflow - ຈຸດ​ເລີ່ມ​ຕົ້ນ​: ການ​ໂຕ້​ຕອບ​ຜູ້​ໃຊ້​ໃນ​ຮູບ​ພາບ​
• ຄວາມເຂົ້າໃຈແນວຄວາມຄິດທີ່ສໍາຄັນຂອງ Apache Airflow - ແນວຄວາມຄິດພື້ນຖານໄດ້ຖືກອະທິບາຍໄດ້ດີ, ຖ້າ (ທັນທີທັນໃດ!) ເຈົ້າບໍ່ເຂົ້າໃຈບາງສິ່ງບາງຢ່າງຈາກຂ້ອຍ.
• Blog ຂອງ Tianlong — ຄູ່ມືກ່ຽວກັບວິທີການສ້າງ Airflow Server/Cluster - ຄູ່ມືສັ້ນໆສໍາລັບການຕັ້ງກຸ່ມ Airflow.
• ແລ່ນ Apache Airflow ຢູ່ Lyft - ເກືອບບົດຄວາມທີ່ຫນ້າສົນໃຈດຽວກັນ, ຍົກເວັ້ນບາງທີອາດເປັນທາງການຫຼາຍ, ແລະຕົວຢ່າງຫນ້ອຍ.
• ວິທີການ Apache Airflow ແຈກຢາຍວຽກກ່ຽວກັບພະນັກງານ Celery — ກ່ຽວ​ກັບ​ການ​ເຮັດ​ວຽກ​ຮ່ວມ​ກັບ Celery​.
• DAG ຂຽນການປະຕິບັດທີ່ດີທີ່ສຸດໃນ Apache Airflow - ກ່ຽວກັບ ideempotency ຂອງວຽກງານ, ການໂຫຼດໂດຍ ID ແທນທີ່ຈະເປັນວັນທີ, ການຫັນເປັນ, ໂຄງສ້າງໄຟລ໌ແລະສິ່ງທີ່ຫນ້າສົນໃຈອື່ນໆ.
• ການຄຸ້ມຄອງການເພິ່ງພາອາໄສໃນ Apache Airflow - ຄວາມເພິ່ງພາອາໄສຂອງວຽກງານແລະກົດລະບຽບຜົນກະທົບຕໍ່, ທີ່ຂ້າພະເຈົ້າໄດ້ກ່າວເຖິງພຽງແຕ່ໃນການຖ່າຍທອດ.
• Airflow: ເມື່ອ DAG ຂອງເຈົ້າຢູ່ໄກຫລັງຕາຕະລາງ - ວິທີການເອົາຊະນະບາງ "ເຮັດວຽກຕາມຈຸດປະສົງ" ໃນຕົວກໍານົດເວລາ, ໂຫຼດຂໍ້ມູນສູນເສຍແລະຈັດລໍາດັບຄວາມສໍາຄັນຂອງວຽກງານ.
• ຄໍາຖາມ SQL ທີ່ເປັນປະໂຫຍດສໍາລັບ Apache Airflow — ຄໍາຖາມ SQL ທີ່ເປັນປະໂຫຍດຕໍ່ metadata ຂອງ Airflow.
• ເລີ່ມຕົ້ນພັດທະນາຂັ້ນຕອນການເຮັດວຽກດ້ວຍ Apache Airflow - ມີສ່ວນທີ່ເປັນປະໂຫຍດກ່ຽວກັບການສ້າງເຊັນເຊີທີ່ກໍາຫນົດເອງ.
• ການສ້າງ Fetchr Data Science Infra ເທິງ AWS ດ້ວຍ Presto ແລະ Airflow — ຂໍ້ສັງເກດສັ້ນໆທີ່ໜ້າສົນໃຈກ່ຽວກັບການສ້າງໂຄງສ້າງພື້ນຖານໃນ AWS ສໍາລັບວິທະຍາສາດຂໍ້ມູນ.
• 7 ຄວາມຜິດພາດທົ່ວໄປທີ່ຈະກວດສອບໃນເວລາທີ່ Debugging Airflow DAGs - ຄວາມຜິດພາດທົ່ວໄປ (ໃນເວລາທີ່ຜູ້ໃດຜູ້ຫນຶ່ງຍັງບໍ່ໄດ້ອ່ານຄໍາແນະນໍາ).
• ເກັບຮັກສາແລະເຂົ້າເຖິງລະຫັດຜ່ານໂດຍໃຊ້ Apache Airflow - ຍິ້ມວິທີທີ່ຄົນເຮົາເກັບລະຫັດຜ່ານ, ເຖິງແມ່ນວ່າທ່ານພຽງແຕ່ສາມາດໃຊ້ການເຊື່ອມຕໍ່.
• Zen ຂອງ Python ແລະ Apache Airflow - ການສົ່ງຕໍ່ DAG implicit, context throwing ໃນຫນ້າທີ່, ອີກເທື່ອຫນຶ່ງກ່ຽວກັບການຂຶ້ນກັບ, ແລະຍັງກ່ຽວກັບການຂ້າມການເປີດຕົວວຽກງານ.
• Airflow: ຄໍາແນະນໍາທີ່ຮູ້ຈັກຫນ້ອຍ, ເຄັດລັບ, ແລະການປະຕິບັດທີ່ດີທີ່ສຸດ - ກ່ຽວ​ກັບ​ການ​ນໍາ​ໃຊ້​ default arguments и params ໃນແມ່ແບບ, ເຊັ່ນດຽວກັນກັບຕົວແປແລະການເຊື່ອມຕໍ່.
• ການສ້າງຂໍ້ມູນຕາຕະລາງການໄຫຼວຽນຂອງອາກາດ - ເລື່ອງກ່ຽວກັບວິທີທີ່ຜູ້ວາງແຜນກໍາລັງກະກຽມສໍາລັບ Airflow 2.0.
• Apache Airflow ກັບ 3 ຄົນງານ Celery ໃນ docker-compose - ບົດຄວາມທີ່ລ້າສະໄຫມເລັກນ້ອຍກ່ຽວກັບການນໍາກຸ່ມຂອງພວກເຮົາເຂົ້າໃນ docker-compose.
• 4 Templating Tasks ໂດຍໃຊ້ Airflow Context - ວຽກ​ງານ​ເຄື່ອນ​ໄຫວ​ໂດຍ​ນໍາ​ໃຊ້​ແມ່​ແບບ​ແລະ​ການ​ສົ່ງ​ຕໍ່​ສະ​ພາບ​ການ​.
• ການແຈ້ງເຕືອນຄວາມຜິດພາດໃນ Airflow — ການ​ແຈ້ງ​ການ​ມາດ​ຕະ​ຖານ​ແລະ​ກໍາ​ນົດ​ເອງ​ໂດຍ mail ແລະ Slack​.
• Airflow Workshop: DAGs ສະລັບສັບຊ້ອນໂດຍບໍ່ມີໄມ້ຄ້ອນເທົ້າ - ສາຂາວຽກ, ມະຫາພາກ ແລະ XCom.

ແລະການເຊື່ອມຕໍ່ທີ່ໃຊ້ໃນບົດຄວາມ:

• ການອ້າງອີງມະຫາພາກ - placeholder ມີ​ສໍາ​ລັບ​ການ​ນໍາ​ໃຊ້​ໃນ​ແມ່​ແບບ​.
• ຂຸມ​ທົ່ວ​ໄປ—ການ​ໄຫຼ​ຂອງ​ອາ​ກາດ - ຄວາມຜິດພາດທົ່ວໄປໃນເວລາສ້າງ dags.
• puckel/docker-airflow: Docker Apache Airflow - docker-compose ສໍາລັບການທົດລອງ, debugging ແລະອື່ນໆ.
• python-telegram-bot/python-telegram-bot: ພວກເຮົາໄດ້ຕັ້ງໃຫ້ທ່ານເປັນ wrapper ທີ່ທ່ານບໍ່ສາມາດປະຕິເສດໄດ້. - Python wrapper ສໍາລັບ Telegram REST API.

ແຫຼ່ງຂໍ້ມູນ: www.habr.com