অ্যাপাচি এয়ারফ্লো: ইটিএলকে আরও সহজ করা

হাই, আমি দিমিত্রি লগভিনেঙ্কো - ভেজেট গ্রুপ অফ কোম্পানির বিশ্লেষণ বিভাগের ডেটা ইঞ্জিনিয়ার।

আমি আপনাকে ETL প্রক্রিয়াগুলি বিকাশের জন্য একটি দুর্দান্ত সরঞ্জাম সম্পর্কে বলব - অ্যাপাচি এয়ারফ্লো। তবে এয়ারফ্লো এতই বহুমুখী এবং বহুমুখী যে আপনি ডেটা প্রবাহের সাথে জড়িত না হলেও এটিকে আরও ঘনিষ্ঠভাবে দেখে নেওয়া উচিত, তবে পর্যায়ক্রমে যে কোনও প্রক্রিয়া চালু করা এবং তাদের সম্পাদন নিরীক্ষণ করার প্রয়োজন রয়েছে।

এবং হ্যাঁ, আমি কেবল বলব না, তবে দেখাব: প্রোগ্রামটিতে প্রচুর কোড, স্ক্রিনশট এবং সুপারিশ রয়েছে।

এয়ারফ্লো / উইকিমিডিয়া কমন্স শব্দটি গুগল করলে আপনি সাধারণত যা দেখতে পান

বিষয়বস্তু সূচি

• ভূমিকা
• প্রধান অংশ, ব্যবহারিক (এবং একটু তাত্ত্বিক)
  • কেন আমরা (এবং আপনি)
  • ক্লাস্টার একত্রিত করা
  • মৌলিক ধারণা
  • আমরা কাজ তৈরি করি
  • ফুল সম্পর্কে একটু
  • আমরা underloaded লোড
  • সংযোগ, হুক এবং অন্যান্য ভেরিয়েবল
  • কাস্টম অপারেটর পার্সিং
  • আপনি ঢালা যাচ্ছেন?
  • বুদ্ধিমান
• পার্ট ফাইনাল, রেফারেন্স এবং তথ্য
  • আমরা আপনার জন্য সংগ্রহ করেছি রেক
  • আরও অটোমেশন টুল
• রেফারেন্স

ভূমিকা

অ্যাপাচি এয়ারফ্লো ঠিক জ্যাঙ্গোর মতো:

• পাইথনে লেখা
• একটি মহান অ্যাডমিন প্যানেল আছে,
• অনির্দিষ্টকালের জন্য প্রসারণযোগ্য

- শুধুমাত্র ভাল, এবং এটি সম্পূর্ণ ভিন্ন উদ্দেশ্যে তৈরি করা হয়েছিল, যথা (যেমন এটি ক্যাটের আগে লেখা আছে):

• সীমাহীন সংখ্যক মেশিনে কাজ চালানো এবং পর্যবেক্ষণ করা (যত বেশি সেলারি/কুবারনেটস এবং আপনার বিবেক আপনাকে অনুমতি দেবে)
• ডায়নামিক ওয়ার্কফ্লো জেনারেশনের সাথে পাইথন কোড লিখতে এবং বুঝতে খুব সহজ থেকে
• এবং রেডিমেড কম্পোনেন্ট এবং ঘরে তৈরি প্লাগইন উভয় ব্যবহার করে যেকোন ডাটাবেস এবং এপিআই একে অপরের সাথে সংযোগ করার ক্ষমতা (যা অত্যন্ত সহজ)।

আমরা এই মত Apache Airflow ব্যবহার করি:

• আমরা বিভিন্ন উৎস থেকে ডেটা সংগ্রহ করি (অনেক SQL সার্ভার এবং PostgreSQL উদাহরণ, অ্যাপ্লিকেশন মেট্রিক্স সহ বিভিন্ন API, এমনকি 1C) DWH এবং ODS (আমাদের Vertica এবং Clickhouse আছে)।
• কত উন্নত cron, যা ওডিএস-এ ডেটা একত্রীকরণ প্রক্রিয়া শুরু করে এবং তাদের রক্ষণাবেক্ষণও পর্যবেক্ষণ করে।

