અપાચે એરફ્લો: ETL ને સરળ બનાવવું

હાય, હું દિમિત્રી લોગવિનેન્કો છું - વેઝેટ જૂથની કંપનીઓના એનાલિટિક્સ વિભાગનો ડેટા એન્જિનિયર.

હું તમને ETL પ્રક્રિયાઓ વિકસાવવા માટે એક અદ્ભુત સાધન વિશે કહીશ - Apache Airflow. પરંતુ એરફ્લો એટલો સર્વતોમુખી અને બહુપક્ષીય છે કે તમે ડેટા પ્રવાહમાં સામેલ ન હોવ તો પણ તમારે તેને નજીકથી જોવું જોઈએ, પરંતુ સમયાંતરે કોઈપણ પ્રક્રિયાઓ શરૂ કરવાની અને તેના અમલ પર દેખરેખ રાખવાની જરૂર છે.

અને હા, હું માત્ર કહીશ નહીં, પણ બતાવીશ: પ્રોગ્રામમાં ઘણા બધા કોડ, સ્ક્રીનશૉટ્સ અને ભલામણો છે.

જ્યારે તમે એરફ્લો / વિકિમીડિયા કોમન્સ શબ્દને ગૂગલ કરો છો ત્યારે તમે સામાન્ય રીતે શું જુઓ છો

સમાવિષ્ટોનું કોષ્ટક

• પરિચય
• મુખ્ય ભાગ, વ્યવહારુ (અને થોડો સૈદ્ધાંતિક)
  • શા માટે અમે (અને તમે)
  • ક્લસ્ટર એસેમ્બલીંગ
  • મૂળભૂત ખ્યાલો
  • અમે કાર્યો જનરેટ કરીએ છીએ
  • ફૂલ વિશે થોડું
  • અમે અન્ડરલોડેડ લોડ કરીએ છીએ
  • જોડાણો, હુક્સ અને અન્ય ચલો
  • કસ્ટમ ઓપરેટરનું પદચ્છેદન
  • શું તમે રેડવાના છો?
  • એકત્ર કરવું
• ભાગ અંતિમ, સંદર્ભ અને માહિતી
  • દાંતી અમે તમારા માટે એકત્રિત કરી છે
  • વધુ ઓટોમેશન સાધનો
• સંદર્ભો

પરિચય

અપાચે એરફ્લો ડીજેંગોની જેમ જ છે:

• અજગરમાં લખેલું
• એક સરસ એડમિન પેનલ છે,
• અનિશ્ચિત સમય માટે વિસ્તૃત

- ફક્ત વધુ સારું, અને તે સંપૂર્ણપણે અલગ હેતુઓ માટે બનાવવામાં આવ્યું હતું, એટલે કે (જેમ કે તે કાતા પહેલા લખાયેલ છે):

• અમર્યાદિત સંખ્યામાં મશીનો પર કાર્ય ચલાવવું અને તેનું નિરીક્ષણ કરવું (જેટલા સેલરી / કુબરનેટ્સ અને તમારો અંતરાત્મા તમને પરવાનગી આપશે)
• પાયથોન કોડ લખવા અને સમજવા માટે ખૂબ જ સરળ થી ડાયનેમિક વર્કફ્લો જનરેશન સાથે
• અને કોઈપણ ડેટાબેસેસ અને API ને તૈયાર ઘટકો અને હોમમેઇડ પ્લગઈન્સ બંનેનો ઉપયોગ કરીને એકબીજા સાથે કનેક્ટ કરવાની ક્ષમતા (જે અત્યંત સરળ છે).

અમે આ રીતે અપાચે એરફ્લોનો ઉપયોગ કરીએ છીએ:

• અમે DWH અને ODS (અમારી પાસે વર્ટીકા અને ક્લિકહાઉસ છે) માં વિવિધ સ્ત્રોતો (ઘણા SQL સર્વર અને PostgreSQL ઉદાહરણો, એપ્લિકેશન મેટ્રિક્સ સાથેના વિવિધ API, 1C પણ) માંથી ડેટા એકત્રિત કરીએ છીએ.
• કેવી રીતે અદ્યતન cron, જે ODS પર ડેટા એકત્રીકરણ પ્રક્રિયાઓ શરૂ કરે છે, અને તેમની જાળવણી પર પણ નજર રાખે છે.

તાજેતરમાં સુધી, અમારી જરૂરિયાતો 32 કોરો અને 50 GB RAM સાથે એક નાના સર્વર દ્વારા આવરી લેવામાં આવી હતી. એરફ્લોમાં, આ કામ કરે છે:

• વધુ 200 ડૅગ્સ (ખરેખર વર્કફ્લો, જેમાં આપણે કાર્યો ભર્યા છે),
• દરેકમાં સરેરાશ 70 કાર્યો,
• આ ભલાઈ શરૂ થાય છે (સરેરાશ પણ) કલાકમાં એકવાર.

