അപ്പാച്ചെ എയർഫ്ലോ: ETL എളുപ്പമാക്കുന്നു

ഹായ്, ഞാൻ ദിമിത്രി ലോഗ്‌വിനെങ്കോ - വെസെറ്റ് ഗ്രൂപ്പ് ഓഫ് കമ്പനികളുടെ അനലിറ്റിക്‌സ് ഡിപ്പാർട്ട്‌മെന്റിന്റെ ഡാറ്റാ എഞ്ചിനീയർ.

ETL പ്രക്രിയകൾ വികസിപ്പിക്കുന്നതിനുള്ള ഒരു അത്ഭുതകരമായ ഉപകരണത്തെക്കുറിച്ച് ഞാൻ നിങ്ങളോട് പറയും - അപ്പാച്ചെ എയർഫ്ലോ. എന്നാൽ എയർഫ്ലോ വളരെ വൈവിധ്യമാർന്നതും ബഹുമുഖവുമാണ്, നിങ്ങൾ ഡാറ്റാ ഫ്ലോകളിൽ ഏർപ്പെട്ടിട്ടില്ലെങ്കിലും നിങ്ങൾ അത് സൂക്ഷ്മമായി പരിശോധിക്കണം, എന്നാൽ ഏതെങ്കിലും പ്രക്രിയകൾ ഇടയ്ക്കിടെ സമാരംഭിക്കുകയും അവയുടെ നിർവ്വഹണം നിരീക്ഷിക്കുകയും ചെയ്യേണ്ടത് ആവശ്യമാണ്.

അതെ, ഞാൻ പറയുക മാത്രമല്ല, കാണിക്കുകയും ചെയ്യും: പ്രോഗ്രാമിന് ധാരാളം കോഡുകളും സ്ക്രീൻഷോട്ടുകളും ശുപാർശകളും ഉണ്ട്.

എയർഫ്ലോ / വിക്കിമീഡിയ കോമൺസ് എന്ന വാക്ക് ഗൂഗിൾ ചെയ്യുമ്പോൾ നിങ്ങൾ സാധാരണയായി കാണുന്നത്

ഉള്ളടക്ക പട്ടിക

• ആമുഖം
• പ്രധാന ഭാഗം, പ്രായോഗികം (അല്പം സൈദ്ധാന്തികവും)
  • എന്തുകൊണ്ടാണ് ഞങ്ങൾ (നിങ്ങളും)
  • ക്ലസ്റ്റർ കൂട്ടിച്ചേർക്കുന്നു
  • അടിസ്ഥാന ആശയങ്ങൾ
  • ഞങ്ങൾ ടാസ്ക്കുകൾ സൃഷ്ടിക്കുന്നു
  • പുഷ്പത്തെക്കുറിച്ച് കുറച്ച്
  • ഞങ്ങൾ അണ്ടർലോഡഡ് ലോഡ് ചെയ്യുന്നു
  • കണക്ഷനുകൾ, ഹുക്കുകൾ, മറ്റ് വേരിയബിളുകൾ
  • ഇഷ്‌ടാനുസൃത ഓപ്പറേറ്ററെ പാഴ്‌സ് ചെയ്യുന്നു
  • നീ ഒഴിക്കാൻ പോവുകയാണോ?
  • സംഗ്രഹിക്കുന്നു
• ഭാഗം അന്തിമവും റഫറൻസും വിവരങ്ങളും
  • ഞങ്ങൾ നിങ്ങൾക്കായി ശേഖരിച്ച റേക്ക്
  • കൂടുതൽ ഓട്ടോമേഷൻ ടൂളുകൾ
• റെഫറൻസുകൾ

ആമുഖം

ജാങ്കോ പോലെയാണ് അപ്പാച്ചെ എയർഫ്ലോ:

• പൈത്തണിൽ എഴുതിയിരിക്കുന്നു
• ഒരു വലിയ അഡ്മിൻ പാനൽ ഉണ്ട്,
• അനിശ്ചിതമായി വികസിപ്പിക്കാവുന്ന

- നല്ലത് മാത്രം, ഇത് തികച്ചും വ്യത്യസ്തമായ ഉദ്ദേശ്യങ്ങൾക്കായി നിർമ്മിച്ചതാണ്, അതായത് (കാറ്റിന് മുമ്പ് എഴുതിയിരിക്കുന്നതുപോലെ):

• പരിധിയില്ലാത്ത മെഷീനുകളിൽ ടാസ്‌ക്കുകൾ പ്രവർത്തിപ്പിക്കുന്നതും നിരീക്ഷിക്കുന്നതും (നിരവധി സെലറി / കുബർനെറ്റുകളും നിങ്ങളുടെ മനസ്സാക്ഷിയും നിങ്ങളെ അനുവദിക്കും)
• പൈത്തൺ കോഡ് എഴുതാനും മനസ്സിലാക്കാനും വളരെ എളുപ്പമുള്ള ഡൈനാമിക് വർക്ക്ഫ്ലോ ജനറേഷൻ
• കൂടാതെ റെഡിമെയ്ഡ് ഘടകങ്ങളും ഹോം-മെയ്ഡ് പ്ലഗിനുകളും ഉപയോഗിച്ച് ഏതെങ്കിലും ഡാറ്റാബേസുകളും API-കളും പരസ്പരം ബന്ധിപ്പിക്കാനുള്ള കഴിവ് (അത് വളരെ ലളിതമാണ്).

ഞങ്ങൾ അപ്പാച്ചെ എയർഫ്ലോ ഉപയോഗിക്കുന്നത് ഇതുപോലെയാണ്:

• DWH-ലും ODS-ലും (ഞങ്ങൾക്ക് വെർട്ടിക്കയും ക്ലിക്ക്ഹൗസും ഉണ്ട്) വിവിധ ഉറവിടങ്ങളിൽ നിന്ന് (നിരവധി SQL സെർവർ, PostgreSQL സംഭവങ്ങൾ, ആപ്ലിക്കേഷൻ മെട്രിക്കുകളുള്ള വിവിധ API-കൾ, 1C പോലും) ഞങ്ങൾ ഡാറ്റ ശേഖരിക്കുന്നു.
• എത്ര പുരോഗമിച്ചു cron, ഇത് ODS-ൽ ഡാറ്റ ഏകീകരണ പ്രക്രിയകൾ ആരംഭിക്കുകയും അവയുടെ പരിപാലനം നിരീക്ഷിക്കുകയും ചെയ്യുന്നു.

അടുത്ത കാലം വരെ, ഞങ്ങളുടെ ആവശ്യങ്ങൾ 32 കോറുകളും 50 GB റാമും ഉള്ള ഒരു ചെറിയ സെർവറാണ്. എയർഫ്ലോയിൽ, ഇത് പ്രവർത്തിക്കുന്നു:

• более 200 ദിവസം (യഥാർത്ഥത്തിൽ വർക്ക്ഫ്ലോകൾ, അതിൽ ഞങ്ങൾ ജോലികൾ സ്റ്റഫ് ചെയ്യുന്നു)
• ഓരോന്നിലും ശരാശരി 70 ജോലികൾ,
• ഈ നന്മ ആരംഭിക്കുന്നു (ശരാശരിയിലും) മണിക്കൂറിൽ ഒരിക്കൽ.

ഞങ്ങൾ എങ്ങനെ വിപുലീകരിച്ചു എന്നതിനെക്കുറിച്ച്, ഞാൻ ചുവടെ എഴുതാം, എന്നാൽ ഇപ്പോൾ നമ്മൾ പരിഹരിക്കുന്ന ഉബർ-പ്രശ്നം നിർവചിക്കാം:

മൂന്ന് യഥാർത്ഥ SQL സെർവറുകൾ ഉണ്ട്, ഓരോന്നിനും 50 ഡാറ്റാബേസുകൾ ഉണ്ട് - യഥാക്രമം ഒരു പ്രോജക്റ്റിന്റെ ഉദാഹരണങ്ങൾ, അവയ്ക്ക് ഒരേ ഘടനയുണ്ട് (ഏതാണ്ട് എല്ലായിടത്തും, mua-ha-ha), അതായത് ഓരോന്നിനും ഒരു ഓർഡർ പട്ടികയുണ്ട് (ഭാഗ്യവശാൽ, അതിനുള്ള ഒരു പട്ടിക ഏത് ബിസിനസ്സിലേക്കും പേര് തള്ളാം). സേവന ഫീൽഡുകൾ (ഉറവിട സെർവർ, ഉറവിട ഡാറ്റാബേസ്, ETL ടാസ്‌ക് ഐഡി) ചേർത്ത് ഞങ്ങൾ ഡാറ്റ എടുക്കുകയും നിഷ്‌കളങ്കമായി വെർട്ടിക്കയിലേക്ക് എറിയുകയും ചെയ്യുന്നു.

പോകാം!

പ്രധാന ഭാഗം, പ്രായോഗികം (അല്പം സൈദ്ധാന്തികവും)

എന്തുകൊണ്ടാണ് ഞങ്ങൾ (നിങ്ങളും)

മരങ്ങൾ വലുതായിരുന്നപ്പോൾ ഞാൻ ലളിതയായിരുന്നു SQLഒരു റഷ്യൻ റീട്ടെയിലിൽ -schik, ഞങ്ങൾക്ക് ലഭ്യമായ രണ്ട് ടൂളുകൾ ഉപയോഗിച്ച് ഞങ്ങൾ ETL പ്രക്രിയകൾ അല്ലെങ്കിൽ ഡാറ്റാ ഫ്ലോകൾ തട്ടിപ്പ് നടത്തി:

• ഇൻഫോർമാറ്റിക്ക പവർ സെന്റർ - അതിന്റേതായ ഹാർഡ്‌വെയറും സ്വന്തം പതിപ്പും ഉള്ള, അങ്ങേയറ്റം വ്യാപിക്കുന്ന, അത്യധികം ഉൽപ്പാദനക്ഷമതയുള്ള ഒരു സിസ്റ്റം. അതിന്റെ കഴിവിന്റെ 1% ദൈവം വിലക്കുന്നുവെന്ന് ഞാൻ ഉപയോഗിച്ചു. എന്തുകൊണ്ട്? ശരി, ഒന്നാമതായി, ഈ ഇന്റർഫേസ്, 380-കളിൽ എവിടെയോ, മാനസികമായി ഞങ്ങളെ സമ്മർദ്ദത്തിലാക്കി. രണ്ടാമതായി, ഈ കോൺട്രാപ്ഷൻ വളരെ ഫാൻസി പ്രക്രിയകൾ, ഉഗ്രമായ ഘടകത്തിന്റെ പുനരുപയോഗം, മറ്റ് വളരെ പ്രധാനപ്പെട്ട-എന്റർപ്രൈസ്-തന്ത്രങ്ങൾ എന്നിവയ്ക്കായി രൂപകൽപ്പന ചെയ്തിട്ടുള്ളതാണ്. പ്രതിവർഷം എയർബസ് എ XNUMX ന്റെ ചിറക് പോലെ ഇതിന് ചിലവാകും എന്ന വസ്തുതയെക്കുറിച്ച് ഞങ്ങൾ ഒന്നും പറയില്ല.

  സൂക്ഷിക്കുക, ഒരു സ്ക്രീൻഷോട്ട് 30 വയസ്സിന് താഴെയുള്ളവരെ അൽപ്പം വേദനിപ്പിക്കും

  

• SQL സെർവർ ഇന്റഗ്രേഷൻ സെർവർ - ഞങ്ങളുടെ ഇൻട്രാ പ്രോജക്റ്റ് ഫ്ലോകളിൽ ഞങ്ങൾ ഈ സഖാവിനെ ഉപയോഗിച്ചു. ശരി, വാസ്തവത്തിൽ: ഞങ്ങൾ ഇതിനകം തന്നെ SQL സെർവർ ഉപയോഗിക്കുന്നു, മാത്രമല്ല അതിന്റെ ETL ഉപകരണങ്ങൾ ഉപയോഗിക്കാതിരിക്കുന്നത് എങ്ങനെയെങ്കിലും യുക്തിരഹിതമാണ്. ഇതിലെ എല്ലാം നല്ലതാണ്: രണ്ട് ഇന്റർഫേസും മനോഹരമാണ്, കൂടാതെ പുരോഗതി റിപ്പോർട്ടുകളും ... എന്നാൽ ഞങ്ങൾ സോഫ്റ്റ്‌വെയർ ഉൽപ്പന്നങ്ങളെ ഇഷ്ടപ്പെടുന്നത് അതുകൊണ്ടല്ല, ഓ, ഇതിനല്ല. അതിന്റെ പതിപ്പ് dtsx (സംരക്ഷിക്കുമ്പോൾ നോഡുകൾ ഷഫിൾ ചെയ്ത XML ആണ്) നമുക്ക് കഴിയും, എന്നാൽ എന്താണ് കാര്യം? ഒരു സെർവറിൽ നിന്ന് മറ്റൊന്നിലേക്ക് നൂറുകണക്കിന് ടേബിളുകൾ വലിച്ചിടുന്ന ഒരു ടാസ്‌ക് പാക്കേജ് എങ്ങനെ നിർമ്മിക്കാം? അതെ, എന്ത് നൂറ്, നിങ്ങളുടെ ചൂണ്ടു വിരൽ ഇരുപത് കഷണങ്ങളിൽ നിന്ന് വീഴും, മൗസ് ബട്ടണിൽ ക്ലിക്ക് ചെയ്യുക. എന്നാൽ ഇത് തീർച്ചയായും കൂടുതൽ ഫാഷനായി കാണപ്പെടുന്നു:

  

തീർച്ചയായും ഞങ്ങൾ അതിനുള്ള വഴികൾ തേടി. കേസ് പോലും ഏതാണ്ട് സ്വയം എഴുതിയ SSIS പാക്കേജ് ജനറേറ്ററിലേക്ക് വന്നു ...

…പിന്നെ ഒരു പുതിയ ജോലി എന്നെ കണ്ടെത്തി. അപ്പാച്ചെ എയർഫ്ലോ എന്നെ അതിൽ മറികടന്നു.

ETL പ്രോസസ്സ് വിവരണങ്ങൾ ലളിതമായ പൈത്തൺ കോഡാണെന്ന് കണ്ടെത്തിയപ്പോൾ, ഞാൻ സന്തോഷത്തിനായി നൃത്തം ചെയ്തില്ല. ഇങ്ങനെയാണ് ഡാറ്റാ സ്ട്രീമുകൾ പതിപ്പിക്കുകയും വ്യത്യസ്തമാക്കുകയും ചെയ്തത്, നൂറുകണക്കിന് ഡാറ്റാബേസുകളിൽ നിന്ന് ഒരൊറ്റ ഘടനയുള്ള പട്ടികകൾ ഒരു ലക്ഷ്യത്തിലേക്ക് പകരുന്നത് ഒന്നരയോ രണ്ടോ 13 ”സ്‌ക്രീനുകളിലെ പൈത്തൺ കോഡിന്റെ കാര്യമായി മാറി.

ക്ലസ്റ്റർ കൂട്ടിച്ചേർക്കുന്നു

നമുക്ക് പൂർണ്ണമായും കിന്റർഗാർട്ടൻ ക്രമീകരിക്കരുത്, എയർഫ്ലോ, നിങ്ങൾ തിരഞ്ഞെടുത്ത ഡാറ്റാബേസ്, സെലറി, ഡോക്കുകളിൽ വിവരിച്ചിരിക്കുന്ന മറ്റ് കേസുകൾ എന്നിവ ഇൻസ്റ്റാൾ ചെയ്യുന്നത് പോലെയുള്ള വ്യക്തമായ കാര്യങ്ങളെക്കുറിച്ച് ഇവിടെ സംസാരിക്കരുത്.

അതിനാൽ നമുക്ക് ഉടൻ തന്നെ പരീക്ഷണങ്ങൾ ആരംഭിക്കാം, ഞാൻ സ്കെച്ച് ചെയ്തു docker-compose.yml അതിൽ:

• യഥാർത്ഥത്തിൽ ഉയർത്താം വായു പ്രവാഹം: ഷെഡ്യൂളർ, വെബ്‌സെർവർ. സെലറി ടാസ്‌ക്കുകൾ നിരീക്ഷിക്കാൻ ഫ്ലവറും അവിടെ കറങ്ങും (കാരണം ഇത് ഇതിനകം തന്നെ തള്ളപ്പെട്ടിരിക്കുന്നു apache/airflow:1.10.10-python3.7, പക്ഷേ ഞങ്ങൾ കാര്യമാക്കുന്നില്ല)
• PostgreSQL എന്നീ, അതിൽ എയർഫ്ലോ അതിന്റെ സേവന വിവരങ്ങൾ (ഷെഡ്യൂളർ ഡാറ്റ, എക്സിക്യൂഷൻ സ്റ്റാറ്റിസ്റ്റിക്സ് മുതലായവ) എഴുതുകയും സെലറി പൂർത്തിയാക്കിയ ജോലികൾ അടയാളപ്പെടുത്തുകയും ചെയ്യും;
• രെദിസ്, ഇത് സെലറിയുടെ ഒരു ടാസ്‌ക് ബ്രോക്കറായി പ്രവർത്തിക്കും;
• സെലറി തൊഴിലാളി, ടാസ്ക്കുകളുടെ നേരിട്ടുള്ള നിർവ്വഹണത്തിൽ ഏർപ്പെട്ടിരിക്കും.
• ഫോൾഡറിലേക്ക് ./dags ഡാഗുകളുടെ വിവരണത്തോടൊപ്പം ഞങ്ങൾ ഞങ്ങളുടെ ഫയലുകൾ ചേർക്കും. ഈച്ചയിൽ അവ എടുക്കപ്പെടും, അതിനാൽ ഓരോ തുമ്മലിനു ശേഷവും മുഴുവൻ സ്റ്റാക്കും കബളിപ്പിക്കേണ്ട ആവശ്യമില്ല.

ചില സ്ഥലങ്ങളിൽ, ഉദാഹരണങ്ങളിലെ കോഡ് പൂർണ്ണമായി കാണിച്ചിട്ടില്ല (അതിനാൽ വാചകം അലങ്കോലപ്പെടുത്താതിരിക്കാൻ), എന്നാൽ എവിടെയെങ്കിലും അത് പ്രക്രിയയിൽ പരിഷ്ക്കരിച്ചിരിക്കുന്നു. പൂർണ്ണമായ വർക്കിംഗ് കോഡ് ഉദാഹരണങ്ങൾ റിപ്പോസിറ്ററിയിൽ കാണാം https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

version: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

services:
  # Redis as a Celery broker
  broker:
    image: redis:6.0.5-alpine

  # DB for the Airflow metadata
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Main container with Airflow Webserver, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <<: *airflow-config

      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'
      AIRFLOW__SCHEDULER__MAX_THREADS: 8

      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10

    depends_on:
      - airflow-db
      - broker

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Airflow Webserver
      - 8080:8080

  # Celery worker, will be scaled using `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <<: *airflow-config

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

കുറിപ്പുകൾ:

• കോമ്പോസിഷന്റെ അസംബ്ലിയിൽ, ഞാൻ പ്രധാനമായും അറിയപ്പെടുന്ന ചിത്രത്തെ ആശ്രയിച്ചു puckel/docker-airflow - അത് പരിശോധിക്കുന്നത് ഉറപ്പാക്കുക. ഒരുപക്ഷേ നിങ്ങളുടെ ജീവിതത്തിൽ മറ്റൊന്നും ആവശ്യമില്ല.
• വഴി മാത്രമല്ല എല്ലാ എയർഫ്ലോ ക്രമീകരണങ്ങളും ലഭ്യമാണ് airflow.cfg, മാത്രമല്ല പരിസ്ഥിതി വേരിയബിളുകൾ വഴിയും (ഡെവലപ്പർമാർക്ക് നന്ദി), ഞാൻ അത് ദുരുദ്ദേശ്യത്തോടെ പ്രയോജനപ്പെടുത്തി.
• സ്വാഭാവികമായും, ഇത് ഉൽ‌പാദനത്തിന് തയ്യാറല്ല: ഞാൻ മനഃപൂർവം കണ്ടെയ്‌നറുകളിൽ ഹൃദയമിടിപ്പ് ഇട്ടില്ല, സുരക്ഷയെക്കുറിച്ച് ഞാൻ വിഷമിച്ചില്ല. എന്നാൽ ഞങ്ങളുടെ പരീക്ഷണക്കാർക്ക് ഏറ്റവും അനുയോജ്യമായത് ഞാൻ ചെയ്തു.
• അതല്ല:
  • ഡാഗ് ഫോൾഡർ ഷെഡ്യൂളർക്കും തൊഴിലാളികൾക്കും ആക്സസ് ചെയ്യാവുന്നതായിരിക്കണം.
  • എല്ലാ മൂന്നാം കക്ഷി ലൈബ്രറികൾക്കും ഇത് ബാധകമാണ് - അവയെല്ലാം ഒരു ഷെഡ്യൂളറും തൊഴിലാളികളും ഉള്ള മെഷീനുകളിൽ ഇൻസ്റ്റാൾ ചെയ്തിരിക്കണം.

ശരി, ഇപ്പോൾ ഇത് ലളിതമാണ്:

$ docker-compose up --scale worker=3

എല്ലാം ഉയർന്നതിന് ശേഷം, നിങ്ങൾക്ക് വെബ് ഇന്റർഫേസുകൾ നോക്കാം:

• വായു പ്രവാഹം: http://127.0.0.1:8080/admin/
• പുഷ്പം: http://127.0.0.1:5555/dashboard

അടിസ്ഥാന ആശയങ്ങൾ

ഈ "ഡാഗുകളിൽ" നിങ്ങൾക്ക് ഒന്നും മനസ്സിലായില്ലെങ്കിൽ, ഇതാ ഒരു ചെറിയ നിഘണ്ടു:

• ഷെഡ്യൂളർ - എയർഫ്ലോയിലെ ഏറ്റവും പ്രധാനപ്പെട്ട അമ്മാവൻ, റോബോട്ടുകൾ കഠിനാധ്വാനം ചെയ്യുന്നുവെന്ന് നിയന്ത്രിക്കുന്നു, ഒരു വ്യക്തിയല്ല: ഷെഡ്യൂൾ നിരീക്ഷിക്കുന്നു, ഡാഗുകൾ അപ്ഡേറ്റ് ചെയ്യുന്നു, ടാസ്ക്കുകൾ സമാരംഭിക്കുന്നു.

  പൊതുവേ, പഴയ പതിപ്പുകളിൽ, അദ്ദേഹത്തിന് മെമ്മറിയിൽ പ്രശ്നങ്ങളുണ്ടായിരുന്നു (ഇല്ല, ഓർമ്മക്കുറവല്ല, ചോർച്ച) കൂടാതെ ലെഗസി പാരാമീറ്റർ കോൺഫിഗറുകളിൽ പോലും തുടർന്നു. run_duration - അതിന്റെ പുനരാരംഭിക്കൽ ഇടവേള. എന്നാൽ ഇപ്പോൾ എല്ലാം ശരിയാണ്.

• DAG (അതായത് "ഡാഗ്") - "ഡയറക്ട്ഡ് അസൈക്ലിക് ഗ്രാഫ്", എന്നാൽ അത്തരമൊരു നിർവചനം കുറച്ച് ആളുകളോട് പറയും, എന്നാൽ വാസ്തവത്തിൽ ഇത് പരസ്പരം ഇടപഴകുന്ന ജോലികൾക്കായുള്ള ഒരു കണ്ടെയ്നറാണ് (ചുവടെ കാണുക) അല്ലെങ്കിൽ SSIS-ലെ പാക്കേജിന്റെയും ഇൻഫോർമാറ്റിക്കയിലെ വർക്ക്ഫ്ലോയുടെയും അനലോഗ് .

  ഡാഗുകൾക്ക് പുറമേ, ഇപ്പോഴും സബ്ഡാഗുകൾ ഉണ്ടാകാം, പക്ഷേ നമുക്ക് അവ ലഭിക്കില്ല.

• 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]] എയർഫ്ലോ കണക്ഷനുകളിൽ നിന്നുള്ള കണക്ഷനുകളുടെ പേരുകളും ഞങ്ങളുടെ പ്ലേറ്റ് എടുക്കുന്ന ഡാറ്റാബേസുകളും;
• dag - ഞങ്ങളുടെ ഡാഗിന്റെ അറിയിപ്പ്, അത് നിർബന്ധമായും ഉണ്ടായിരിക്കണം globals(), അല്ലെങ്കിൽ എയർഫ്ലോ അത് കണ്ടെത്തുകയില്ല. ഡോഗും പറയേണ്ടതുണ്ട്:
  • അവന്റെ പേരെന്താണ്? orders - ഈ പേര് പിന്നീട് വെബ് ഇന്റർഫേസിൽ ദൃശ്യമാകും,
  • ജൂലൈ എട്ടാം തീയതി അർദ്ധരാത്രി മുതൽ താൻ ജോലി ചെയ്യുമെന്ന്,
  • ഇത് ഏകദേശം ഓരോ 6 മണിക്കൂറിലും പ്രവർത്തിക്കണം (ഇവിടെ ബുദ്ധിമുട്ടുള്ള ആളുകൾക്ക് പകരം timedelta() സ്വീകാര്യമായ cron-ലൈൻ 0 0 0/6 ? * * *, കുറഞ്ഞ തണുപ്പിന് - പോലെയുള്ള ഒരു പദപ്രയോഗം @daily);
• workflow() പ്രധാന ജോലി ചെയ്യും, പക്ഷേ ഇപ്പോൾ അല്ല. ഇപ്പോൾ, ഞങ്ങൾ ഞങ്ങളുടെ സന്ദർഭം ലോഗിലേക്ക് ഡംപ് ചെയ്യും.
• ഇപ്പോൾ ടാസ്ക്കുകൾ സൃഷ്ടിക്കുന്നതിനുള്ള ലളിതമായ മാജിക്:
  • ഞങ്ങൾ ഞങ്ങളുടെ ഉറവിടങ്ങളിലൂടെ ഓടുന്നു;
  • സമാരംഭിക്കുക PythonOperator, അത് നമ്മുടെ ഡമ്മിയെ നിർവ്വഹിക്കും workflow(). ടാസ്‌ക്കിന്റെ ഒരു അദ്വിതീയ (ഡാഗിനുള്ളിൽ) പേര് വ്യക്തമാക്കാനും ഡാഗ് തന്നെ കെട്ടാനും മറക്കരുത്. പതാക provide_context അതാകട്ടെ, ഫംഗ്ഷനിലേക്ക് അധിക ആർഗ്യുമെന്റുകൾ പകരും, അത് ഞങ്ങൾ ഉപയോഗിച്ച് ശ്രദ്ധാപൂർവ്വം ശേഖരിക്കും **context.

തൽക്കാലം അത്രമാത്രം. ഞങ്ങൾക്ക് ലഭിച്ചത്:

• വെബ് ഇന്റർഫേസിൽ പുതിയ ഡാഗ്,
• ഒന്നരനൂറ് ജോലികൾ സമാന്തരമായി നടപ്പിലാക്കും (എയർഫ്ലോ, സെലറി ക്രമീകരണങ്ങൾ, സെർവർ ശേഷി എന്നിവ അനുവദിക്കുകയാണെങ്കിൽ).

ശരി, ഏകദേശം മനസ്സിലായി.

ആരാണ് ഡിപൻഡൻസികൾ ഇൻസ്റ്റാൾ ചെയ്യുക?

ഇത് മുഴുവൻ ലളിതമാക്കാൻ, ഞാൻ സ്ക്രൂ ചെയ്തു docker-compose.yml പ്രോസസ്സിംഗ് requirements.txt എല്ലാ നോഡുകളിലും.

ഇപ്പോൾ ഇതാ:

ഷെഡ്യൂളർ പ്രോസസ്സ് ചെയ്യുന്ന ടാസ്‌ക് ഇൻസ്‌റ്റൻസുകളാണ് ഗ്രേ സ്‌ക്വയറുകൾ.

ഞങ്ങൾ അൽപ്പം കാത്തിരിക്കുന്നു, ജോലിക്കാർ ജോലികൾ സ്നാപ്പ് ചെയ്യുന്നു:

പച്ചയായവർ തീർച്ചയായും അവരുടെ ജോലി വിജയകരമായി പൂർത്തിയാക്കി. ചുവപ്പ് വളരെ വിജയകരമല്ല.

വഴിയിൽ, ഞങ്ങളുടെ ഉൽപ്പന്നത്തിൽ ഒരു ഫോൾഡറും ഇല്ല ./dags, മെഷീനുകൾക്കിടയിൽ സിൻക്രൊണൈസേഷൻ ഇല്ല - എല്ലാ ഡാഗുകളും കിടക്കുന്നു git ഞങ്ങളുടെ Gitlab-ൽ, ഒപ്പം Gitlab CI, ലയിപ്പിക്കുമ്പോൾ മെഷീനുകളിലേക്ക് അപ്‌ഡേറ്റുകൾ വിതരണം ചെയ്യുന്നു master.

പുഷ്പത്തെക്കുറിച്ച് കുറച്ച്

തൊഴിലാളികൾ നമ്മുടെ പസിഫയറുകൾ തല്ലിക്കൊല്ലുമ്പോൾ, നമുക്ക് എന്തെങ്കിലും കാണിക്കാൻ കഴിയുന്ന മറ്റൊരു ഉപകരണം ഓർക്കാം - പുഷ്പം.

