Airflow គឺជាឧបករណ៍សម្រាប់បង្កើត និងថែរក្សាដំណើរការដំណើរការទិន្នន័យជាបាច់យ៉ាងងាយស្រួល និងឆាប់រហ័ស

សួស្តី ហាប! នៅក្នុងអត្ថបទនេះ ខ្ញុំចង់និយាយអំពីឧបករណ៍ដ៏អស្ចារ្យមួយសម្រាប់បង្កើតដំណើរការដំណើរការទិន្នន័យជាក្រុម ឧទាហរណ៍នៅក្នុងហេដ្ឋារចនាសម្ព័ន្ធនៃក្រុមហ៊ុន DWH ឬ DataLake របស់អ្នក។ យើងនឹងនិយាយអំពី Apache Airflow (តទៅនេះហៅថា Airflow)។ វាត្រូវបានដកហូតដោយអយុត្តិធម៌ចំពោះការយកចិត្តទុកដាក់លើHabre ហើយនៅក្នុងផ្នែកសំខាន់ ខ្ញុំនឹងព្យាយាមបញ្ចុះបញ្ចូលអ្នកថា យ៉ាងហោចណាស់ Airflow គឺមានតម្លៃមើលនៅពេលជ្រើសរើសកម្មវិធីកំណត់ពេលសម្រាប់ដំណើរការ ETL/ELT របស់អ្នក។

ពីមុនខ្ញុំបានសរសេរអត្ថបទជាបន្តបន្ទាប់លើប្រធានបទ DWH នៅពេលដែលខ្ញុំធ្វើការនៅធនាគារ Tinkoff ។ ឥឡូវនេះខ្ញុំបានក្លាយជាផ្នែកមួយនៃក្រុម Mail.Ru Group ហើយកំពុងបង្កើតវេទិកាសម្រាប់ការវិភាគទិន្នន័យនៅក្នុងតំបន់ហ្គេម។ ជាការពិតណាស់ នៅពេលដែលព័ត៌មាន និងដំណោះស្រាយគួរឱ្យចាប់អារម្មណ៍លេចឡើង ក្រុមរបស់ខ្ញុំ និងខ្ញុំនឹងនិយាយនៅទីនេះអំពីវេទិការបស់យើងសម្រាប់ការវិភាគទិន្នន័យ។

Prologue

ដូច្នេះសូមចាប់ផ្តើម។ តើលំហូរខ្យល់គឺជាអ្វី? នេះគឺជាបណ្ណាល័យ (ឬ សំណុំនៃបណ្ណាល័យ) ដើម្បីអភិវឌ្ឍ រៀបចំផែនការ និងត្រួតពិនិត្យដំណើរការការងារ។ លក្ខណៈពិសេសចម្បងនៃលំហូរខ្យល់៖ កូដ Python ត្រូវបានប្រើដើម្បីពិពណ៌នា (អភិវឌ្ឍ) ដំណើរការ។ វាមានអត្ថប្រយោជន៍ជាច្រើនសម្រាប់ការរៀបចំគម្រោង និងការអភិវឌ្ឍន៍របស់អ្នក៖ ជាសំខាន់ គម្រោង ETL របស់អ្នកគ្រាន់តែជាគម្រោង Python ហើយអ្នកអាចរៀបចំវាបានតាមដែលអ្នកចង់បាន ដោយគិតគូរពីភាពជាក់លាក់នៃហេដ្ឋារចនាសម្ព័ន្ធ ទំហំក្រុម និង តម្រូវការផ្សេងទៀត។ ឧបករណ៍ទាំងអស់គឺសាមញ្ញ។ ប្រើឧទាហរណ៍ PyCharm + Git ។ វាពិតជាអស្ចារ្យ និងងាយស្រួលណាស់!

ឥឡូវនេះសូមក្រឡេកមើលធាតុសំខាន់ៗនៃ Airflow ។ តាមរយៈការយល់ដឹងពីខ្លឹមសារ និងគោលបំណងរបស់ពួកគេ អ្នកអាចរៀបចំស្ថាបត្យកម្មដំណើរការរបស់អ្នកបានយ៉ាងល្អប្រសើរ។ ប្រហែលជាអង្គភាពសំខាន់គឺ Directed Acyclic Graph (តទៅនេះហៅថា DAG)។

DAG