અને અમે કેવી રીતે વિસ્તરણ કર્યું તે વિશે, હું નીચે લખીશ, પરંતુ હવે ચાલો über-સમસ્યાને વ્યાખ્યાયિત કરીએ જે આપણે હલ કરીશું:

ત્યાં ત્રણ મૂળ SQL સર્વર છે, દરેકમાં 50 ડેટાબેસેસ છે - અનુક્રમે એક પ્રોજેક્ટના ઉદાહરણો, તેમની પાસે સમાન માળખું છે (લગભગ દરેક જગ્યાએ, મુઆ-હા-હા), જેનો અર્થ છે કે દરેક પાસે ઓર્ડર ટેબલ છે (સદનસીબે, તે સાથેનું ટેબલ નામ કોઈપણ વ્યવસાયમાં દબાણ કરી શકાય છે). અમે સર્વિસ ફીલ્ડ્સ (સોર્સ સર્વર, સોર્સ ડેટાબેઝ, ETL ટાસ્ક આઈડી) ઉમેરીને ડેટા લઈએ છીએ અને નિખાલસપણે તેને વર્ટિકામાં નાખીએ છીએ.

ચાલો જઈએ!

મુખ્ય ભાગ, વ્યવહારુ (અને થોડો સૈદ્ધાંતિક)

શા માટે અમે (અને તમે)

જ્યારે વૃક્ષો મોટા હતા અને હું સાદો હતો SQL- એક રશિયન રિટેલમાં, અમે અમારા માટે ઉપલબ્ધ બે ટૂલ્સનો ઉપયોગ કરીને ETL પ્રક્રિયાઓ ઉર્ફે ડેટા ફ્લો સાથે કૌભાંડ કર્યું છે:

• ઇન્ફોર્મેટિકા પાવર સેન્ટર - એક અત્યંત ફેલાતી સિસ્ટમ, અત્યંત ઉત્પાદક, તેના પોતાના હાર્ડવેર સાથે, તેનું પોતાનું વર્ઝનિંગ. મેં તેની ક્ષમતાઓના 1% ભગવાનને પ્રતિબંધિત કરવાનો ઉપયોગ કર્યો. શા માટે? સારું, સૌ પ્રથમ, આ ઇન્ટરફેસ, ક્યાંક 380 ના દાયકાથી, માનસિક રીતે આપણા પર દબાણ લાવે છે. બીજું, આ કોન્ટ્રાપશન અત્યંત ફેન્સી પ્રક્રિયાઓ, ગુસ્સે ઘટકોના પુનઃઉપયોગ અને અન્ય અત્યંત-મહત્વપૂર્ણ-ઉદ્યોગ-યુક્તિઓ માટે રચાયેલ છે. એરબસ AXNUMX / વર્ષ ની પાંખની જેમ તેની કિંમત છે તે હકીકત વિશે, અમે કંઈ કહીશું નહીં.

  સાવચેત રહો, સ્ક્રીનશૉટ 30 વર્ષથી ઓછી ઉંમરના લોકોને થોડું નુકસાન પહોંચાડી શકે છે

  

• SQL સર્વર એકીકરણ સર્વર - અમે અમારા ઇન્ટ્રા-પ્રોજેક્ટ પ્રવાહમાં આ સાથીનો ઉપયોગ કર્યો. ઠીક છે, હકીકતમાં: અમે પહેલાથી જ SQL સર્વરનો ઉપયોગ કરીએ છીએ, અને તેના ETL સાધનોનો ઉપયોગ ન કરવો તે કોઈક રીતે ગેરવાજબી હશે. તેમાં બધું સારું છે: બંને ઇન્ટરફેસ સુંદર છે, અને પ્રોગ્રેસ રિપોર્ટ્સ... પરંતુ આ માટે અમે સોફ્ટવેર પ્રોડક્ટ્સને પસંદ નથી કરતા, ઓહ, આ માટે નહીં. તેનું સંસ્કરણ dtsx (જે સેવ પર શફલ્ડ નોડ્સ સાથે XML છે) અમે કરી શકીએ છીએ, પરંતુ મુદ્દો શું છે? એક ટાસ્ક પેકેજ બનાવવા વિશે કે જે સેંકડો કોષ્ટકોને એક સર્વરથી બીજા સર્વર પર ખેંચશે? હા, શું સો, તમારી તર્જની આંગળી વીસ ટુકડાઓમાંથી પડી જશે, માઉસ બટન પર ક્લિક કરો. પરંતુ તે ચોક્કસપણે વધુ ફેશનેબલ લાગે છે:

  

અમે ચોક્કસપણે રસ્તાઓ શોધી રહ્યા છીએ. કેસ પણ લગભગ સ્વ-લેખિત SSIS પેકેજ જનરેટર પર આવ્યા ...

