Airflow არის ინსტრუმენტი მოხერხებულად და სწრაფად, სერიული მონაცემთა დამუშავების პროცესების განვითარებისა და შენარჩუნებისთვის

გამარჯობა, ჰაბრ! ამ სტატიაში მინდა ვისაუბრო ერთ დიდ ინსტრუმენტზე სერიული მონაცემთა დამუშავების პროცესების განვითარებისთვის, მაგალითად, კორპორატიული DWH-ის ან თქვენი DataLake-ის ინფრასტრუქტურაში. ჩვენ ვისაუბრებთ Apache Airflow-ზე (შემდგომში Airflow). ის უსამართლოდ მოკლებულია ყურადღებას Habré-ზე და ძირითად ნაწილში ვეცდები დაგარწმუნოთ, რომ მინიმუმ Airflow ღირს ყურება თქვენი ETL/ELT პროცესების განრიგის არჩევისას.

ადრე დავწერე სტატიების სერია DWH-ის თემაზე, როდესაც ვმუშაობდი Tinkoff Bank-ში. ახლა მე გავხდი Mail.Ru ჯგუფის გუნდის წევრი და ვავითარებ პლატფორმას მონაცემთა ანალიზისთვის სათამაშო ზონაში. რეალურად, როგორც ახალი ამბები და საინტერესო გადაწყვეტილებები გამოჩნდება, მე და ჩემი გუნდი აქ ვისაუბრებთ მონაცემთა ანალიტიკის ჩვენს პლატფორმაზე.

პროლოგი

მაშ ასე, დავიწყოთ. რა არის ჰაერის ნაკადი? ეს არის ბიბლიოთეკა (ან ბიბლიოთეკების ნაკრები) სამუშაო პროცესების შემუშავება, დაგეგმვა და მონიტორინგი. Airflow-ის მთავარი მახასიათებელი: პითონის კოდი გამოიყენება პროცესების აღსაწერად (განვითარებისთვის). ამას ბევრი უპირატესობა აქვს თქვენი პროექტის ორგანიზებისა და განვითარებისთვის: არსებითად, თქვენი (მაგალითად) ETL პროექტი მხოლოდ პითონის პროექტია და შეგიძლიათ მისი ორგანიზება ისე, როგორც გსურთ, ინფრასტრუქტურის სპეციფიკის, გუნდის ზომისა და მახასიათებლების გათვალისწინებით. სხვა მოთხოვნები. ინსტრუმენტულად ყველაფერი მარტივია. გამოიყენეთ მაგალითად PyCharm + Git. მშვენიერია და ძალიან მოსახერხებელი!

ახლა მოდით გადავხედოთ Airflow-ის ძირითად ერთეულებს. მათი არსის და მიზნის გაგებით, თქვენ შეგიძლიათ ოპტიმალურად მოაწყოთ თქვენი პროცესის არქიტექტურა. შესაძლოა, მთავარი ერთეული არის მიმართული აციკლური გრაფიკი (შემდგომში DAG).

DAG

DAG არის თქვენი ამოცანების გარკვეული მნიშვნელოვანი ასოციაცია, რომელიც გსურთ შეასრულოთ მკაცრად განსაზღვრული თანმიმდევრობით კონკრეტული გრაფიკის მიხედვით. Airflow უზრუნველყოფს მოსახერხებელ ვებ ინტერფეისს DAG-ებთან და სხვა ობიექტებთან მუშაობისთვის:

DAG შეიძლება ასე გამოიყურებოდეს:

დეველოპერი, DAG-ის დიზაინის შექმნისას, აყალიბებს ოპერატორების ერთობლიობას, რომლებზეც აშენდება დავალებები DAG-ის ფარგლებში. აქ მივედით კიდევ ერთ მნიშვნელოვან არსებამდე: ჰაერის ნაკადის ოპერატორი.

ოპერატორები

ოპერატორი არის ერთეული, რომლის საფუძველზეც იქმნება სამუშაო ინსტანციები, რომელიც აღწერს რა მოხდება სამუშაო ინსტანციის შესრულების დროს. ჰაერის ნაკადი გამოდის GitHub-დან უკვე შეიცავს გამოსაყენებლად მზად ოპერატორების კომპლექტს. მაგალითები:

• BashOperator - ოპერატორი bash ბრძანების შესასრულებლად.
• PythonOperator - ოპერატორი Python კოდის გამოძახებისთვის.
• EmailOperator — ოპერატორი ელექტრონული ფოსტის გაგზავნისთვის.
• HTTPOoperator - ოპერატორი http მოთხოვნებთან მუშაობისთვის.
• SqlOperator - SQL კოდის შემსრულებელი ოპერატორი.
• სენსორი არის მოვლენის მოლოდინის ოპერატორი (საჭირო დროის დადგომა, საჭირო ფაილის გამოჩენა, ხაზი მონაცემთა ბაზაში, პასუხი API-დან და ა.შ. და ა.შ.).

არსებობს უფრო კონკრეტული ოპერატორები: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

თქვენ ასევე შეგიძლიათ განავითაროთ ოპერატორები საკუთარი მახასიათებლების საფუძველზე და გამოიყენოთ ისინი თქვენს პროექტში. მაგალითად, ჩვენ შევქმენით MongoDBToHiveViaHdfsTransfer, ოპერატორი MongoDB-დან Hive-ში დოკუმენტების ექსპორტისთვის და რამდენიმე ოპერატორი, რომელთანაც მუშაობს. clickhouse: CHLoadFromHiveOperator და CHTableLoaderOperator. არსებითად, როგორც კი პროექტი ხშირად გამოიყენებს ძირითად განცხადებებზე აგებულ კოდს, შეგიძლიათ იფიქროთ მის ახალ განცხადებაში გადაკეთებაზე. ეს გაამარტივებს შემდგომ განვითარებას და თქვენ გააფართოვებთ თქვენს ოპერატორთა ბიბლიოთეკას პროექტში.

შემდეგი, დავალებების ყველა ეს შემთხვევა უნდა შესრულდეს და ახლა ჩვენ ვისაუბრებთ განრიგის შესახებ.

განრიგი

Airflow-ის ამოცანების განრიგი აგებულია ნიახური. Celery არის პითონის ბიბლიოთეკა, რომელიც საშუალებას გაძლევთ მოაწყოთ რიგი და ამოცანების ასინქრონული და განაწილებული შესრულება. ჰაერის ნაკადის მხრივ, ყველა დავალება დაყოფილია აუზებად. აუზები იქმნება ხელით. როგორც წესი, მათი მიზანია წყაროსთან მუშაობის დატვირთვის შეზღუდვა ან DWH-ის ფარგლებში ამოცანების ტიპირება. აუზების მართვა შესაძლებელია ვებ ინტერფეისის საშუალებით:

თითოეულ აუზს აქვს ლიმიტი სლოტების რაოდენობაზე. DAG-ის შექმნისას მას ეძლევა აუზი:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

DAG-ის დონეზე განსაზღვრული აუზი შეიძლება გადაილახოს დავალების დონეზე.
ცალკე პროცესი, Scheduler, პასუხისმგებელია Airflow-ში ყველა დავალების დაგეგმვაზე. სინამდვილეში, Scheduler ეხება შესრულებისთვის ამოცანების დაყენების ყველა მექანიზმს. დავალება შესრულებამდე გადის რამდენიმე ეტაპს:

1. წინა დავალებები შესრულებულია DAG-ში, ახალი შეიძლება დადგეს რიგში.
2. რიგი დალაგებულია ამოცანების პრიორიტეტის მიხედვით (პრიორიტეტების კონტროლიც შესაძლებელია), ხოლო თუ აუზში თავისუფალი სლოტია, დავალება შეიძლება ამოქმედდეს.
3. თუ არის თავისუფალი მუშა ნიახური, დავალება ეგზავნება მას; მუშაობა, რომელიც თქვენ დაპროგრამეთ პრობლემაში, იწყება ამა თუ იმ ოპერატორის გამოყენებით.

საკმარისად მარტივი.

Scheduler მუშაობს ყველა DAG-ის კომპლექტზე და ყველა დავალება DAG-ებში.

იმისათვის, რომ Scheduler-მა დაიწყოს მუშაობა DAG-თან, DAG-მა უნდა დაადგინოს განრიგი:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

არსებობს მზა წინასწარ პარამეტრების ნაკრები: @once, @hourly, @daily, @weekly, @monthly, @yearly.