DAG គឺជាទំនាក់ទំនងដ៏មានអត្ថន័យនៃកិច្ចការរបស់អ្នក ដែលអ្នកចង់បំពេញតាមលំដាប់ដែលបានកំណត់យ៉ាងតឹងរ៉ឹងយោងទៅតាមកាលវិភាគជាក់លាក់មួយ។ Airflow ផ្តល់នូវចំណុចប្រទាក់បណ្ដាញងាយស្រួលសម្រាប់ធ្វើការជាមួយ DAGs និងអង្គភាពផ្សេងទៀត៖

DAG អាចមើលទៅដូចនេះ៖

អ្នកអភិវឌ្ឍន៍ នៅពេលរចនា DAG បានដាក់សំណុំនៃប្រតិបត្តិករដែលភារកិច្ចនៅក្នុង DAG នឹងត្រូវបានសាងសង់។ នៅទីនេះយើងមកដល់អង្គភាពសំខាន់មួយទៀត: ប្រតិបត្តិករលំហូរខ្យល់។

ប្រតិបត្តិករ

ប្រតិបត្តិករគឺជាអង្គភាពមួយនៅលើមូលដ្ឋាននៃករណីការងារត្រូវបានបង្កើត ដែលពិពណ៌នាអំពីអ្វីដែលនឹងកើតឡើងក្នុងអំឡុងពេលអនុវត្តការងារ។ លំហូរខ្យល់ចេញពី GitHub មានសំណុំនៃប្រតិបត្តិករដែលត្រៀមរួចជាស្រេចក្នុងការប្រើប្រាស់។ ឧទាហរណ៍:

• BashOperator - ប្រតិបត្តិករសម្រាប់ប្រតិបត្តិពាក្យបញ្ជា bash ។
• PythonOperator - ប្រតិបត្តិករសម្រាប់ការហៅកូដ Python ។
• EmailOperator - ប្រតិបត្តិករសម្រាប់ការផ្ញើអ៊ីមែល។
• HTTPOperator - ប្រតិបត្តិករសម្រាប់ធ្វើការជាមួយសំណើ http ។
• SqlOperator - ប្រតិបត្តិករសម្រាប់ប្រតិបត្តិកូដ SQL ។
• ឧបករណ៍ចាប់សញ្ញាគឺជាប្រតិបត្តិករសម្រាប់រង់ចាំព្រឹត្តិការណ៍មួយ (ការមកដល់នៃពេលវេលាដែលត្រូវការ រូបរាងនៃឯកសារដែលត្រូវការ បន្ទាត់នៅក្នុងមូលដ្ឋានទិន្នន័យ ការឆ្លើយតបពី API ជាដើម)។

មានប្រតិបត្តិករជាក់លាក់បន្ថែមទៀត៖ DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator ។

អ្នកក៏អាចអភិវឌ្ឍប្រតិបត្តិករដោយផ្អែកលើលក្ខណៈផ្ទាល់ខ្លួនរបស់អ្នក ហើយប្រើពួកវានៅក្នុងគម្រោងរបស់អ្នក។ ឧទាហរណ៍ យើងបានបង្កើត MongoDBToHiveViaHdfsTransfer ដែលជាប្រតិបត្តិករសម្រាប់ការនាំចេញឯកសារពី MongoDB ទៅ Hive និងប្រតិបត្តិករជាច្រើនសម្រាប់ធ្វើការជាមួយ ចុចផ្ទះ៖ CHLoadFromHiveOperator និង CHTableLoaderOperator ។ ជាការសំខាន់ ដរាបណាគម្រោងមួយបានប្រើជាញឹកញាប់កូដដែលបង្កើតឡើងនៅលើសេចក្តីថ្លែងការណ៍មូលដ្ឋាន អ្នកអាចគិតអំពីការកសាងវាទៅជាសេចក្តីថ្លែងការណ៍ថ្មីមួយ។ វានឹងជួយសម្រួលដល់ការអភិវឌ្ឍន៍បន្ថែមទៀត ហើយអ្នកនឹងពង្រីកបណ្ណាល័យប្រតិបត្តិកររបស់អ្នកនៅក្នុងគម្រោង។

បន្ទាប់មក កិច្ចការទាំងអស់នេះចាំបាច់ត្រូវអនុវត្ត ហើយឥឡូវនេះយើងនឹងនិយាយអំពីកម្មវិធីកំណត់ពេល។