വർക്കർ നോഡുകളെക്കുറിച്ചുള്ള സംഗ്രഹ വിവരങ്ങളുള്ള ആദ്യ പേജ്:

ജോലിക്ക് പോയ ടാസ്ക്കുകളുള്ള ഏറ്റവും തീവ്രമായ പേജ്:

ഞങ്ങളുടെ ബ്രോക്കറുടെ പദവിയുള്ള ഏറ്റവും വിരസമായ പേജ്:

ടാസ്‌ക് സ്റ്റാറ്റസ് ഗ്രാഫുകളും അവയുടെ നിർവ്വഹണ സമയവുമാണ് ഏറ്റവും തിളക്കമുള്ള പേജ്:

ഞങ്ങൾ അണ്ടർലോഡഡ് ലോഡ് ചെയ്യുന്നു

അതിനാൽ, എല്ലാ ജോലികളും പൂർത്തിയായി, നിങ്ങൾക്ക് പരിക്കേറ്റവരെ കൊണ്ടുപോകാൻ കഴിയും.

കൂടാതെ നിരവധി പേർക്ക് പരിക്കേറ്റു - ഒരു കാരണത്താലോ മറ്റൊരു കാരണത്താലോ. എയർഫ്ലോയുടെ ശരിയായ ഉപയോഗത്തിന്റെ കാര്യത്തിൽ, ഡാറ്റ തീർച്ചയായും എത്തിയിട്ടില്ലെന്ന് ഇതേ സ്ക്വയറുകൾ സൂചിപ്പിക്കുന്നു.

നിങ്ങൾ ലോഗ് കാണുകയും വീണുപോയ ടാസ്‌ക് സന്ദർഭങ്ങൾ പുനരാരംഭിക്കുകയും വേണം.

ഏതെങ്കിലും ചതുരത്തിൽ ക്ലിക്കുചെയ്യുന്നതിലൂടെ, ഞങ്ങൾക്ക് ലഭ്യമായ പ്രവർത്തനങ്ങൾ ഞങ്ങൾ കാണും:

നിങ്ങൾക്ക് എടുത്ത് വീണത് ക്ലിയർ ചെയ്യാം. അതായത്, അവിടെ എന്തെങ്കിലും പരാജയപ്പെട്ടുവെന്ന് ഞങ്ങൾ മറക്കുന്നു, അതേ ഇൻസ്‌റ്റൻസ് ടാസ്‌ക് ഷെഡ്യൂളറിലേക്ക് പോകും.

എല്ലാ ചുവന്ന ചതുരങ്ങളുമുള്ള മൗസ് ഉപയോഗിച്ച് ഇത് ചെയ്യുന്നത് വളരെ മാനുഷികമല്ലെന്ന് വ്യക്തമാണ് - ഇത് എയർഫ്ലോയിൽ നിന്ന് ഞങ്ങൾ പ്രതീക്ഷിക്കുന്നില്ല. സ്വാഭാവികമായും, നമുക്ക് കൂട്ട നശീകരണ ആയുധങ്ങളുണ്ട്: Browse/Task Instances

നമുക്ക് എല്ലാം ഒറ്റയടിക്ക് തിരഞ്ഞെടുത്ത് പൂജ്യത്തിലേക്ക് പുനഃസജ്ജമാക്കാം, ശരിയായ ഇനത്തിൽ ക്ലിക്കുചെയ്യുക:

വൃത്തിയാക്കിയ ശേഷം, ഞങ്ങളുടെ ടാക്സികൾ ഇതുപോലെ കാണപ്പെടുന്നു (അവർ ഷെഡ്യൂളർ ഷെഡ്യൂൾ ചെയ്യുന്നതിനായി കാത്തിരിക്കുകയാണ്):

കണക്ഷനുകൾ, ഹുക്കുകൾ, മറ്റ് വേരിയബിളുകൾ