…અને પછી મને નવી નોકરી મળી. અને અપાચે એરફ્લો તેના પર મને આગળ નીકળી ગયો.

જ્યારે મને જાણવા મળ્યું કે ETL પ્રક્રિયા વર્ણનો સરળ પાયથોન કોડ છે, ત્યારે મેં આનંદ માટે નૃત્ય કર્યું નહીં. આ રીતે ડેટા સ્ટ્રીમનું વર્ઝન અને ભિન્નતા કરવામાં આવી હતી, અને સેંકડો ડેટાબેઝમાંથી એક જ સ્ટ્રક્ચર સાથે કોષ્ટકોને એક લક્ષ્યમાં રેડવું એ દોઢ કે બે 13” સ્ક્રીનમાં પાયથોન કોડની બાબત બની ગઈ હતી.

ક્લસ્ટર એસેમ્બલીંગ

ચાલો સંપૂર્ણપણે બાલમંદિરની વ્યવસ્થા ન કરીએ, અને અહીં સંપૂર્ણપણે સ્પષ્ટ વસ્તુઓ વિશે વાત ન કરીએ, જેમ કે એરફ્લો, તમારા પસંદ કરેલા ડેટાબેઝ, સેલરી અને ડોક્સમાં વર્ણવેલ અન્ય કેસો ઇન્સ્ટોલ કરવા.

જેથી અમે તરત જ પ્રયોગો શરૂ કરી શકીએ, મેં સ્કેચ બનાવ્યું docker-compose.yml જેમાં:

• ચાલો ખરેખર વધારીએ હવા પ્રવાહ: શેડ્યૂલર, વેબસર્વર. સેલરીના કાર્યોને મોનિટર કરવા માટે ફ્લાવર પણ ત્યાં ફરતું હશે (કારણ કે તે પહેલાથી જ apache/airflow:1.10.10-python3.7, પણ અમને વાંધો નથી)
• પોસ્ટગ્રે એસક્યુએલ, જેમાં એરફ્લો તેની સેવા માહિતી (શેડ્યુલર ડેટા, એક્ઝેક્યુશનના આંકડા, વગેરે) લખશે અને સેલરી પૂર્ણ થયેલા કાર્યોને ચિહ્નિત કરશે;
• Redis, જે સેલરી માટે ટાસ્ક બ્રોકર તરીકે કામ કરશે;
• સેલરી કામદાર, જે કાર્યોના સીધા અમલમાં રોકાયેલ હશે.
• ફોલ્ડરમાં ./dags અમે ડેગ્સના વર્ણન સાથે અમારી ફાઇલો ઉમેરીશું. તેઓ ફ્લાય પર લેવામાં આવશે, તેથી દરેક છીંક પછી સમગ્ર સ્ટેકને જગલ કરવાની જરૂર નથી.

કેટલાક સ્થળોએ, ઉદાહરણોમાંનો કોડ સંપૂર્ણ રીતે બતાવવામાં આવતો નથી (જેથી ટેક્સ્ટમાં ગડબડ ન થાય), પરંતુ ક્યાંક તે પ્રક્રિયામાં સંશોધિત કરવામાં આવે છે. સંપૂર્ણ વર્કિંગ કોડ ઉદાહરણો રીપોઝીટરીમાં મળી શકે છે https://github.com/dm-logv/airflow-tutorial.

ડોકર-કંપોઝ.એમએમએલ

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

નોંધ:

• રચનાની એસેમ્બલીમાં, હું મોટે ભાગે જાણીતી છબી પર આધાર રાખતો હતો પકલ/ડોકર-એરફ્લો - તેને તપાસવાની ખાતરી કરો. કદાચ તમારે તમારા જીવનમાં બીજી કોઈ વસ્તુની જરૂર નથી.
• તમામ એરફ્લો સેટિંગ્સ માત્ર મારફતે જ ઉપલબ્ધ નથી airflow.cfg, પણ પર્યાવરણ ચલો દ્વારા (વિકાસકર્તાઓને આભાર), જેનો મેં દૂષિત રીતે લાભ લીધો.
• સ્વાભાવિક રીતે, તે ઉત્પાદન માટે તૈયાર નથી: મેં ઇરાદાપૂર્વક હૃદયના ધબકારા કન્ટેનર પર મૂક્યા નથી, મને સુરક્ષાની ચિંતા નથી. પરંતુ મેં અમારા પ્રયોગકારો માટે ઓછામાં ઓછું યોગ્ય કર્યું.
• નોંધ કરો કે:
  • ડેગ ફોલ્ડર શેડ્યૂલર અને કામદારો બંને માટે સુલભ હોવું જોઈએ.
  • આ જ બધી તૃતીય-પક્ષ લાઇબ્રેરીઓને લાગુ પડે છે - તે બધા શેડ્યૂલર અને કામદારો સાથે મશીનો પર ઇન્સ્ટોલ કરેલા હોવા જોઈએ.