អ្នករៀបចំកាលវិភាគ

កម្មវិធីកំណត់ពេលភារកិច្ចរបស់ Airflow ត្រូវបានបង្កើតឡើងនៅលើ សាឡី. Celery គឺជាបណ្ណាល័យ Python ដែលអនុញ្ញាតឱ្យអ្នករៀបចំជួរ បូករួមទាំងការប្រតិបត្តិកិច្ចការដែលមិនសមកាលកម្ម និងការចែកចាយ។ នៅផ្នែក Airflow កិច្ចការទាំងអស់ត្រូវបានបែងចែកទៅជាថ្លុក។ អាងទឹកត្រូវបានបង្កើតដោយដៃ។ ជាធម្មតា គោលបំណងរបស់ពួកគេគឺកំណត់បន្ទុកការងារជាមួយប្រភព ឬដើម្បីវាយបញ្ចូលភារកិច្ចនៅក្នុង 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. ប្រសិនបើមាន celery កម្មករដោយឥតគិតថ្លៃ, ភារកិច្ចត្រូវបានផ្ញើទៅវា; ការងារដែលអ្នកសរសេរកម្មវិធីក្នុងបញ្ហាចាប់ផ្តើម ដោយប្រើប្រតិបត្តិករមួយ ឬផ្សេងទៀត។

សាមញ្ញគ្រប់គ្រាន់។

កម្មវិធីកំណត់ពេលដំណើរការលើសំណុំនៃ DAGs និងកិច្ចការទាំងអស់នៅក្នុង DAGs ។

សម្រាប់ 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 * * * *')

កាលបរិច្ឆេទប្រតិបត្តិ

ដើម្បីយល់ពីរបៀបដែល Airflow ដំណើរការ វាជារឿងសំខាន់ដែលត្រូវយល់ពីកាលបរិច្ឆេទប្រតិបត្តិសម្រាប់ DAG ។ នៅក្នុង Airflow DAG មានវិមាត្រកាលបរិច្ឆេទប្រតិបត្តិ ពោលគឺអាស្រ័យលើកាលវិភាគការងាររបស់ DAG ករណីកិច្ចការត្រូវបានបង្កើតសម្រាប់កាលបរិច្ឆេទប្រតិបត្តិនីមួយៗ។ ហើយសម្រាប់កាលបរិច្ឆេទប្រតិបត្តិនីមួយៗ ភារកិច្ចអាចត្រូវបានប្រតិបត្តិឡើងវិញ ឬឧទាហរណ៍ DAG អាចដំណើរការក្នុងពេលដំណាលគ្នាក្នុងកាលបរិច្ឆេទប្រតិបត្តិជាច្រើន។ នេះត្រូវបានបង្ហាញយ៉ាងច្បាស់នៅទីនេះ៖

ជាអកុសល (ឬប្រហែលជាសំណាង៖ វាអាស្រ័យលើស្ថានភាព) ប្រសិនបើការអនុវត្តភារកិច្ចនៅក្នុង DAG ត្រូវបានកែតម្រូវ នោះការប្រតិបត្តិក្នុងកាលបរិច្ឆេទប្រតិបត្តិមុននឹងបន្តដោយគិតគូរពីការកែតម្រូវ។ វាល្អប្រសិនបើអ្នកត្រូវគណនាទិន្នន័យឡើងវិញក្នុងរយៈពេលកន្លងមកដោយប្រើក្បួនដោះស្រាយថ្មី ប៉ុន្តែវាមិនល្អទេព្រោះការផលិតឡើងវិញនៃលទ្ធផលត្រូវបានបាត់បង់ (ជាការពិតណាស់គ្មាននរណាម្នាក់រំខានអ្នកឱ្យត្រឡប់កំណែដែលត្រូវការនៃកូដប្រភពពី Git ហើយគណនាអ្វី អ្នកត្រូវការតែម្តងជាវិធីដែលអ្នកត្រូវការ)

ការបង្កើតភារកិច្ច