അടുത്ത DAG നോക്കാനുള്ള സമയമാണിത്, update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='DWH Reports updated',
    html_content=dedent("""Господа хорошие, отчеты обновлены"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Наташ, просыпайся, мы {{ dag.dag_id }} уронили
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

എല്ലാവരും എപ്പോഴെങ്കിലും ഒരു റിപ്പോർട്ട് അപ്ഡേറ്റ് ചെയ്തിട്ടുണ്ടോ? ഇത് വീണ്ടും അവളാണ്: ഡാറ്റ എവിടെ നിന്ന് ലഭിക്കും എന്നതിന്റെ ഉറവിടങ്ങളുടെ ഒരു ലിസ്റ്റ് ഉണ്ട്; എവിടെ വയ്ക്കണമെന്ന് ഒരു ലിസ്റ്റ് ഉണ്ട്; എല്ലാം സംഭവിക്കുമ്പോഴോ തകരുമ്പോഴോ ഹോൺ ചെയ്യാൻ മറക്കരുത് (ശരി, ഇത് ഞങ്ങളെക്കുറിച്ചല്ല, ഇല്ല).

നമുക്ക് വീണ്ടും ഫയലിലൂടെ പോയി പുതിയ അവ്യക്തമായ കാര്യങ്ങൾ നോക്കാം:

• from commons.operators import TelegramBotSendMessage - അൺബ്ലോക്ക് ചെയ്തതിലേക്ക് സന്ദേശങ്ങൾ അയയ്‌ക്കുന്നതിന് ഒരു ചെറിയ റാപ്പർ നിർമ്മിച്ചുകൊണ്ട് ഞങ്ങൾ പ്രയോജനപ്പെടുത്തിയ ഞങ്ങളുടെ സ്വന്തം ഓപ്പറേറ്റർമാരെ ഉണ്ടാക്കുന്നതിൽ നിന്ന് ഒന്നും ഞങ്ങളെ തടയുന്നില്ല. (ഈ ഓപ്പറേറ്ററെ കുറിച്ച് ഞങ്ങൾ താഴെ സംസാരിക്കും);
• default_args={} - dag-ന് അതിന്റെ എല്ലാ ഓപ്പറേറ്റർമാർക്കും ഒരേ ആർഗ്യുമെന്റുകൾ വിതരണം ചെയ്യാൻ കഴിയും;
• to='{{ var.value.all_the_kings_men }}' - ഫീൽഡ് to ഞങ്ങൾ ഹാർഡ്‌കോഡ് ചെയ്യില്ല, പക്ഷേ ജിൻജയും ഇമെയിലുകളുടെ ഒരു ലിസ്റ്റ് ഉള്ള ഒരു വേരിയബിളും ഉപയോഗിച്ച് ചലനാത്മകമായി ജനറേറ്റ് ചെയ്‌തിരിക്കുന്നു, അത് ഞാൻ ശ്രദ്ധാപൂർവ്വം ഇട്ടു Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS - ഓപ്പറേറ്റർ ആരംഭിക്കുന്നതിനുള്ള വ്യവസ്ഥ. ഞങ്ങളുടെ കാര്യത്തിൽ, എല്ലാ ആശ്രിതത്വങ്ങളും പ്രവർത്തിച്ചാൽ മാത്രമേ കത്ത് മേലധികാരികൾക്ക് പറക്കും വിജയകരമായി;
• tg_bot_conn_id='tg_main' - വാദങ്ങൾ conn_id ഞങ്ങൾ സൃഷ്ടിക്കുന്ന കണക്ഷൻ ഐഡികൾ സ്വീകരിക്കുക Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED - വീണുപോയ ടാസ്‌ക്കുകൾ ഉണ്ടെങ്കിൽ മാത്രമേ ടെലിഗ്രാമിലെ സന്ദേശങ്ങൾ പറന്നുപോകൂ;
• task_concurrency=1 - ഒരു ടാസ്‌ക്കിന്റെ നിരവധി ടാസ്‌ക് സന്ദർഭങ്ങൾ ഒരേസമയം സമാരംഭിക്കുന്നത് ഞങ്ങൾ നിരോധിക്കുന്നു. അല്ലാത്തപക്ഷം, നമുക്ക് പലതിന്റെയും ഒരേസമയം ലോഞ്ച് ലഭിക്കും VerticaOperator (ഒരു മേശയിലേക്ക് നോക്കുന്നു);
• report_update >> [email, tg] - എല്ലാം VerticaOperator ഇതുപോലുള്ള അക്ഷരങ്ങളും സന്ദേശങ്ങളും അയയ്ക്കുന്നതിൽ ഒത്തുചേരുക:
  

  എന്നാൽ നോട്ടിഫയർ ഓപ്പറേറ്റർമാർക്ക് വ്യത്യസ്‌ത ലോഞ്ച് വ്യവസ്ഥകൾ ഉള്ളതിനാൽ, ഒരെണ്ണം മാത്രമേ പ്രവർത്തിക്കൂ. ട്രീ വ്യൂവിൽ, എല്ലാം കുറച്ചുകൂടി ദൃശ്യപരമായി കാണപ്പെടുന്നു:
  

ഞാൻ ഇതിനെക്കുറിച്ച് കുറച്ച് വാക്കുകൾ പറയും മാക്രോകൾ അവരുടെ സുഹൃത്തുക്കളും - വേരിയബിളുകൾ.

വിവിധ ഉപയോഗപ്രദമായ വിവരങ്ങൾ ഓപ്പറേറ്റർ ആർഗ്യുമെന്റുകളിലേക്ക് പകരാൻ കഴിയുന്ന ജിഞ്ച പ്ലെയ്‌സ്‌ഹോൾഡറുകളാണ് മാക്രോകൾ. ഉദാഹരണത്തിന്, ഇതുപോലെ:

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} സന്ദർഭ വേരിയബിളിന്റെ ഉള്ളടക്കങ്ങളിലേക്ക് വികസിപ്പിക്കും execution_date ഫോർമാറ്റിൽ YYYY-MM-DD: 2020-07-14. സന്ദർഭ വേരിയബിളുകൾ ഒരു നിർദ്ദിഷ്‌ട ടാസ്‌ക് ഇൻസ്‌റ്റൻസിലേക്ക് (ട്രീ വ്യൂവിലെ ഒരു സ്‌ക്വയർ) നെയിൽ ചെയ്‌തിരിക്കുന്നു എന്നതാണ് ഏറ്റവും നല്ല ഭാഗം, പുനരാരംഭിക്കുമ്പോൾ, പ്ലെയ്‌സ്‌ഹോൾഡറുകൾ അതേ മൂല്യങ്ങളിലേക്ക് വികസിക്കും.

ഓരോ ടാസ്‌ക് ഇൻസ്‌റ്റൻസിലും റെൻഡർ ചെയ്‌ത ബട്ടൺ ഉപയോഗിച്ച് നിയുക്ത മൂല്യങ്ങൾ കാണാൻ കഴിയും. ഒരു കത്ത് അയയ്ക്കുന്നതിനുള്ള ചുമതല ഇങ്ങനെയാണ്:

അങ്ങനെ ഒരു സന്ദേശം അയക്കുന്ന ചുമതലയിൽ:

ലഭ്യമായ ഏറ്റവും പുതിയ പതിപ്പിനായുള്ള ബിൽറ്റ്-ഇൻ മാക്രോകളുടെ പൂർണ്ണമായ ലിസ്റ്റ് ഇവിടെ ലഭ്യമാണ്: മാക്രോ റഫറൻസ്

മാത്രമല്ല, പ്ലഗിനുകളുടെ സഹായത്തോടെ, നമുക്ക് നമ്മുടെ സ്വന്തം മാക്രോകൾ പ്രഖ്യാപിക്കാൻ കഴിയും, പക്ഷേ അത് മറ്റൊരു കഥയാണ്.

മുൻകൂട്ടി നിശ്ചയിച്ച കാര്യങ്ങൾക്ക് പുറമേ, നമുക്ക് നമ്മുടെ വേരിയബിളുകളുടെ മൂല്യങ്ങൾ മാറ്റിസ്ഥാപിക്കാം (മുകളിലുള്ള കോഡിൽ ഞാൻ ഇത് ഇതിനകം ഉപയോഗിച്ചു). നമുക്ക് സൃഷ്ടിക്കാം Admin/Variables ഒന്നുരണ്ടു കാര്യങ്ങൾ:

നിങ്ങൾക്ക് ഉപയോഗിക്കാൻ കഴിയുന്ന എല്ലാം:

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

മൂല്യം ഒരു സ്കെയിലർ ആകാം, അല്ലെങ്കിൽ അത് JSON ആകാം. JSON-ന്റെ കാര്യത്തിൽ:

bot_config

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

ആവശ്യമുള്ള കീയിലേക്കുള്ള പാത ഉപയോഗിക്കുക: {{ var.json.bot_config.bot.token }}.

ഞാൻ അക്ഷരാർത്ഥത്തിൽ ഒരു വാക്ക് പറയുകയും ഒരു സ്ക്രീൻഷോട്ട് കാണിക്കുകയും ചെയ്യും соединения. എല്ലാം ഇവിടെ പ്രാഥമികമാണ്: പേജിൽ Admin/Connections ഞങ്ങൾ ഒരു കണക്ഷൻ സൃഷ്ടിക്കുന്നു, അവിടെ ഞങ്ങളുടെ ലോഗിനുകൾ / പാസ്‌വേഡുകൾ, കൂടുതൽ നിർദ്ദിഷ്ട പാരാമീറ്ററുകൾ എന്നിവ ചേർക്കുക. ഇതുപോലെ:

പാസ്‌വേഡുകൾ എൻക്രിപ്റ്റ് ചെയ്യാൻ കഴിയും (ഡിഫോൾട്ടിനെക്കാൾ കൂടുതൽ നന്നായി), അല്ലെങ്കിൽ നിങ്ങൾക്ക് കണക്ഷൻ തരം ഉപേക്ഷിക്കാം (ഞാൻ ചെയ്തതുപോലെ tg_main) - എയർഫ്ലോ മോഡലുകളിൽ തരങ്ങളുടെ ലിസ്റ്റ് ഹാർഡ് വയർഡ് ആയതിനാൽ സോഴ്സ് കോഡുകളിലേക്ക് കടക്കാതെ വിപുലീകരിക്കാൻ കഴിയില്ല എന്നതാണ് വസ്തുത (ഞാൻ പെട്ടെന്ന് എന്തെങ്കിലും ഗൂഗിൾ ചെയ്തില്ലെങ്കിൽ, ദയവായി എന്നെ ശരിയാക്കുക), എന്നാൽ ക്രെഡിറ്റുകൾ ലഭിക്കുന്നതിൽ നിന്ന് ഒന്നും ഞങ്ങളെ തടയില്ല പേര്.

നിങ്ങൾക്ക് ഒരേ പേരിൽ നിരവധി കണക്ഷനുകൾ ഉണ്ടാക്കാം: ഈ സാഹചര്യത്തിൽ, രീതി BaseHook.get_connection(), നമുക്ക് പേരുകൾ ഉപയോഗിച്ച് കണക്ഷനുകൾ ലഭിക്കുന്നത്, തരും ക്രമരഹിതമായ നിരവധി പേരുകളിൽ നിന്ന് (റൗണ്ട് റോബിൻ നിർമ്മിക്കുന്നത് കൂടുതൽ യുക്തിസഹമായിരിക്കും, പക്ഷേ നമുക്ക് അത് എയർഫ്ലോ ഡെവലപ്പർമാരുടെ മനസ്സാക്ഷിക്ക് വിടാം).

വേരിയബിളുകളും കണക്ഷനുകളും തീർച്ചയായും രസകരമായ ഉപകരണങ്ങളാണ്, എന്നാൽ ബാലൻസ് നഷ്‌ടപ്പെടാതിരിക്കേണ്ടത് പ്രധാനമാണ്: നിങ്ങളുടെ ഫ്ലോകളുടെ ഏതൊക്കെ ഭാഗങ്ങൾ കോഡിൽ തന്നെ സംഭരിക്കുന്നു, സംഭരണത്തിനായി എയർഫ്ലോയ്‌ക്ക് ഏതൊക്കെ ഭാഗങ്ങൾ നൽകുന്നു. ഒരു വശത്ത്, മൂല്യം വേഗത്തിൽ മാറ്റുന്നത് സൗകര്യപ്രദമായിരിക്കും, ഉദാഹരണത്തിന്, ഒരു മെയിലിംഗ് ബോക്സ്, യുഐ വഴി. മറുവശത്ത്, ഇത് ഇപ്പോഴും മൗസ് ക്ലിക്കിലേക്കുള്ള മടക്കമാണ്, അതിൽ നിന്ന് ഞങ്ങൾ (ഞാൻ) രക്ഷപ്പെടാൻ ആഗ്രഹിച്ചു.

കണക്ഷനുകൾക്കൊപ്പം പ്രവർത്തിക്കുന്നത് ചുമതലകളിൽ ഒന്നാണ് കൊളുത്തുകൾ. പൊതുവേ, എയർഫ്ലോ ഹുക്കുകൾ മൂന്നാം കക്ഷി സേവനങ്ങളിലേക്കും ലൈബ്രറികളിലേക്കും ബന്ധിപ്പിക്കുന്നതിനുള്ള പോയിന്റുകളാണ്. ഉദാ, JiraHook ജിറയുമായി ഇടപഴകാൻ ഞങ്ങൾക്ക് ഒരു ക്ലയന്റ് തുറക്കും (നിങ്ങൾക്ക് ടാസ്‌ക്കുകൾ അങ്ങോട്ടും ഇങ്ങോട്ടും നീക്കാൻ കഴിയും), കൂടാതെ SambaHook നിങ്ങൾക്ക് ഒരു ലോക്കൽ ഫയൽ പുഷ് ചെയ്യാം smb-പോയിന്റ്.

ഇഷ്‌ടാനുസൃത ഓപ്പറേറ്ററെ പാഴ്‌സ് ചെയ്യുന്നു

അത് എങ്ങനെ ഉണ്ടാക്കുന്നു എന്ന് നോക്കാൻ ഞങ്ങൾ അടുത്തു TelegramBotSendMessage

കോഡ് commons/operators.py യഥാർത്ഥ ഓപ്പറേറ്റർക്കൊപ്പം:

from typing import Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Send message to chat_id using TelegramBotHook

    Example:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Union[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Send "{self.message}" to the chat {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

ഇവിടെ, എയർഫ്ലോയിലെ മറ്റെല്ലാ കാര്യങ്ങളും പോലെ, എല്ലാം വളരെ ലളിതമാണ്:

• പാരമ്പര്യമായി ലഭിച്ചത് BaseOperator, ഇത് കുറച്ച് എയർഫ്ലോ-നിർദ്ദിഷ്ട കാര്യങ്ങൾ നടപ്പിലാക്കുന്നു (നിങ്ങളുടെ ഒഴിവുസമയം നോക്കുക)
• പ്രഖ്യാപിത ഫീൽഡുകൾ template_fields, അതിൽ പ്രോസസ്സ് ചെയ്യാൻ ജിൻജ മാക്രോകൾ നോക്കും.
• ശരിയായ വാദങ്ങൾ ക്രമീകരിച്ചു __init__(), ആവശ്യമുള്ളിടത്ത് ഡിഫോൾട്ടുകൾ സജ്ജമാക്കുക.
• പൂർവ്വികന്റെ തുടക്കത്തെക്കുറിച്ചും ഞങ്ങൾ മറന്നില്ല.
• അനുബന്ധ ഹുക്ക് തുറന്നു TelegramBotHookഅതിൽ നിന്ന് ഒരു ക്ലയന്റ് ഒബ്ജക്റ്റ് ലഭിച്ചു.
• അസാധുവാക്കപ്പെട്ട (പുനർനിർവചിക്കപ്പെട്ട) രീതി BaseOperator.execute(), ഓപ്പറേറ്റർ സമാരംഭിക്കുന്നതിനുള്ള സമയം വരുമ്പോൾ ഏത് എയർഫോ വലിക്കും - അതിൽ ഞങ്ങൾ ലോഗിൻ ചെയ്യാൻ മറന്ന് പ്രധാന പ്രവർത്തനം നടപ്പിലാക്കും. (ഞങ്ങൾ നേരിട്ട് ലോഗിൻ ചെയ്യുന്നു stdout и stderr - വായുപ്രവാഹം എല്ലാം തടസ്സപ്പെടുത്തുകയും മനോഹരമായി പൊതിയുകയും ആവശ്യമുള്ളിടത്ത് വിഘടിപ്പിക്കുകയും ചെയ്യും.)

നമുക്ക് എന്താണ് ഉള്ളതെന്ന് നോക്കാം commons/hooks.py. ഫയലിന്റെ ആദ്യ ഭാഗം, ഹുക്ക് തന്നെ:

from typing import Union

from airflow.hooks.base_hook import BaseHook
from requests_toolbelt.sessions import BaseUrlSession

class TelegramBotHook(BaseHook):
    """Telegram Bot API hook

    Note: add a connection with empty Conn Type and don't forget
    to fill Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

ഇവിടെ എന്താണ് വിശദീകരിക്കേണ്ടതെന്ന് എനിക്കറിയില്ല, പ്രധാനപ്പെട്ട കാര്യങ്ങൾ ഞാൻ ശ്രദ്ധിക്കും:

• ഞങ്ങൾക്ക് അവകാശമുണ്ട്, വാദങ്ങളെക്കുറിച്ച് ചിന്തിക്കുക - മിക്ക കേസുകളിലും ഇത് ഒന്നായിരിക്കും: conn_id;
• സ്റ്റാൻഡേർഡ് രീതികൾ മറികടക്കുന്നു: ഞാൻ എന്നെത്തന്നെ പരിമിതപ്പെടുത്തി get_conn(), അതിൽ എനിക്ക് കണക്ഷൻ പാരാമീറ്ററുകൾ പേര് പ്രകാരം ലഭിക്കുകയും വിഭാഗം നേടുകയും ചെയ്യുന്നു extra (ഇതൊരു JSON ഫീൽഡാണ്), അതിൽ ഞാൻ (എന്റെ സ്വന്തം നിർദ്ദേശങ്ങൾ അനുസരിച്ച്!) ടെലിഗ്രാം ബോട്ട് ടോക്കൺ ഇട്ടു: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• ഞാൻ ഞങ്ങളുടെ ഒരു ഉദാഹരണം സൃഷ്ടിക്കുന്നു TelegramBot, അതിന് ഒരു പ്രത്യേക ടോക്കൺ നൽകുന്നു.

അത്രയേയുള്ളൂ. ഒരു ഹുക്ക് ഉപയോഗിച്ച് നിങ്ങൾക്ക് ഒരു ക്ലയന്റ് ലഭിക്കും TelegramBotHook().clent അഥവാ TelegramBotHook().get_conn().

ടെലിഗ്രാം 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 സെർവറിൽ നിന്ന് വെർട്ടിക്കയിലേക്ക് ഡാറ്റ ട്രാൻസ്ഫർ ചെയ്യാമെന്ന് അദ്ദേഹം വാഗ്ദാനം ചെയ്തതായി തോന്നുന്നു, എന്നിട്ട് അദ്ദേഹം അത് എടുത്ത് വിഷയത്തിൽ നിന്ന് മാറി, അഴിമതി!

ഈ ക്രൂരത മനഃപൂർവമായിരുന്നു, എനിക്ക് നിങ്ങൾക്കായി ചില പദാവലികൾ മനസ്സിലാക്കേണ്ടിവന്നു. ഇപ്പോൾ നിങ്ങൾക്ക് കൂടുതൽ മുന്നോട്ട് പോകാം.

ഞങ്ങളുടെ പദ്ധതി ഇതായിരുന്നു:

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. ഒരു ഹുക്ക് സഹായത്തോടെ നമുക്ക് എയർഫ്ലോയിൽ നിന്ന് ലഭിക്കും pymssql- ബന്ധിപ്പിക്കുക
2. അഭ്യർത്ഥനയിലേക്ക് ഒരു തീയതിയുടെ രൂപത്തിൽ ഒരു നിയന്ത്രണം മാറ്റിസ്ഥാപിക്കാം - ഇത് ടെംപ്ലേറ്റ് എഞ്ചിൻ ഫംഗ്ഷനിലേക്ക് എറിയപ്പെടും.
3. ഞങ്ങളുടെ അഭ്യർത്ഥന തീർക്കുന്നു pandasആരാണ് നമ്മെ നേടുക DataFrame - ഇത് ഭാവിയിൽ ഞങ്ങൾക്ക് ഉപയോഗപ്രദമാകും.

ഞാൻ പകരം വയ്ക്കുന്നത് ഉപയോഗിക്കുന്നു {dt} ഒരു അഭ്യർത്ഥന പരാമീറ്ററിന് പകരം %s ഞാനൊരു ദുഷ്ടനായ പിനോച്ചിയോ ആയതുകൊണ്ടല്ല pandas കൈകാര്യം ചെയ്യാൻ കഴിയില്ല pymssql അവസാനത്തേത് സ്ലിപ്പുചെയ്യുകയും ചെയ്യുന്നു params: Listഅവൻ ശരിക്കും ആഗ്രഹിക്കുന്നുവെങ്കിലും tuple.
ഡെവലപ്പർ എന്നതും ശ്രദ്ധിക്കുക pymssql ഇനി അവനെ പിന്തുണയ്ക്കേണ്ടതില്ലെന്ന് തീരുമാനിച്ചു, പുറത്തുപോകാനുള്ള സമയമാണിത് pyodbc.

എയർഫ്ലോ ഞങ്ങളുടെ ഫംഗ്‌ഷനുകളുടെ ആർഗ്യുമെന്റുകൾ എന്തെല്ലാമായാണ് നിറച്ചതെന്ന് നമുക്ക് നോക്കാം:

ഡാറ്റ ഇല്ലെങ്കിൽ, തുടരുന്നതിൽ അർത്ഥമില്ല. എന്നാൽ പൂരിപ്പിക്കൽ വിജയകരമാണെന്ന് കരുതുന്നതും വിചിത്രമാണ്. എന്നാൽ ഇത് ഒരു തെറ്റല്ല. ആ-ആഹ്, എന്ത് ചെയ്യണം?! കൂടാതെ ഇവിടെ എന്താണ്:

if df.empty:
    raise AirflowSkipException('No rows to load')

AirflowSkipException പിശകുകളൊന്നുമില്ലെന്ന് എയർഫ്ലോയോട് പറയും, പക്ഷേ ഞങ്ങൾ ടാസ്ക് ഒഴിവാക്കുന്നു. ഇന്റർഫേസിന് പച്ചയോ ചുവപ്പോ ചതുരം ഉണ്ടായിരിക്കില്ല, പക്ഷേ പിങ്ക്.

നമുക്ക് നമ്മുടെ ഡാറ്റ ടോസ് ചെയ്യാം ഒന്നിലധികം നിരകൾ:

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

ഇവർ:

• ഞങ്ങൾ ഓർഡറുകൾ സ്വീകരിച്ച ഡാറ്റാബേസ്,
• ഞങ്ങളുടെ വെള്ളപ്പൊക്ക സെഷന്റെ ഐഡി (അത് വ്യത്യസ്തമായിരിക്കും ഓരോ ജോലിക്കും),
• ഉറവിടത്തിൽ നിന്നും ഓർഡർ ഐഡിയിൽ നിന്നും ഒരു ഹാഷ് - അതിനാൽ അന്തിമ ഡാറ്റാബേസിൽ (എല്ലാം ഒരു ടേബിളിലേക്ക് പകരുന്നിടത്ത്) ഞങ്ങൾക്ക് ഒരു അദ്വിതീയ ഓർഡർ ഐഡി ഉണ്ട്.

അവസാന ഘട്ടം അവശേഷിക്കുന്നു: എല്ലാം വെർട്ടിക്കയിലേക്ക് ഒഴിക്കുക. കൂടാതെ, വിചിത്രമെന്നു പറയട്ടെ, ഇത് ചെയ്യാനുള്ള ഏറ്റവും ഗംഭീരവും കാര്യക്ഷമവുമായ മാർഗ്ഗങ്ങളിലൊന്ന് 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-ഉം ഒരു മൗസും ഒപ്പം ഞാനും എയർഫ്ലോയും ... തുടർന്ന് ഞങ്ങൾ അറ്റകുറ്റപ്പണിയുടെ എളുപ്പവും താരതമ്യം ചെയ്യും ... കൊള്ളാം, ഞാൻ അവരെ എല്ലാ മുന്നണികളിലും തോൽപ്പിക്കുമെന്ന് നിങ്ങൾ സമ്മതിക്കുമെന്ന് ഞാൻ കരുതുന്നു!

കുറച്ചുകൂടി ഗൗരവമാണെങ്കിൽ, അപ്പാച്ചെ എയർഫ്ലോ - പ്രോഗ്രാം കോഡിന്റെ രൂപത്തിൽ പ്രക്രിയകൾ വിവരിച്ചുകൊണ്ട് - എന്റെ ജോലി ചെയ്തു വളരെയധികം കൂടുതൽ സുഖകരവും ആസ്വാദ്യകരവുമാണ്.

പ്ലഗ്-ഇന്നുകളുടെയും സ്കേലബിലിറ്റിയുടെ മുൻകരുതലിന്റെയും കാര്യത്തിൽ അതിന്റെ പരിധിയില്ലാത്ത വിപുലീകരണം, മിക്കവാറും ഏത് മേഖലയിലും എയർഫ്ലോ ഉപയോഗിക്കാനുള്ള അവസരം നിങ്ങൾക്ക് നൽകുന്നു: ഡാറ്റ ശേഖരിക്കുന്നതിനും തയ്യാറാക്കുന്നതിനും പ്രോസസ്സ് ചെയ്യുന്നതിനുമുള്ള മുഴുവൻ ചക്രത്തിലും, റോക്കറ്റുകൾ വിക്ഷേപിക്കുമ്പോഴും (ചൊവ്വയിലേക്ക്, കോഴ്സ്).

ഭാഗം അന്തിമവും റഫറൻസും വിവരങ്ങളും

ഞങ്ങൾ നിങ്ങൾക്കായി ശേഖരിച്ച റേക്ക്

• start_date. അതെ, ഇത് ഇതിനകം ഒരു പ്രാദേശിക മെമ്മാണ്. ഡഗിന്റെ പ്രധാന വാദം വഴി start_date എല്ലാം കടന്നുപോകുക. ചുരുക്കത്തിൽ, നിങ്ങൾ വ്യക്തമാക്കുകയാണെങ്കിൽ start_date നിലവിലെ തീയതി, ഒപ്പം schedule_interval - ഒരു ദിവസം, പിന്നെ DAG നാളെ ആരംഭിക്കും.

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

  പിന്നെ പ്രശ്‌നങ്ങളൊന്നുമില്ല.

  ഇതുമായി ബന്ധപ്പെട്ട മറ്റൊരു റൺടൈം പിശക് ഉണ്ട്: Task is missing the start_date parameter, ഇത് മിക്കപ്പോഴും സൂചിപ്പിക്കുന്നത് നിങ്ങൾ ഡാഗ് ഓപ്പറേറ്ററുമായി ബന്ധിപ്പിക്കാൻ മറന്നുവെന്നാണ്.

• എല്ലാം ഒരു മെഷീനിൽ. അതെ, കൂടാതെ ബേസുകളും (എയർഫ്ലോയും ഞങ്ങളുടെ കോട്ടിംഗും), ഒരു വെബ് സെർവറും ഒരു ഷെഡ്യൂളറും തൊഴിലാളികളും. അത് പോലും പ്രവർത്തിച്ചു. എന്നാൽ കാലക്രമേണ, സേവനങ്ങൾക്കായുള്ള ജോലികളുടെ എണ്ണം വർദ്ധിച്ചു, PostgreSQL 20 ms-ന് പകരം 5 സെക്കൻഡിൽ സൂചികയോട് പ്രതികരിക്കാൻ തുടങ്ങിയപ്പോൾ, ഞങ്ങൾ അത് എടുത്ത് കൊണ്ടുപോയി.
• ലോക്കൽ എക്സിക്യൂട്ടർ. അതെ, ഞങ്ങൾ ഇപ്പോഴും അതിൽ ഇരിക്കുകയാണ്, ഞങ്ങൾ ഇതിനകം അഗാധത്തിന്റെ അരികിൽ എത്തിയിരിക്കുന്നു. LocalExecutor ഞങ്ങൾക്ക് ഇതുവരെ മതിയായിരുന്നു, എന്നാൽ ഇപ്പോൾ ഒരു തൊഴിലാളിയെങ്കിലുമായി വികസിപ്പിക്കാനുള്ള സമയമായി, CeleryExecutor-ലേക്ക് മാറാൻ ഞങ്ങൾ കഠിനമായി പരിശ്രമിക്കേണ്ടിവരും. നിങ്ങൾക്ക് ഒരു മെഷീനിൽ ഇത് ഉപയോഗിച്ച് പ്രവർത്തിക്കാൻ കഴിയും എന്ന വസ്തുത കണക്കിലെടുത്ത്, ഒരു സെർവറിൽ പോലും സെലറി ഉപയോഗിക്കുന്നതിൽ നിന്ന് ഒന്നും നിങ്ങളെ തടയുന്നില്ല, അത് "തീർച്ചയായും, ഒരിക്കലും ഉൽപാദനത്തിലേക്ക് പോകില്ല, സത്യസന്ധമായി!"
• ഉപയോഗിക്കാത്തത് അന്തർനിർമ്മിത ഉപകരണങ്ങൾ:
  • കണക്ഷനുകൾ സേവന ക്രെഡൻഷ്യലുകൾ സൂക്ഷിക്കാൻ,
  • SLA മിസ്സുകൾ കൃത്യസമയത്ത് പ്രവർത്തിക്കാത്ത ജോലികളോട് പ്രതികരിക്കാൻ,
  • xcom മെറ്റാഡാറ്റ കൈമാറ്റത്തിനായി (ഞാൻ പറഞ്ഞു മെറ്റാഡാറ്റ!) ഡാഗ് ടാസ്ക്കുകൾക്കിടയിൽ.
• മെയിൽ ദുരുപയോഗം. ശരി, എനിക്ക് എന്ത് പറയാൻ കഴിയും? വീഴ്ച വരുത്തിയ ജോലികളുടെ എല്ലാ ആവർത്തനങ്ങൾക്കും അലേർട്ടുകൾ സജ്ജീകരിച്ചു. ഇപ്പോൾ എന്റെ വർക്ക് Gmail-ൽ എയർഫ്ലോയിൽ നിന്ന് 90k ഇമെയിലുകൾ ഉണ്ട്, കൂടാതെ വെബ് മെയിൽ മൂക്ക് ഒരു സമയം 100-ൽ കൂടുതൽ എടുക്കാനും ഇല്ലാതാക്കാനും വിസമ്മതിക്കുന്നു.

കൂടുതൽ കുഴപ്പങ്ങൾ: Apache Airflow Pitfails

കൂടുതൽ ഓട്ടോമേഷൻ ടൂളുകൾ

കൈകൊണ്ടല്ല, തലകൊണ്ട് കൂടുതൽ പ്രവർത്തിക്കാൻ, എയർഫ്ലോ ഞങ്ങൾക്കായി ഇത് തയ്യാറാക്കിയിട്ടുണ്ട്:

• REST API - അദ്ദേഹത്തിന് ഇപ്പോഴും പരീക്ഷണാത്മക പദവിയുണ്ട്, അത് അവനെ പ്രവർത്തിക്കുന്നതിൽ നിന്ന് തടയുന്നില്ല. ഇത് ഉപയോഗിച്ച്, നിങ്ങൾക്ക് ഡാഗുകളേയും ടാസ്ക്കുകളേയും കുറിച്ചുള്ള വിവരങ്ങൾ മാത്രമല്ല, ഒരു ഡാഗ് നിർത്താനും / ആരംഭിക്കാനും, ഒരു DAG റൺ അല്ലെങ്കിൽ ഒരു പൂൾ സൃഷ്ടിക്കാനും കഴിയും.
• CLI - WebUI വഴി ഉപയോഗിക്കാൻ അസൗകര്യം മാത്രമല്ല, പൊതുവെ ഇല്ലാത്തതുമായ നിരവധി ടൂളുകൾ കമാൻഡ് ലൈനിലൂടെ ലഭ്യമാണ്. ഉദാഹരണത്തിന്:
  • backfill ടാസ്‌ക് സന്ദർഭങ്ങൾ പുനരാരംഭിക്കേണ്ടതുണ്ട്.
    ഉദാഹരണത്തിന്, വിശകലന വിദഗ്ധർ വന്നു പറഞ്ഞു: “സഖാവേ, ജനുവരി 1 മുതൽ 13 വരെയുള്ള ഡാറ്റയിൽ നിങ്ങൾക്ക് അസംബന്ധമുണ്ട്! ശരിയാക്കുക, ശരിയാക്കുക, ശരിയാക്കുക, ശരിയാക്കുക!" നിങ്ങൾ അത്തരമൊരു ഹോബ് ആണ്:

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

  • അടിസ്ഥാന സേവനം: initdb, resetdb, upgradedb, checkdb.
  • run, ഒരു ഇൻസ്‌റ്റൻസ് ടാസ്‌ക് പ്രവർത്തിപ്പിക്കാനും എല്ലാ ഡിപൻഡൻസികളിലും സ്‌കോർ ചെയ്യാനും ഇത് നിങ്ങളെ അനുവദിക്കുന്നു. കൂടാതെ, നിങ്ങൾക്ക് ഇത് വഴി പ്രവർത്തിപ്പിക്കാൻ കഴിയും LocalExecutor, നിങ്ങൾക്ക് ഒരു സെലറി ക്ലസ്റ്റർ ഉണ്ടെങ്കിൽ പോലും.
  • ഏതാണ്ട് ഒരേ കാര്യം ചെയ്യുന്നു test, അടിസ്ഥാനങ്ങളിൽ മാത്രം ഒന്നും എഴുതുന്നില്ല.
  • connections ഷെല്ലിൽ നിന്നുള്ള കണക്ഷനുകൾ വൻതോതിൽ സൃഷ്ടിക്കാൻ അനുവദിക്കുന്നു.
• പൈത്തൺ 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)