સારું, હવે તે સરળ છે:

$ docker-compose up --scale worker=3

બધું વધી જાય પછી, તમે વેબ ઈન્ટરફેસ જોઈ શકો છો:

• હવા પ્રવાહ: http://127.0.0.1:8080/admin/
• ફૂલ: http://127.0.0.1:5555/dashboard

મૂળભૂત ખ્યાલો

જો તમે આ બધા "ડેગ્સ" માં કંઈપણ સમજી શક્યા નથી, તો અહીં એક નાનો શબ્દકોશ છે:

• શેડ્યૂલર - એરફ્લોમાં સૌથી મહત્વપૂર્ણ કાકા, જે નિયંત્રિત કરે છે કે રોબોટ્સ સખત મહેનત કરે છે, વ્યક્તિ નહીં: શેડ્યૂલનું નિરીક્ષણ કરે છે, ડેગ અપડેટ કરે છે, કાર્યો શરૂ કરે છે.

  સામાન્ય રીતે, જૂના સંસ્કરણોમાં, તેને મેમરીમાં સમસ્યા હતી (ના, સ્મૃતિ ભ્રંશ નથી, પરંતુ લીક્સ) અને લેગસી પેરામીટર પણ રૂપરેખાઓમાં જ રહ્યું. run_duration - તેનું પુનઃપ્રારંભ અંતરાલ. પરંતુ હવે બધું બરાબર છે.

• ડૅગ (ઉર્ફે "ડેગ") - "નિર્દેશિત એસાયક્લિક ગ્રાફ", પરંતુ આવી વ્યાખ્યા થોડા લોકોને કહેશે, પરંતુ હકીકતમાં તે એકબીજા સાથે ક્રિયાપ્રતિક્રિયા કરતા કાર્યો માટેનું કન્ટેનર છે (નીચે જુઓ) અથવા SSIS માં પેકેજ અને ઇન્ફોર્મેટિકામાં વર્કફ્લોનું એનાલોગ છે. .

  ડેગ્સ ઉપરાંત, હજી પણ સબડેગ્સ હોઈ શકે છે, પરંતુ અમે મોટે ભાગે તે મેળવી શકતા નથી.

• ડીએજી રન - પ્રારંભિક ડેગ, જે તેની પોતાની સોંપેલ છે execution_date. સમાન ડેગના ડાગ્રન્સ સમાંતર રીતે કામ કરી શકે છે (જો તમે તમારા કાર્યોને અસ્પષ્ટ બનાવ્યા છે, અલબત્ત).
• ઑપરેટર ચોક્કસ ક્રિયા કરવા માટે જવાબદાર કોડના ટુકડાઓ છે. ત્યાં ત્રણ પ્રકારના ઓપરેટરો છે:
  • ક્રિયાઅમારા પ્રિયની જેમ PythonOperator, જે કોઈપણ (માન્ય) Python કોડ ચલાવી શકે છે;
  • ટ્રાન્સફર, જે ડેટાને સ્થળેથી બીજા સ્થાને પરિવહન કરે છે, કહે છે, 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

ચાલો એક જ સમયે બધું પસંદ કરીએ અને શૂન્ય પર ફરીથી સેટ કરીએ, સાચી આઇટમ પર ક્લિક કરો:

સફાઈ કર્યા પછી, અમારી ટેક્સીઓ આના જેવી દેખાય છે (તેઓ પહેલેથી જ શેડ્યૂલર દ્વારા તેમને શેડ્યૂલ કરવા માટે રાહ જોઈ રહ્યા છે):

જોડાણો, હુક્સ અને અન્ય ચલો

આગામી ડીએજી જોવાનો સમય આવી ગયો છે, 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={} - ડેગ તેના તમામ ઓપરેટરોને સમાન દલીલો વિતરિત કરી શકે છે;
• to='{{ var.value.all_the_kings_men }}' - ક્ષેત્ર to અમારી પાસે હાર્ડકોડ નહીં હોય, પરંતુ જિન્જાનો ઉપયોગ કરીને ગતિશીલ રીતે જનરેટ કરવામાં આવે છે અને ઇમેઇલ્સની સૂચિ સાથેના ચલ, જે મેં કાળજીપૂર્વક મૂક્યું છે Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS - ઓપરેટર શરૂ કરવા માટેની શરત. અમારા કિસ્સામાં, જો બધી નિર્ભરતાઓ કામ કરી ગઈ હોય તો જ પત્ર બોસને જશે સફળતાપૂર્વક;
• tg_bot_conn_id='tg_main' - દલીલો conn_id અમે જે કનેક્શન 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) - હકીકત એ છે કે એરફ્લો મોડલ્સમાં પ્રકારોની સૂચિ હાર્ડવાયર્ડ છે અને સ્રોત કોડમાં પ્રવેશ્યા વિના વિસ્તૃત કરી શકાતી નથી (જો અચાનક મેં કંઈક Google ન કર્યું હોય, તો કૃપા કરીને મને સુધારો), પરંતુ કંઈપણ અમને ક્રેડિટ મેળવવાથી રોકશે નહીં નામ