ការអនុវត្ត DAG គឺជាកូដ Python ដូច្នេះយើងមានវិធីងាយស្រួលបំផុតក្នុងការកាត់បន្ថយចំនួនកូដនៅពេលធ្វើការ ឧទាហរណ៍ជាមួយប្រភពដែលបានបំបែក។ ឧបមាថាអ្នកមាន MySQL shards ចំនួនបីជាប្រភព អ្នកត្រូវឡើងចូលទៅក្នុងនីមួយៗ ហើយយកទិន្នន័យមួយចំនួន។ លើសពីនេះទៅទៀតដោយឯករាជ្យនិងស្របគ្នា។ កូដ 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 មើលទៅដូចនេះ៖

ក្នុងករណីនេះ អ្នកអាចបន្ថែម ឬលុប shard ដោយគ្រាន់តែកែតម្រូវការកំណត់ និងធ្វើបច្ចុប្បន្នភាព DAG ។ មាន​ផា​សុខភាព!

អ្នកក៏អាចប្រើការបង្កើតកូដស្មុគ្រស្មាញជាងនេះផងដែរ ឧទាហរណ៍ ធ្វើការជាមួយប្រភពក្នុងទម្រង់នៃមូលដ្ឋានទិន្នន័យ ឬពិពណ៌នាអំពីរចនាសម្ព័ន្ធតារាង ក្បួនដោះស្រាយសម្រាប់ធ្វើការជាមួយតារាង និងដោយគិតគូរពីលក្ខណៈពិសេសនៃហេដ្ឋារចនាសម្ព័ន្ធ DWH បង្កើតដំណើរការមួយ។ សម្រាប់ផ្ទុកតារាង N ទៅក្នុងកន្លែងផ្ទុករបស់អ្នក។ ឬឧទាហរណ៍ ការធ្វើការជាមួយ API ដែលមិនគាំទ្រការធ្វើការជាមួយប៉ារ៉ាម៉ែត្រក្នុងទម្រង់នៃបញ្ជី អ្នកអាចបង្កើតកិច្ចការ N ក្នុង DAG ពីបញ្ជីនេះ កំណត់ភាពស្របគ្នានៃសំណើនៅក្នុង API ទៅអាងមួយ ហើយបំបែក ទិន្នន័យចាំបាច់ពី API ។ បត់បែន!

ឃ្លាំង

Airflow មានឃ្លាំង backend ផ្ទាល់ខ្លួន មូលដ្ឋានទិន្នន័យ (អាចជា MySQL ឬ Postgres យើងមាន Postgres) ដែលផ្ទុកនូវស្ថានភាពកិច្ចការ DAGs ការកំណត់ការតភ្ជាប់ អថេរសកល។ល។ នៅទីនេះ ខ្ញុំចង់និយាយបានថា ឃ្លាំងនៅក្នុង Airflow គឺសាមញ្ញណាស់ (ប្រហែល 20 តុ) ហើយងាយស្រួលប្រសិនបើអ្នកចង់បង្កើតដំណើរការផ្ទាល់ខ្លួនរបស់អ្នកនៅលើវា។ ខ្ញុំចាំតារាងចំនួន 100500 នៅក្នុងឃ្លាំង Informatica ដែលត្រូវសិក្សាអស់រយៈពេលជាយូរមុនពេលស្វែងយល់ពីរបៀបបង្កើតសំណួរ។

ការត្រួតពិនិត្យ

ដោយគិតពីភាពសាមញ្ញនៃឃ្លាំង អ្នកអាចបង្កើតដំណើរការត្រួតពិនិត្យកិច្ចការដែលងាយស្រួលសម្រាប់អ្នក។ យើងប្រើ notepad នៅក្នុង Zeppelin ដែលយើងមើលស្ថានភាពនៃកិច្ចការ៖

នេះក៏អាចជាចំណុចប្រទាក់បណ្ដាញរបស់ Airflow ខ្លួនវាផងដែរ៖

លេខកូដ Airflow គឺជាប្រភពបើកចំហ ដូច្នេះយើងបានបន្ថែមការជូនដំណឹងទៅកាន់ Telegram ។ រាល់ករណីដែលកំពុងដំណើរការនៃកិច្ចការមួយ ប្រសិនបើមានកំហុសកើតឡើង នោះ spam ក្រុមនៅក្នុង Telegram ដែលក្រុមអភិវឌ្ឍន៍ និងជំនួយទាំងមូលមាន។

យើងទទួលបានការឆ្លើយតបភ្លាមៗតាមរយៈ Telegram (ប្រសិនបើចាំបាច់) ហើយតាមរយៈ Zeppelin យើងទទួលបានរូបភាពទាំងមូលនៃកិច្ចការនៅក្នុង Airflow ។