თქვენ ასევე შეგიძლიათ გამოიყენოთ cron გამონათქვამები:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

აღსრულების თარიღი

იმის გასაგებად, თუ როგორ მუშაობს ჰაერის ნაკადი, მნიშვნელოვანია გვესმოდეს, თუ რა არის შესრულების თარიღი DAG-ისთვის. Airflow-ში DAG-ს აქვს შესრულების თარიღის განზომილება, ანუ, DAG-ის სამუშაო გრაფიკიდან გამომდინარე, თითოეული შესრულების თარიღისთვის იქმნება დავალების ინსტანციები. და თითოეული შესრულების თარიღისთვის, ამოცანები შეიძლება ხელახლა შესრულდეს - ან, მაგალითად, DAG-ს შეუძლია ერთდროულად იმუშაოს რამდენიმე შესრულების თარიღში. ეს ნათლად არის ნაჩვენები აქ:

სამწუხაროდ (ან შესაძლოა საბედნიეროდ: ეს დამოკიდებულია სიტუაციაზე), თუ დავალების შესრულება DAG-ში შესწორებულია, მაშინ შესრულება წინა შესრულების თარიღში გაგრძელდება კორექტირების გათვალისწინებით. ეს კარგია, თუ საჭიროა ახალი ალგორითმის გამოყენებით გასულ პერიოდებში მონაცემების ხელახლა გამოთვლა, მაგრამ ცუდია, რადგან შედეგის განმეორებადობა დაკარგულია (რა თქმა უნდა, არავინ შეგაწუხებთ წყაროს კოდის საჭირო ვერსიის დაბრუნებას Git-დან და გამოთვალოთ რა გჭირდებათ ერთჯერადი, ისე, როგორც გჭირდებათ).

ამოცანების გენერირება

DAG-ის დანერგვა არის კოდი Python-ში, ამიტომ ჩვენ გვაქვს ძალიან მოსახერხებელი გზა, რომ შევამციროთ კოდის რაოდენობა, მაგალითად, გახეხილი წყაროებით მუშაობისას. ვთქვათ, თქვენ გაქვთ სამი MySQL ფრაგმენტი, როგორც წყარო, თქვენ უნდა შეხვიდეთ თითოეულში და აიღოთ გარკვეული მონაცემები. თანაც დამოუკიდებლად და პარალელურად. Python კოდი DAG-ში შეიძლება ასე გამოიყურებოდეს:

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

DAG ასე გამოიყურება:

ამ შემთხვევაში, თქვენ შეგიძლიათ დაამატოთ ან ამოიღოთ ნატეხი უბრალოდ პარამეტრების რეგულირებით და DAG-ის განახლებით. კომფორტული!

თქვენ ასევე შეგიძლიათ გამოიყენოთ უფრო რთული კოდის გენერაცია, მაგალითად, წყაროებთან მუშაობა მონაცემთა ბაზის სახით ან აღწეროთ ცხრილის სტრუქტურა, ცხრილთან მუშაობის ალგორითმი და, DWH ინფრასტრუქტურის მახასიათებლების გათვალისწინებით, წარმოქმნათ პროცესი. N ცხრილის თქვენს საცავში ჩასატვირთად. ან, მაგალითად, API-სთან მუშაობა, რომელიც არ უჭერს მხარს პარამეტრთან მუშაობას სიის სახით, შეგიძლიათ შექმნათ N ამოცანები DAG-ში ამ სიიდან, შეზღუდოთ API-ში მოთხოვნის პარალელურობა აუზამდე და გადაფურცლეთ. საჭირო მონაცემები API-დან. მოქნილი!

საცავი

Airflow-ს აქვს საკუთარი backend საცავი, მონაცემთა ბაზა (შეიძლება იყოს MySQL ან Postgres, ჩვენ გვაქვს Postgres), რომელიც ინახავს ამოცანების მდგომარეობას, DAG-ებს, კავშირის პარამეტრებს, გლობალურ ცვლადებს და ა.შ. და ა.შ. აქ მინდა ვთქვა, რომ Airflow-ის საცავი ძალიან მარტივია (დაახლოებით 20 ცხრილი) და მოსახერხებელია, თუ გსურთ მასზე საკუთარი პროცესის შექმნა. მახსოვს 100500 ცხრილი Informatica-ს საცავში, რომლებიც დიდი ხნის განმავლობაში უნდა შესწავლილიყო, სანამ გავიგებდი, თუ როგორ უნდა აეწყო შეკითხვა.