તમે સમાન નામ સાથે ઘણા જોડાણો પણ બનાવી શકો છો: આ કિસ્સામાં, પદ્ધતિ BaseHook.get_connection(), જે અમને નામ દ્વારા જોડાણો મેળવે છે, આપશે રેન્ડમ કેટલાક નામોમાંથી (રાઉન્ડ રોબિન બનાવવું વધુ તાર્કિક હશે, પરંતુ ચાલો તેને એરફ્લો વિકાસકર્તાઓના અંતરાત્મા પર છોડી દઈએ).

વેરિયેબલ્સ અને કનેક્શન્સ ચોક્કસપણે સરસ સાધનો છે, પરંતુ સંતુલન ન ગુમાવવું મહત્વપૂર્ણ છે: તમારા પ્રવાહના કયા ભાગો તમે કોડમાં જ સંગ્રહિત કરો છો અને સ્ટોરેજ માટે તમે એરફ્લોને કયા ભાગો આપો છો. એક તરફ, UI દ્વારા, ઉદાહરણ તરીકે, મેઇલિંગ બોક્સ, મૂલ્યને ઝડપથી બદલવું અનુકૂળ હોઈ શકે છે. બીજી બાજુ, આ હજી પણ માઉસ ક્લિક પરનું વળતર છે, જેમાંથી આપણે (હું) છૂટકારો મેળવવા માગતા હતા.

કનેક્શન્સ સાથે કામ કરવું એ કાર્યોમાંનું એક છે હુક્સ. સામાન્ય રીતે, એરફ્લો હુક્સ તેને તૃતીય-પક્ષ સેવાઓ અને પુસ્તકાલયો સાથે કનેક્ટ કરવા માટેના બિંદુઓ છે. દા.ત. JiraHook અમારા માટે જીરા (તમે કાર્યોને આગળ અને પાછળ ખસેડી શકો છો) સાથે સંપર્ક કરવા માટે ક્લાયન્ટ ખોલશે અને તેની મદદથી SambaHook તમે સ્થાનિક ફાઇલ પર દબાણ કરી શકો છો smb-બિંદુ.

કસ્ટમ ઓપરેટરનું પદચ્છેદન

અને અમે તે કેવી રીતે બને છે તે જોવાની નજીક પહોંચી ગયા TelegramBotSendMessage

કોડ commons/operators.py વાસ્તવિક ઓપરેટર સાથે:

from typing import Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

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

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

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

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

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

અહીં, એરફ્લોની અન્ય દરેક વસ્તુની જેમ, બધું ખૂબ જ સરળ છે:

• પાસેથી વારસામાં મળેલ છે 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 - પ્લગઇનમાં, સાર્વજનિક રિપોઝીટરીમાં મૂકો અને તેને ઓપન સોર્સને આપો.

જ્યારે અમે આ બધું અભ્યાસ કરી રહ્યા હતા, ત્યારે અમારા રિપોર્ટ અપડેટ્સ સફળતાપૂર્વક નિષ્ફળ થવામાં વ્યવસ્થાપિત થયા અને મને ચેનલમાં એક ભૂલ સંદેશ મોકલ્યો. હું તપાસ કરીશ કે શું તે ખોટું છે...

અમારા ડોગમાં કંઈક તૂટી ગયું! શું આપણે એવી જ અપેક્ષા રાખતા નથી? બરાબર!

શું તમે રેડવાના છો?

શું તમને લાગે છે કે હું કંઈક ચૂકી ગયો છું? એવું લાગે છે કે તેણે એસક્યુએલ સર્વરથી વર્ટિકામાં ડેટા ટ્રાન્સફર કરવાનું વચન આપ્યું હતું, અને પછી તેણે તે લીધું અને વિષય, બદમાશોને છોડી દીધો!

આ અત્યાચાર ઇરાદાપૂર્વક કરવામાં આવ્યો હતો, મારે તમારા માટે કેટલીક પરિભાષા સમજવાની હતી. હવે તમે વધુ આગળ વધી શકો છો.

અમારી યોજના આ હતી:

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']])

નામ:

• ડેટાબેઝ જેમાંથી અમે ઓર્ડર લીધો હતો,
• અમારા પૂર સત્રનું ID (તે અલગ હશે દરેક કાર્ય માટે),
• સ્ત્રોત અને ઓર્ડર ID માંથી હેશ - જેથી કરીને અંતિમ ડેટાબેઝમાં (જ્યાં બધું એક ટેબલમાં રેડવામાં આવે છે) અમારી પાસે એક અનન્ય ઓર્ડર ID છે.