សរុប

Airflow គឺជាប្រភពបើកចំហជាចម្បង ហើយអ្នកមិនគួររំពឹងថានឹងមានអព្ភូតហេតុពីវាទេ។ ត្រៀមខ្លួនដើម្បីដាក់ពេលវេលា និងការខិតខំប្រឹងប្រែងដើម្បីកសាងដំណោះស្រាយដែលដំណើរការ គោលដៅគឺអាចសម្រេចបាន ជឿខ្ញុំ វាមានតម្លៃ។ ល្បឿននៃការអភិវឌ្ឍន៍ភាពបត់បែនភាពងាយស្រួលនៃការបន្ថែមដំណើរការថ្មី - អ្នកនឹងចូលចិត្តវា។ ជាការពិតណាស់អ្នកត្រូវយកចិត្តទុកដាក់យ៉ាងខ្លាំងចំពោះការរៀបចំគម្រោងស្ថេរភាពនៃលំហូរខ្យល់ដោយខ្លួនឯង: អព្ភូតហេតុមិនកើតឡើងទេ។

ឥឡូវនេះយើងមាន Airflow ធ្វើការជារៀងរាល់ថ្ងៃ ប្រហែល 6,5 ពាន់ភារកិច្ច. ពួកគេមានចរិតខុសគ្នាខ្លាំង។ មានភារកិច្ចផ្ទុកទិន្នន័យទៅក្នុង DWH ចម្បងពីប្រភពផ្សេងៗគ្នា និងជាក់លាក់បំផុត មានភារកិច្ចគណនាមុខទំនិញនៅខាងក្នុង DWH សំខាន់ មានភារកិច្ចបោះពុម្ពទិន្នន័យទៅក្នុង DWH លឿន មានការងារច្រើន ច្រើនផ្សេងគ្នា និងលំហូរខ្យល់ ទំពារវាពេញមួយថ្ងៃ។ និយាយ​ជា​លេខ​គឺ​នេះ​ឯង ៥២ ម៉ឺន ភារកិច្ច ELT នៃភាពស្មុគស្មាញផ្សេងៗគ្នានៅក្នុង DWH (Hadoop) ប្រហាក់ប្រហែល។ មូលដ្ឋានទិន្នន័យ 2,5 រយ ប្រភព, នេះគឺជាក្រុមមកពី 4 អ្នកអភិវឌ្ឍន៍ ETLដែលត្រូវបានបែងចែកទៅជាដំណើរការទិន្នន័យ ETL នៅក្នុង DWH និង ELT ដំណើរការទិន្នន័យខាងក្នុង DWH និងជាការពិតណាស់ជាច្រើនទៀត អ្នកគ្រប់គ្រងម្នាក់ដែលដោះស្រាយជាមួយហេដ្ឋារចនាសម្ព័ន្ធនៃសេវាកម្ម។

ផែនការសម្រាប់អនាគត

ចំនួននៃដំណើរការកំពុងកើនឡើងដោយជៀសមិនរួច ហើយរឿងសំខាន់ដែលយើងនឹងធ្វើទាក់ទងនឹងហេដ្ឋារចនាសម្ព័ន្ធ Airflow គឺការធ្វើមាត្រដ្ឋាន។ យើងចង់បង្កើតក្រុម Airflow បែងចែកជើងមួយគូសម្រាប់កម្មករ Celery និងបង្កើតក្បាលស្ទួនដោយខ្លួនឯងជាមួយនឹងដំណើរការរៀបចំកាលវិភាគការងារ និងកន្លែងផ្ទុក។

Epilogue

ជាការពិតណាស់ នេះមិនមែនជាអ្វីគ្រប់យ៉ាងដែលខ្ញុំចង់ប្រាប់អំពី Airflow នោះទេ ប៉ុន្តែខ្ញុំបានព្យាយាមគូសបញ្ជាក់ចំណុចសំខាន់ៗ។ ចំណង់​ចំណូល​ចិត្ត​មក​ជាមួយ​នឹង​ការ​ញ៉ាំ សាកល្បង​វា​ហើយ​អ្នក​នឹង​ចូលចិត្ត​វា :)

ប្រភព: www.habr.com