সম্প্রতি অবধি, আমাদের চাহিদাগুলি 32 কোর এবং 50 GB RAM সহ একটি ছোট সার্ভার দ্বারা আচ্ছাদিত ছিল৷ এয়ারফ্লোতে, এটি কাজ করে:

• অধিক 200 ড্যাগ (আসলে ওয়ার্কফ্লো, যেখানে আমরা কাজগুলি স্টাফ করেছি)
• গড়ে প্রতিটিতে 70টি কাজ,
• এই ধার্মিকতা শুরু হয় (গড়েও) ঘন্টায় একবার.

এবং আমরা কীভাবে প্রসারিত করেছি সে সম্পর্কে, আমি নীচে লিখব, কিন্তু এখন আসুন über-সমস্যাটি সংজ্ঞায়িত করি যা আমরা সমাধান করব:

তিনটি উৎস এসকিউএল সার্ভার রয়েছে, প্রতিটিতে 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.

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

মন্তব্য সমূহ:

• রচনাটির সমাবেশে, আমি মূলত সুপরিচিত চিত্রের উপর নির্ভর করেছিলাম পাকেল/ডকার-এয়ারফ্লো - এটি চেক আউট করতে ভুলবেন না. হয়তো তোমার জীবনে আর কিছুর দরকার নেই।
• সমস্ত এয়ারফ্লো সেটিংস না শুধুমাত্র মাধ্যমে উপলব্ধ 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)

আসুন এটি বের করা যাক:

• প্রথমত, আমরা প্রয়োজনীয় libs এবং আমদানি করি অন্যকিছু;
• 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 আমাদের গিটল্যাবে, এবং গিটল্যাব সিআই মেশিনে একত্রিত হওয়ার সময় আপডেটগুলি বিতরণ করে master.

ফুল সম্পর্কে একটু

শ্রমিকরা যখন আমাদের প্রশান্তিকে মারছে, তখন আসুন আরেকটি হাতিয়ার মনে রাখি যা আমাদের কিছু দেখাতে পারে - ফ্লাওয়ার।

কর্মী নোডের সংক্ষিপ্ত তথ্য সহ প্রথম পৃষ্ঠা:

কাজগুলি সহ সবচেয়ে তীব্র পৃষ্ঠা যা কাজ করতে গিয়েছিল:

আমাদের ব্রোকারের স্ট্যাটাস সহ সবচেয়ে বিরক্তিকর পৃষ্ঠা:

সবচেয়ে উজ্জ্বল পৃষ্ঠাটি টাস্ক স্ট্যাটাস গ্রাফ এবং তাদের কার্যকর করার সময় সহ:

আমরা underloaded লোড

সুতরাং, সমস্ত কাজ শেষ হয়েছে, আপনি আহতদের বহন করতে পারেন।

এবং সেখানে অনেক আহত হয়েছিল - এক বা অন্য কারণে। এয়ারফ্লো সঠিক ব্যবহারের ক্ষেত্রে, এই খুব স্কোয়ারগুলি নির্দেশ করে যে ডেটা নিশ্চিতভাবে আসেনি।

আপনাকে লগটি দেখতে হবে এবং পতিত টাস্ক ইনস্ট্যান্সগুলি পুনরায় চালু করতে হবে।

যেকোনো স্কোয়ারে ক্লিক করে, আমরা আমাদের জন্য উপলব্ধ অ্যাকশন দেখতে পাব:

আপনি নিতে পারেন এবং পতিত পরিষ্কার করতে পারেন. অর্থাৎ, আমরা ভুলে যাই যে সেখানে কিছু ব্যর্থ হয়েছে এবং একই উদাহরণ টাস্ক শিডিউলারের কাছে যাবে।