અંતિમ પગલું બાકી છે: વર્ટિકામાં બધું રેડવું. અને, વિચિત્ર રીતે, આ કરવા માટેની સૌથી અદભૂત અને કાર્યક્ષમ રીતોમાંની એક CSV દ્વારા છે!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

1. અમે એક ખાસ રીસીવર બનાવી રહ્યા છીએ StringIO.
2. pandas કૃપા કરીને અમારા મૂકવામાં આવશે DataFrame ફોર્મમાં CSV-લાઇન્સ.
3. ચાલો હૂક વડે અમારા મનપસંદ વર્ટિકાને કનેક્શન ખોલીએ.
4. અને હવે મદદ સાથે copy() અમારો ડેટા સીધો વર્ટિકામાં મોકલો!

અમે ડ્રાઈવર પાસેથી લઈશું કે કેટલી લાઈનો ભરાઈ છે, અને સેશન મેનેજરને કહીશું કે બધું બરાબર છે:

session.loaded_rows = cursor.rowcount
session.successful = True

બસ એટલું જ.

વેચાણ પર, અમે લક્ષ્ય પ્લેટ જાતે બનાવીએ છીએ. અહીં મેં મારી જાતને એક નાની મશીનની મંજૂરી આપી:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

હું વાપરું છું VerticaOperator() હું ડેટાબેઝ સ્કીમા અને ટેબલ બનાવું છું (જો તે પહેલાથી અસ્તિત્વમાં નથી, અલબત્ત). મુખ્ય વસ્તુ નિર્ભરતાને યોગ્ય રીતે ગોઠવવાનું છે:

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

એકત્ર કરવું

- સારું, - નાના ઉંદરે કહ્યું, - તે હવે નથી
શું તમને ખાતરી છે કે હું જંગલનો સૌથી ભયંકર પ્રાણી છું?

જુલિયા ડોનાલ્ડસન, ધ ગ્રુફાલો

મને લાગે છે કે જો મારા સાથીદારો અને મારી વચ્ચે સ્પર્ધા હોય: કોણ ઝડપથી શરૂઆતથી ETL પ્રક્રિયા બનાવશે અને લોન્ચ કરશે: તેઓ તેમના SSIS અને માઉસ સાથે અને હું એરફ્લો સાથે... અને પછી અમે જાળવણીની સરળતાની પણ તુલના કરીશું ... વાહ, મને લાગે છે કે તમે સંમત થશો કે હું તેમને તમામ મોરચે હરાવીશ!

જો થોડી વધુ ગંભીરતાથી, તો અપાચે એરફ્લો - પ્રોગ્રામ કોડના રૂપમાં પ્રક્રિયાઓનું વર્ણન કરીને - મારું કામ કર્યું ઘણું વધુ આરામદાયક અને આનંદપ્રદ.

તેની અમર્યાદિત એક્સ્ટેન્સિબિલિટી, પ્લગ-ઇન્સની દ્રષ્ટિએ અને માપનીયતાના વલણ બંનેમાં, તમને લગભગ કોઈપણ ક્ષેત્રમાં એરફ્લોનો ઉપયોગ કરવાની તક આપે છે: ડેટા એકત્રિત કરવા, તૈયાર કરવા અને પ્રક્રિયા કરવાના સંપૂર્ણ ચક્રમાં પણ, રોકેટ લોન્ચ કરવામાં પણ (મંગળ પર, કોર્સ).

ભાગ અંતિમ, સંદર્ભ અને માહિતી

દાંતી અમે તમારા માટે એકત્રિત કરી છે

• start_date. હા, આ પહેલેથી જ સ્થાનિક મેમ છે. વાયા ડૉગની મુખ્ય દલીલ start_date બધા પાસ. સંક્ષિપ્તમાં, જો તમે સ્પષ્ટ કરો છો start_date વર્તમાન તારીખ, અને schedule_interval - એક દિવસ, પછી ડીએજી કાલે શરૂ થશે, વહેલું નહીં.

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

  અને કોઈ વધુ સમસ્યા નથી.

  તેની સાથે સંકળાયેલ બીજી રનટાઇમ ભૂલ છે: Task is missing the start_date parameter, જે મોટે ભાગે સૂચવે છે કે તમે ડેગ ઓપરેટર સાથે જોડવાનું ભૂલી ગયા છો.