მონიტორინგი

საცავის სიმარტივის გათვალისწინებით, თქვენ შეგიძლიათ შექმნათ ამოცანების მონიტორინგის პროცესი, რომელიც თქვენთვის მოსახერხებელია. ჩვენ ვიყენებთ რვეულს Zeppelin-ში, სადაც ვათვალიერებთ დავალებების სტატუსს:

ეს ასევე შეიძლება იყოს თავად Airflow-ის ვებ ინტერფეისი:

Airflow კოდი ღია წყაროა, ამიტომ ჩვენ დავამატეთ გაფრთხილება Telegram-ში. დავალების ყოველი გაშვებული მაგალითი, შეცდომის შემთხვევაში, აგზავნის ჯგუფს Telegram-ში, სადაც შედგება განვითარებისა და მხარდაჭერის მთელი გუნდი.

ჩვენ ვიღებთ სწრაფ პასუხს Telegram-ის საშუალებით (საჭიროების შემთხვევაში) და Zeppelin-ის მეშვეობით ვიღებთ ამოცანების საერთო სურათს Airflow-ში.

საერთო ჯამში

ჰაერის ნაკადი, პირველ რიგში, ღია წყაროა და მისგან სასწაულებს არ უნდა ელოდოთ. მზად იყავით დახარჯოთ დრო და ძალისხმევა ეფექტური გადაწყვეტის შესაქმნელად. მიზანი მიღწევადია, დამიჯერე, ღირს. განვითარების სიჩქარე, მოქნილობა, ახალი პროცესების დამატების სიმარტივე - მოგეწონებათ. რა თქმა უნდა, დიდი ყურადღება უნდა მიაქციოთ პროექტის ორგანიზებას, თავად საჰაერო ნაკადის სტაბილურობას: სასწაულები არ ხდება.

ახლა ჩვენ ყოველდღიურად მუშაობს Airflow დაახლოებით 6,5 ათასი დავალება. ისინი საკმაოდ განსხვავდებიან ხასიათით. არსებობს ძირითადი DWH-ში მონაცემების ჩატვირთვა მრავალი განსხვავებული და ძალიან კონკრეტული წყაროდან, არის ვიტრინების გამოთვლის ამოცანები მთავარი DWH-ში, არის მონაცემთა გამოქვეყნების ამოცანები სწრაფ DWH-ში, არის ბევრი, ბევრი განსხვავებული დავალება - და Airflow ღეჭავს მათ დღითი დღე. ციფრებით რომ ვთქვათ, ეს არის 2,3 ათასი სხვადასხვა სირთულის ELT ამოცანები DWH (Hadoop) ფარგლებში, დაახლ. 2,5 ასეული მონაცემთა ბაზა წყაროები, ეს არის გუნდი 4 ETL დეველოპერი, რომლებიც იყოფა ETL მონაცემთა დამუშავებად DWH-ში და ELT მონაცემთა დამუშავებად DWH-ში და რა თქმა უნდა სხვა ერთი ადმინი, რომელიც ეხება სერვისის ინფრასტრუქტურას.

სამომავლო გეგმები

პროცესების რაოდენობა აუცილებლად იზრდება და მთავარი, რასაც Airflow ინფრასტრუქტურის კუთხით გავაკეთებთ, არის მასშტაბირება. ჩვენ გვინდა ავაშენოთ Airflow კლასტერი, გამოვყოთ წყვილი ფეხი სელერიის მუშაკებისთვის და შევქმნათ თვითგამრავლებადი თავი სამუშაოს დაგეგმვის პროცესებით და საცავი.

ეპილოგი

ეს, რა თქმა უნდა, არ არის ყველაფერი, რისი თქმაც მსურს Airflow-ის შესახებ, მაგრამ შევეცადე გამოვყო ძირითადი პუნქტები. მადა ჭამასთან ერთად მოდის, სცადეთ და მოგეწონებათ :)

წყარო: www.habr.com