এটা স্পষ্ট যে সমস্ত লাল স্কোয়ারের সাথে মাউস দিয়ে এটি করা খুব মানবিক নয় - এটি আমরা এয়ারফ্লো থেকে আশা করি না। স্বাভাবিকভাবেই, আমাদের কাছে গণবিধ্বংসী অস্ত্র রয়েছে: 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(), যা আমাদের নামে সংযোগ পায়, দেবে এলোমেলো বিভিন্ন নামকরণ থেকে (রাউন্ড রবিন তৈরি করা আরও যৌক্তিক হবে, তবে এটি এয়ারফ্লো বিকাশকারীদের বিবেকের উপর ছেড়ে দেওয়া যাক)।

ভেরিয়েবল এবং সংযোগগুলি অবশ্যই দুর্দান্ত সরঞ্জাম, তবে ভারসাম্য না হারানো গুরুত্বপূর্ণ: আপনার প্রবাহের কোন অংশগুলি আপনি কোডে সঞ্চয় করেন এবং আপনি স্টোরেজের জন্য কোন অংশগুলি এয়ারফ্লোকে দেন৷ একদিকে, দ্রুত মান পরিবর্তন করা সুবিধাজনক হতে পারে, উদাহরণস্বরূপ, একটি মেইলিং বক্স, 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 Airflow কে বলবে যে কোন ত্রুটি নেই, কিন্তু আমরা কাজটি এড়িয়ে যাই। ইন্টারফেসে সবুজ বা লাল বর্গক্ষেত্র থাকবে না, কিন্তু গোলাপী।

আসুন আমাদের ডেটা টস করি একাধিক কলাম:

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 - একদিন, তাহলে ডিএজি কাল শুরু হবে আগে না।

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

  এবং আর কোন সমস্যা নেই।

  এর সাথে যুক্ত আরেকটি রানটাইম ত্রুটি রয়েছে: Task is missing the start_date parameter, যা প্রায়শই নির্দেশ করে যে আপনি dag অপারেটরের সাথে আবদ্ধ করতে ভুলে গেছেন।

• সব এক মেশিনে। হ্যাঁ, এবং ঘাঁটি (এয়ারফ্লো নিজেই এবং আমাদের আবরণ), এবং একটি ওয়েব সার্ভার, এবং একটি সময়সূচী, এবং কর্মীরা। এবং এটা এমনকি কাজ. কিন্তু সময়ের সাথে সাথে, পরিষেবাগুলির জন্য কাজের সংখ্যা বাড়তে থাকে, এবং যখন PostgreSQL 20 ms এর পরিবর্তে 5 সেকেন্ডে সূচকে সাড়া দিতে শুরু করে, তখন আমরা এটি নিয়ে গিয়েছিলাম এবং সরিয়ে নিয়েছিলাম।
• লোকাল এক্সিকিউটর। হ্যাঁ, আমরা এখনও এটির উপর বসে আছি এবং আমরা ইতিমধ্যে অতল গহ্বরের ধারে চলে এসেছি। LocalExecutor এখন পর্যন্ত আমাদের জন্য যথেষ্ট ছিল, কিন্তু এখন সময় এসেছে অন্তত একজন কর্মী নিয়ে প্রসারিত করার, এবং CeleryExecutor-এ যাওয়ার জন্য আমাদের কঠোর পরিশ্রম করতে হবে। এবং আপনি এটির সাথে একটি মেশিনে কাজ করতে পারেন এই বিষয়টির পরিপ্রেক্ষিতে, কোনও সার্ভারেও সেলারি ব্যবহার করতে আপনাকে বাধা দেয় না, যা "অবশ্যই, সততার সাথে কখনও উত্পাদনে যাবে না!"
• অ-ব্যবহার অন্তর্নির্মিত সরঞ্জাম:
  • সংযোগ পরিষেবার শংসাপত্র সংরক্ষণ করতে,
  • SLA মিস যে কাজগুলো সময়মতো হয়নি সেগুলোর জবাব দিতে,
  • xcom মেটাডেটা বিনিময়ের জন্য (আমি বললাম মেটাডেটা!) ডেগ কাজের মধ্যে।
• মেল অপব্যবহার. ওয়েল আমি কি বলতে পারেন? পতিত কাজের পুনরাবৃত্তির জন্য সতর্কতা সেট আপ করা হয়েছিল। এখন আমার কাজ Gmail-এ Airflow থেকে >90k ইমেল আছে, এবং ওয়েব মেইল ​​মজেল একবারে 100 টিরও বেশি পিক আপ করতে এবং মুছতে অস্বীকার করে।