• બધા એક મશીન પર. હા, અને પાયા (એરફ્લો પોતે અને અમારું કોટિંગ), અને વેબ સર્વર, અને શેડ્યૂલર અને કામદારો. અને તે કામ પણ કર્યું. પરંતુ સમય જતાં, સેવાઓ માટેના કાર્યોની સંખ્યામાં વધારો થતો ગયો, અને જ્યારે PostgreSQL એ 20 ms ને બદલે 5 s માં અનુક્રમણિકાનો પ્રતિસાદ આપવાનું શરૂ કર્યું, ત્યારે અમે તેને લઈ લીધું અને લઈ ગયા.
• LocalExecutor. હા, અમે હજી પણ તેના પર બેઠા છીએ, અને અમે પહેલેથી જ પાતાળની ધાર પર આવી ગયા છીએ. LocalExecutor અત્યાર સુધી અમારા માટે પૂરતું હતું, પરંતુ હવે ઓછામાં ઓછા એક કાર્યકર સાથે વિસ્તરણ કરવાનો સમય આવી ગયો છે અને CeleryExecutor પર જવા માટે અમારે સખત મહેનત કરવી પડશે. અને એ હકીકતને ધ્યાનમાં રાખીને કે તમે તેની સાથે એક મશીન પર કામ કરી શકો છો, સર્વર પર પણ સેલરીનો ઉપયોગ કરવાથી તમને કંઈપણ રોકતું નથી, જે "અલબત્ત, પ્રમાણિકપણે ક્યારેય ઉત્પાદનમાં જશે નહીં!"
• બિન-ઉપયોગ બિલ્ટ-ઇન સાધનો:
  • કનેક્શન્સ સેવા ઓળખપત્રો સંગ્રહવા માટે,
  • SLA ચૂકી જાય છે સમયસર કામ ન કરતા કાર્યોનો જવાબ આપવા માટે,
  • એક્સકોમ મેટાડેટા વિનિમય માટે (મેં કહ્યું મેટાડેટા!) ડેગ કાર્યો વચ્ચે.
• મેલ દુરુપયોગ. સારું, હું શું કહી શકું? ઘટી ગયેલા કાર્યોના તમામ પુનરાવર્તન માટે ચેતવણીઓ ગોઠવવામાં આવી હતી. હવે મારું કાર્ય Gmail માં એરફ્લો તરફથી >90k ઇમેઇલ્સ છે, અને વેબ મેઇલ મઝલ એક સમયે 100 થી વધુ ઉપાડવાનો અને કાઢી નાખવાનો ઇનકાર કરે છે.

વધુ મુશ્કેલીઓ: અપાચે એરફ્લો પિટફેલ્સ

વધુ ઓટોમેશન સાધનો

અમે અમારા હાથથી નહીં પણ અમારા માથાથી વધુ કામ કરવા માટે, એરફ્લોએ અમારા માટે આ તૈયાર કર્યું છે:

• 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 નો ઉપયોગ કરતાં વધુ ઝડપી અને સરળ હોઈ શકે છે.

  ચાલો આપણે કહીએ કે આપણાં બધાં કાર્યો નિરાધાર નથી, પરંતુ તે ક્યારેક પડી શકે છે, અને આ સામાન્ય છે. પરંતુ થોડા અવરોધો પહેલેથી જ શંકાસ્પદ છે, અને તે તપાસવું જરૂરી છે.

  એસક્યુએલ સાવચેત રહો!

  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 ના ઇશ્યુની પ્રથમ દસ લિંક્સ મારા બુકમાર્ક્સમાંથી એરફ્લો ફોલ્ડરની સામગ્રી છે.