• എയർഫ്ലോ മെറ്റാഡാറ്റബേസിലേക്ക് ബന്ധിപ്പിക്കുന്നു. ഇതിലേക്ക് എഴുതാൻ ഞാൻ ശുപാർശ ചെയ്യുന്നില്ല, എന്നാൽ വിവിധ നിർദ്ദിഷ്ട മെട്രിക്കുകൾക്കായി ടാസ്‌ക് സ്റ്റേറ്റുകൾ ലഭിക്കുന്നത് ഏതെങ്കിലും 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-ന്റെ ഇഷ്യൂവിൽ നിന്നുള്ള ആദ്യത്തെ പത്ത് ലിങ്കുകൾ എന്റെ ബുക്ക്‌മാർക്കുകളിൽ നിന്നുള്ള എയർഫ്ലോ ഫോൾഡറിലെ ഉള്ളടക്കങ്ങളാണ്.

• അപ്പാച്ചെ എയർഫ്ലോ ഡോക്യുമെന്റേഷൻ - തീർച്ചയായും, ഞങ്ങൾ ഓഫീസിൽ നിന്ന് ആരംഭിക്കണം. ഡോക്യുമെന്റേഷൻ, എന്നാൽ ആരാണ് നിർദ്ദേശങ്ങൾ വായിക്കുന്നത്?
• മികച്ച രീതികൾ - ശരി, സ്രഷ്‌ടാക്കളിൽ നിന്നുള്ള ശുപാർശകളെങ്കിലും വായിക്കുക.
• എയർഫ്ലോ യുഐ - തുടക്കം തന്നെ: ചിത്രങ്ങളിലെ ഉപയോക്തൃ ഇന്റർഫേസ്
• അപ്പാച്ചെ എയർഫ്ലോയുടെ പ്രധാന ആശയങ്ങൾ മനസ്സിലാക്കുന്നു - നിങ്ങൾക്ക് എന്നിൽ നിന്ന് എന്തെങ്കിലും മനസ്സിലായില്ലെങ്കിൽ (പെട്ടെന്ന്!) അടിസ്ഥാന ആശയങ്ങൾ നന്നായി വിവരിച്ചിരിക്കുന്നു.
• Tianlong's Blog — ഒരു എയർഫ്ലോ സെർവർ/ക്ലസ്റ്റർ എങ്ങനെ നിർമ്മിക്കാം എന്നതിനെക്കുറിച്ചുള്ള ഒരു ഗൈഡ് - ഒരു എയർഫ്ലോ ക്ലസ്റ്റർ സജ്ജീകരിക്കുന്നതിനുള്ള ഒരു ചെറിയ ഗൈഡ്.
• ലിഫ്റ്റിൽ അപ്പാച്ചെ എയർഫ്ലോ പ്രവർത്തിക്കുന്നു - ഒരുപക്ഷേ കൂടുതൽ ഔപചാരികതയും കുറച്ച് ഉദാഹരണങ്ങളും ഒഴികെ ഏതാണ്ട് അതേ രസകരമായ ലേഖനം.
• അപ്പാച്ചെ എയർഫ്ലോ എങ്ങനെയാണ് സെലറി തൊഴിലാളികളിൽ ജോലികൾ വിതരണം ചെയ്യുന്നത് - സെലറിയുമായി ചേർന്ന് പ്രവർത്തിക്കുന്നതിനെക്കുറിച്ച്.
• അപ്പാച്ചെ എയർ ഫ്ലോയിൽ DAG മികച്ച രീതികൾ എഴുതുന്നു - ടാസ്‌ക്കുകളുടെ ഇംപോട്ടൻസി, തീയതിക്ക് പകരം ഐഡി ലോഡിംഗ്, രൂപാന്തരം, ഫയൽ ഘടന, മറ്റ് രസകരമായ കാര്യങ്ങൾ എന്നിവയെക്കുറിച്ച്.
• അപ്പാച്ചെ എയർഫ്ലോയിലെ ഡിപൻഡൻസികൾ കൈകാര്യം ചെയ്യുന്നു - ഞാൻ കടന്നുപോകുമ്പോൾ മാത്രം സൂചിപ്പിച്ച ടാസ്ക്കുകളുടെയും ട്രിഗർ റൂളിന്റെയും ഡിപൻഡൻസികൾ.
• വായുപ്രവാഹം: നിങ്ങളുടെ DAG ഷെഡ്യൂളിന് വളരെ പിന്നിലായിരിക്കുമ്പോൾ - ഷെഡ്യൂളറിലെ ചില "ഉദ്ദേശിച്ച രീതിയിൽ പ്രവർത്തിക്കുന്നത്" എങ്ങനെ മറികടക്കാം, നഷ്ടപ്പെട്ട ഡാറ്റ ലോഡ് ചെയ്യുക, ടാസ്ക്കുകൾക്ക് മുൻഗണന നൽകുക.
• അപ്പാച്ചെ എയർഫ്ലോയ്‌ക്കുള്ള ഉപയോഗപ്രദമായ SQL അന്വേഷണങ്ങൾ — Airflow മെറ്റാഡാറ്റയിലേക്കുള്ള ഉപയോഗപ്രദമായ SQL അന്വേഷണങ്ങൾ.
• അപ്പാച്ചെ എയർഫ്ലോ ഉപയോഗിച്ച് വർക്ക്ഫ്ലോകൾ വികസിപ്പിക്കാൻ ആരംഭിക്കുക - ഒരു ഇഷ്‌ടാനുസൃത സെൻസർ സൃഷ്‌ടിക്കുന്നതിനെക്കുറിച്ച് ഉപയോഗപ്രദമായ ഒരു വിഭാഗമുണ്ട്.
• പ്രെസ്റ്റോയും എയർഫ്ലോയും ഉപയോഗിച്ച് AWS-ൽ Fetchr ഡാറ്റ സയൻസ് ഇൻഫ്രാ നിർമ്മിക്കുന്നു - ഡാറ്റാ സയൻസിനായി AWS-ൽ ഒരു ഇൻഫ്രാസ്ട്രക്ചർ നിർമ്മിക്കുന്നതിനെക്കുറിച്ചുള്ള രസകരമായ ഒരു ചെറിയ കുറിപ്പ്.
• എയർഫ്ലോ ഡിഎജികൾ ഡീബഗ്ഗ് ചെയ്യുമ്പോൾ പരിശോധിക്കേണ്ട 7 സാധാരണ പിശകുകൾ - സാധാരണ തെറ്റുകൾ (ആരെങ്കിലും ഇപ്പോഴും നിർദ്ദേശങ്ങൾ വായിക്കാത്തപ്പോൾ).
• അപ്പാച്ചെ എയർഫ്ലോ ഉപയോഗിച്ച് പാസ്‌വേഡ് സംഭരിക്കുകയും ആക്‌സസ് ചെയ്യുകയും ചെയ്യുക - നിങ്ങൾക്ക് കണക്ഷനുകൾ ഉപയോഗിക്കാമെങ്കിലും, ആളുകൾ പാസ്‌വേഡുകൾ സംഭരിക്കുന്നത് എങ്ങനെയെന്ന് പുഞ്ചിരിക്കുക.
• പൈത്തണിന്റെ സെൻ, അപ്പാച്ചെ എയർഫ്ലോ - ഇംപ്ലിസിറ്റ് ഡിഎജി ഫോർവേഡിംഗ്, ഫംഗ്‌ഷനുകളിലെ സന്ദർഭം ത്രോ ചെയ്യൽ, വീണ്ടും ഡിപൻഡൻസികൾ, കൂടാതെ ടാസ്‌ക് ലോഞ്ചുകൾ ഒഴിവാക്കുന്നതിനെക്കുറിച്ചും.
• വായുസഞ്ചാരം: അധികം അറിയാത്ത നുറുങ്ങുകൾ, തന്ത്രങ്ങൾ, മികച്ച രീതികൾ - ഉപയോഗത്തെക്കുറിച്ച് default arguments и params ടെംപ്ലേറ്റുകളിലും വേരിയബിളുകളിലും കണക്ഷനുകളിലും.
• എയർഫ്ലോ ഷെഡ്യൂളർ പ്രൊഫൈൽ ചെയ്യുന്നു - എയർഫ്ലോ 2.0-ന് പ്ലാനർ എങ്ങനെ തയ്യാറെടുക്കുന്നു എന്നതിനെക്കുറിച്ചുള്ള ഒരു കഥ.
• ഡോക്കർ-കമ്പോസിൽ 3 സെലറി തൊഴിലാളികളുള്ള അപ്പാച്ചെ എയർഫ്ലോ - ഞങ്ങളുടെ ക്ലസ്റ്ററിനെ വിന്യസിക്കുന്നതിനെക്കുറിച്ചുള്ള അൽപ്പം കാലഹരണപ്പെട്ട ലേഖനം docker-compose.
• എയർഫ്ലോ സന്ദർഭം ഉപയോഗിച്ച് 4 ടെംപ്ലേറ്റിംഗ് ടാസ്‌ക്കുകൾ - ടെംപ്ലേറ്റുകളും സന്ദർഭ ഫോർവേഡിംഗും ഉപയോഗിച്ച് ഡൈനാമിക് ടാസ്ക്കുകൾ.
• എയർഫ്ലോയിലെ പിശക് അറിയിപ്പുകൾ - മെയിൽ, സ്ലാക്ക് എന്നിവ വഴിയുള്ള സ്റ്റാൻഡേർഡ്, ഇഷ്‌ടാനുസൃത അറിയിപ്പുകൾ.
• എയർഫ്ലോ വർക്ക്ഷോപ്പ്: ക്രച്ചുകളില്ലാത്ത സങ്കീർണ്ണമായ DAG-കൾ - ബ്രാഞ്ചിംഗ് ജോലികൾ, മാക്രോകൾ, XCom.

ലേഖനത്തിൽ ഉപയോഗിച്ചിരിക്കുന്ന ലിങ്കുകളും:

• മാക്രോ റഫറൻസ് - ടെംപ്ലേറ്റുകളിൽ ഉപയോഗിക്കുന്നതിന് പ്ലെയ്‌സ്‌ഹോൾഡറുകൾ ലഭ്യമാണ്.
• സാധാരണ അപകടങ്ങൾ - വായുപ്രവാഹം - ഡാഗുകൾ സൃഷ്ടിക്കുമ്പോൾ സാധാരണ തെറ്റുകൾ.
• puckel/docker-airflow: Docker Apache Airflow - docker-compose പരീക്ഷണത്തിനും ഡീബഗ്ഗിംഗിനും മറ്റും.
• python-telegram-bot/python-telegram-bot: നിങ്ങൾക്ക് നിരസിക്കാൻ കഴിയാത്ത ഒരു റാപ്പർ ഞങ്ങൾ ഉണ്ടാക്കി - ടെലിഗ്രാം REST API-യ്ക്കുള്ള പൈത്തൺ റാപ്പർ.

അവലംബം: www.habr.com