আরো ক্ষতি: Apache Airflow Pitfails

আরও অটোমেশন টুল

আমাদের হাত দিয়ে নয় আমাদের মাথা দিয়ে আরও বেশি কাজ করার জন্য, এয়ারফ্লো আমাদের জন্য এটি প্রস্তুত করেছে:

• বিশ্রাম এপিআই - তার এখনও পরীক্ষামূলক মর্যাদা রয়েছে, যা তাকে কাজ করতে বাধা দেয় না। এটির সাহায্যে, আপনি শুধুমাত্র ড্যাগ এবং কাজ সম্পর্কে তথ্য পেতে পারেন না, তবে একটি ড্যাগ বন্ধ/শুরু করতে, একটি 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 শেল থেকে সংযোগ তৈরি করতে দেয়।
• পাইথন এপিআই - মিথস্ক্রিয়া করার একটি বরং হার্ডকোর উপায়, যা প্লাগইনগুলির জন্য উদ্দিষ্ট, এবং সামান্য হাত দিয়ে এতে ঝাঁকুনি না। কিন্তু আমাদের যেতে বাধা দেবে কে /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

রেফারেন্স

এবং অবশ্যই, গুগল ইস্যু করার প্রথম দশটি লিঙ্ক আমার বুকমার্ক থেকে এয়ারফ্লো ফোল্ডারের বিষয়বস্তু।