• અપાચે એરફ્લો દસ્તાવેજીકરણ - અલબત્ત, આપણે ઓફિસથી શરૂઆત કરવી જોઈએ. દસ્તાવેજીકરણ, પરંતુ સૂચનાઓ કોણ વાંચે છે?
• શ્રેષ્ઠ પ્રયાસો - સારું, ઓછામાં ઓછું સર્જકોની ભલામણો વાંચો.
• એરફ્લો UI - ખૂબ જ શરૂઆત: ચિત્રોમાં વપરાશકર્તા ઇન્ટરફેસ
• અપાચે એરફ્લોના મુખ્ય ખ્યાલોને સમજવું - મૂળભૂત ખ્યાલો સારી રીતે વર્ણવેલ છે, જો (અચાનક!) તમે મારા તરફથી કંઈક સમજી શક્યા નથી.
• તિયાનલોંગનો બ્લોગ — એરફ્લો સર્વર/ક્લસ્ટર કેવી રીતે બનાવવું તેની માર્ગદર્શિકા - એરફ્લો ક્લસ્ટર સેટ કરવા માટે ટૂંકી માર્ગદર્શિકા.
• Lyft પર અપાચે એરફ્લો ચાલી રહ્યો છે - લગભગ સમાન રસપ્રદ લેખ, કદાચ વધુ ઔપચારિકતા અને ઓછા ઉદાહરણો સિવાય.
• અપાચે એરફ્લો સેલરી કામદારો પર નોકરીઓનું વિતરણ કેવી રીતે કરે છે - સેલરી સાથે જોડાણમાં કામ કરવા વિશે.
• અપાચે એરફ્લોમાં DAG લેખન શ્રેષ્ઠ પ્રેક્ટિસ - કાર્યોની ક્ષમતા વિશે, તારીખને બદલે ID દ્વારા લોડ કરવું, પરિવર્તન, ફાઇલ માળખું અને અન્ય રસપ્રદ વસ્તુઓ.
• અપાચે એરફ્લોમાં નિર્ભરતાનું સંચાલન - કાર્યોની અવલંબન અને ટ્રિગર નિયમ, જેનો મેં ફક્ત પાસિંગમાં ઉલ્લેખ કર્યો છે.
• એરફ્લો: જ્યારે તમારું DAG શેડ્યૂલની પાછળ છે - શેડ્યૂલરમાં કેટલાક "ઈચ્છિત કાર્યો" પર કેવી રીતે કાબુ મેળવવો, ખોવાયેલ ડેટા લોડ કરો અને કાર્યોને પ્રાથમિકતા આપો.
• Apache Airflow માટે ઉપયોગી SQL ક્વેરીઝ - એરફ્લો મેટાડેટા માટે ઉપયોગી SQL ક્વેરીઝ.
• Apache Airflow સાથે વર્કફ્લો વિકસાવવાનું શરૂ કરો - કસ્ટમ સેન્સર બનાવવા વિશે એક ઉપયોગી વિભાગ છે.
• પ્રેસ્ટો અને એરફ્લો સાથે AWS પર Fetchr ડેટા સાયન્સ ઇન્ફ્રાનું નિર્માણ - ડેટા સાયન્સ માટે AWS પર ઇન્ફ્રાસ્ટ્રક્ચર બનાવવા વિશે એક રસપ્રદ ટૂંકી નોંધ.
• એરફ્લો ડીએજી ડીબગ કરતી વખતે તપાસવા માટેની 7 સામાન્ય ભૂલો - સામાન્ય ભૂલો (જ્યારે કોઈ હજી પણ સૂચનાઓ વાંચતું નથી).
• Apache Airflow નો ઉપયોગ કરીને પાસવર્ડ સ્ટોર કરો અને ઍક્સેસ કરો - સ્મિત કરો કે લોકો કેવી રીતે પાસવર્ડ્સ સંગ્રહિત કરે છે, જો કે તમે ફક્ત જોડાણોનો ઉપયોગ કરી શકો છો.
• પાયથોન અને અપાચે એરફ્લોનો ઝેન - ગર્ભિત DAG ફોરવર્ડિંગ, ફંક્શનમાં સંદર્ભ ફેંકવું, ફરીથી નિર્ભરતા વિશે, અને કાર્ય લૉન્ચને છોડવા વિશે પણ.
• એરફ્લો: ઓછી જાણીતી ટિપ્સ, યુક્તિઓ અને શ્રેષ્ઠ વ્યવહારો - ઉપયોગ વિશે default arguments и params ટેમ્પલેટ્સમાં, તેમજ વેરિયેબલ્સ અને કનેક્શન્સમાં.
• એરફ્લો શેડ્યૂલરની પ્રોફાઇલિંગ - આયોજક એરફ્લો 2.0 માટે કેવી રીતે તૈયારી કરી રહ્યો છે તે વિશેની વાર્તા.
• ડોકર-કંપોઝમાં 3 સેલરી કામદારો સાથે અપાચે એરફ્લો - માં અમારા ક્લસ્ટરને જમાવવા વિશે થોડો જૂનો લેખ docker-compose.
• એરફ્લો સંદર્ભનો ઉપયોગ કરીને 4 ટેમ્પલેટીંગ કાર્યો - ટેમ્પલેટ્સ અને સંદર્ભ ફોરવર્ડિંગનો ઉપયોગ કરીને ગતિશીલ કાર્યો.
• એરફ્લોમાં ભૂલ સૂચનાઓ - મેઇલ અને સ્લેક દ્વારા માનક અને કસ્ટમ સૂચનાઓ.
• એરફ્લો વર્કશોપ: ક્રેચ વિના જટિલ ડીએજી - શાખાના કાર્યો, મેક્રો અને XCom.

અને લેખમાં વપરાયેલ લિંક્સ:

• મેક્રો સંદર્ભ - નમૂનાઓમાં ઉપયોગ માટે પ્લેસહોલ્ડર્સ ઉપલબ્ધ છે.
• સામાન્ય મુશ્કેલીઓ - એરફ્લો - ડેગ બનાવતી વખતે સામાન્ય ભૂલો.
• puckel/docker-airflow: Docker Apache Airflow - docker-compose પ્રયોગો, ડીબગીંગ અને વધુ માટે.
• python-telegram-bot/python-telegram-bot: અમે તમને એક આવરણ બનાવ્યું છે જેને તમે ના પાડી શકો - ટેલિગ્રામ REST API માટે પાયથોન રેપર.

સોર્સ: www.habr.com