• অ্যাপাচি এয়ারফ্লো ডকুমেন্টেশন - অবশ্যই অফিস দিয়ে শুরু করতে হবে। ডকুমেন্টেশন, কিন্তু নির্দেশাবলী কে পড়ে?
• সেরা অনুশীলন - ঠিক আছে, অন্তত নির্মাতাদের কাছ থেকে সুপারিশ পড়ুন।
• এয়ারফ্লো UI - একেবারে শুরু: ছবিতে ইউজার ইন্টারফেস
• Apache Airflow এর মূল ধারণা বোঝা - মৌলিক ধারণাগুলি ভালভাবে বর্ণনা করা হয়েছে, যদি (হঠাৎ!) আপনি আমার কাছ থেকে কিছু বুঝতে না পারেন।
• তিয়ানলং এর ব্লগ — কিভাবে একটি এয়ারফ্লো সার্ভার/ক্লাস্টার তৈরি করতে হয় তার একটি নির্দেশিকা - একটি এয়ারফ্লো ক্লাস্টার সেট আপ করার জন্য একটি সংক্ষিপ্ত নির্দেশিকা।
• Lyft এ Apache এয়ারফ্লো চলছে - প্রায় একই আকর্ষণীয় নিবন্ধ, সম্ভবত আরও আনুষ্ঠানিকতা এবং কম উদাহরণ ছাড়া।
• কিভাবে Apache Airflow সেলারি কর্মীদের চাকরি বিতরণ করে - সেলারির সাথে একযোগে কাজ করার বিষয়ে।
• Apache এয়ারফ্লোতে DAG লেখার সেরা অনুশীলন - কাজের অদম্যতা সম্পর্কে, তারিখের পরিবর্তে আইডি দ্বারা লোড করা, রূপান্তর, ফাইলের কাঠামো এবং অন্যান্য আকর্ষণীয় জিনিস।
• অ্যাপাচি এয়ারফ্লোতে নির্ভরতা পরিচালনা করা - কাজ এবং ট্রিগার নিয়মের নির্ভরতা, যা আমি শুধুমাত্র পাস করার সময় উল্লেখ করেছি।
• বায়ুপ্রবাহ: যখন আপনার DAG সময়সূচীর পিছনে থাকে - কিভাবে শিডিউলারের কিছু "উদ্দেশ্য অনুযায়ী কাজ" কাটিয়ে উঠতে হয়, হারিয়ে যাওয়া ডেটা লোড করতে হয় এবং কাজগুলিকে অগ্রাধিকার দিতে হয়।
• Apache Airflow এর জন্য দরকারী SQL প্রশ্ন — এয়ারফ্লো মেটাডেটাতে দরকারী এসকিউএল প্রশ্ন।
• Apache Airflow এর সাথে ওয়ার্কফ্লো ডেভেলপ করা শুরু করুন - একটি কাস্টম সেন্সর তৈরি সম্পর্কে একটি দরকারী বিভাগ আছে।
• Presto এবং Airflow সহ AWS-এ Fetchr ডেটা সায়েন্স ইনফ্রা তৈরি করা — ডেটা সায়েন্সের জন্য AWS-এ একটি পরিকাঠামো নির্মাণের বিষয়ে একটি আকর্ষণীয় সংক্ষিপ্ত নোট।
• এয়ারফ্লো ডিএজি ডিবাগ করার সময় চেক করার জন্য 7টি সাধারণ ত্রুটি৷ - সাধারণ ভুল (যখন কেউ এখনও নির্দেশাবলী পড়ে না)।
• Apache Airflow ব্যবহার করে পাসওয়ার্ড সংরক্ষণ এবং অ্যাক্সেস করুন - হাসুন লোকেরা কীভাবে পাসওয়ার্ড সংরক্ষণ করে, যদিও আপনি কেবল সংযোগগুলি ব্যবহার করতে পারেন৷
• পাইথন এবং অ্যাপাচি এয়ারফ্লো এর জেন - অন্তর্নিহিত DAG ফরওয়ার্ডিং, ফাংশনে কনটেক্সট থ্রোয়িং, আবার নির্ভরতা সম্পর্কে, এবং টাস্ক লঞ্চ এড়িয়ে যাওয়া সম্পর্কেও।
• বায়ুপ্রবাহ: কম পরিচিত টিপস, কৌশল এবং সর্বোত্তম অনুশীলন - ব্যবহার সম্পর্কে default arguments и params টেমপ্লেট, সেইসাথে ভেরিয়েবল এবং সংযোগগুলিতে।
• এয়ারফ্লো সিডিউলার প্রোফাইলিং - পরিকল্পনাকারী কীভাবে এয়ারফ্লো 2.0 এর জন্য প্রস্তুতি নিচ্ছেন সে সম্পর্কে একটি গল্প।
• ডকার-কম্পোজে 3 সেলারি কর্মীদের সাথে অ্যাপাচি এয়ারফ্লো - আমাদের ক্লাস্টার স্থাপন সম্পর্কে একটি সামান্য পুরানো নিবন্ধ docker-compose.
• বায়ুপ্রবাহের প্রসঙ্গ ব্যবহার করে 4 টেমপ্লেটিং টাস্ক - টেমপ্লেট এবং প্রসঙ্গ ফরওয়ার্ডিং ব্যবহার করে গতিশীল কাজ।
• এয়ারফ্লোতে ত্রুটি বিজ্ঞপ্তি — মেল এবং স্ল্যাকের মাধ্যমে স্ট্যান্ডার্ড এবং কাস্টম বিজ্ঞপ্তি।
• এয়ারফ্লো ওয়ার্কশপ: ক্রাচ ছাড়া জটিল ডিএজি - ব্রাঞ্চিং টাস্ক, ম্যাক্রো এবং এক্সকম।

এবং নিবন্ধে ব্যবহৃত লিঙ্কগুলি:

• ম্যাক্রো রেফারেন্স - টেমপ্লেট ব্যবহারের জন্য স্থানধারক উপলব্ধ।
• সাধারণ ক্ষতি - বায়ুপ্রবাহ - ড্যাগ তৈরি করার সময় সাধারণ ভুল।
• puckel/docker-airflow: Docker Apache Airflow - docker-compose পরীক্ষা, ডিবাগিং এবং আরও অনেক কিছুর জন্য।
• python-telegram-bot/python-telegram-bot: আমরা আপনাকে একটি মোড়ক বানিয়েছি যা আপনি অস্বীকার করতে পারবেন না — টেলিগ্রাম REST API-এর জন্য পাইথন র‍্যাপার।

উত্স: www.habr.com