වායු ප්‍රවාහය යනු කණ්ඩායම් දත්ත සැකසුම් ක්‍රියාවලීන් පහසුවෙන් සහ ඉක්මනින් සංවර්ධනය කිරීම සහ නඩත්තු කිරීම සඳහා වූ මෙවලමකි

හෙලෝ, හබ්ර්! මෙම ලිපියෙන් මට කණ්ඩායම් දත්ත සැකසුම් ක්‍රියාවලීන් සංවර්ධනය කිරීම සඳහා එක් විශිෂ්ට මෙවලමක් ගැන කතා කිරීමට අවශ්‍යයි, උදාහරණයක් ලෙස, ආයතනික DWH හෝ ඔබේ DataLake යටිතල ව්‍යුහය තුළ. අපි Apache Airflow ගැන කතා කරමු (මින් පසුව එය Airflow ලෙස හැඳින්වේ). එය Habré පිළිබඳ අවධානය අසාධාරණ ලෙස අහිමි කර ඇති අතර, ප්‍රධාන කොටසේදී මම ඔබට ඒත්තු ගැන්වීමට උත්සාහ කරන්නේ ඔබේ ETL/ELT ක්‍රියාවලි සඳහා උපලේඛකයක් තෝරාගැනීමේදී අවම වශයෙන් Airflow දෙස බැලීම වටී.

මීට පෙර, මම Tinkoff බැංකුවේ සේවය කරන විට DWH මාතෘකාව පිළිබඳ ලිපි මාලාවක් ලිවීය. දැන් මම Mail.Ru Group කණ්ඩායමේ කොටසක් බවට පත් වී ඇති අතර සූදු ප්රදේශයේ දත්ත විශ්ලේෂණය සඳහා වේදිකාවක් සංවර්ධනය කරමි. ඇත්ත වශයෙන්ම, ප්‍රවෘත්ති සහ රසවත් විසඳුම් දිස්වන විට, මගේ කණ්ඩායම සහ මම දත්ත විශ්ලේෂණ සඳහා අපගේ වේදිකාව ගැන මෙහි කතා කරන්නෙමු.

පූර්විකාව

ඉතින්, අපි පටන් ගනිමු. වායු ප්රවාහය යනු කුමක්ද? මෙය පුස්තකාලයකි (හෝ පුස්තකාල කට්ටලයක්) වැඩ ක්රියාවලිය සංවර්ධනය කිරීම, සැලසුම් කිරීම සහ අධීක්ෂණය කිරීම. වායු ප්‍රවාහයේ ප්‍රධාන ලක්ෂණය: පයිතන් කේතය ක්‍රියාවලි විස්තර කිරීමට (සංවර්ධනය කිරීමට) භාවිතා කරයි. ඔබේ ව්‍යාපෘතිය සහ සංවර්ධනය සංවිධානය කිරීම සඳහා මෙය බොහෝ වාසි ඇත: සාරාංශයක් ලෙස, ඔබේ (උදාහරණයක් ලෙස) ETL ව්‍යාපෘතිය පයිතන් ව්‍යාපෘතියක් පමණක් වන අතර, යටිතල ව්‍යුහයේ විශේෂතා, කණ්ඩායම් ප්‍රමාණය සහ ඔබට අවශ්‍ය පරිදි එය සංවිධානය කළ හැකිය. වෙනත් අවශ්යතා. උපකරණ වශයෙන් සෑම දෙයක්ම සරලයි. උදාහරණයක් ලෙස PyCharm + Git භාවිතා කරන්න. එය පුදුම සහ ඉතා පහසුයි!

දැන් අපි Airflow හි ප්‍රධාන ආයතන බලමු. ඔවුන්ගේ සාරය සහ අරමුණ අවබෝධ කර ගැනීමෙන්, ඔබට ඔබේ ක්‍රියාවලි ගෘහ නිර්මාණ ශිල්පය ප්‍රශස්ත ලෙස සංවිධානය කළ හැකිය. සමහර විට ප්‍රධාන ආයතනය අධ්‍යක්ෂණය කළ Acyclic ප්‍රස්ථාරය විය හැකිය (මෙතැන් සිට DAG ලෙස හැඳින්වේ).

ඩැග්

DAG යනු නිශ්චිත කාලසටහනකට අනුව දැඩි ලෙස නිර්වචනය කරන ලද අනුපිළිවෙලකින් සම්පූර්ණ කිරීමට ඔබට අවශ්‍ය ඔබගේ කාර්යයන්හි යම් අර්ථවත් ඇසුරකි. Airflow DAGs සහ අනෙකුත් ආයතන සමඟ වැඩ කිරීම සඳහා පහසු වෙබ් අතුරු මුහුණතක් සපයයි:

DAG මේ වගේ විය හැකිය:

සංවර්ධකයා, DAG නිර්මාණය කිරීමේදී, DAG තුළ ඇති කාර්යයන් ගොඩනගනු ලබන ක්‍රියාකරුවන් සමූහයක් තබයි. මෙන්න අපි තවත් වැදගත් ආයතනයකට පැමිණෙමු: Airflow Operator.

මෙහෙයුම්කරුවන්

ක්‍රියාකරුවෙකු යනු රැකියා අවස්ථා නිර්මාණය කරන පදනම මත වන ආයතනයකි, එය රැකියා අවස්ථාවක් ක්‍රියාත්මක කිරීමේදී සිදුවන්නේ කුමක්ද යන්න විස්තර කරයි. GitHub වෙතින් වායු ප්‍රවාහ නිකුත් වේ දැනටමත් භාවිතා කිරීමට සූදානම් ක්‍රියාකරුවන් කට්ටලයක් අඩංගු වේ. උදාහරණ:

• BashOperator - bash විධානයක් ක්රියාත්මක කිරීම සඳහා ක්රියාකරු.
• PythonOperator - Python කේතය ඇමතීමේ ක්‍රියාකරු.
• EmailOperator - ඊමේල් යැවීම සඳහා ක්රියාකරු.
• HTTPOoperator - http ඉල්ලීම් සමඟ වැඩ කිරීම සඳහා ක්රියාකරු.
• SqlOperator - SQL කේතය ක්රියාත්මක කිරීම සඳහා ක්රියාකරු.
• සංවේදකය යනු සිදුවීමක් සඳහා බලා සිටීම සඳහා ක්‍රියාකරුවෙකි (අවශ්‍ය කාලය පැමිණීම, අවශ්‍ය ගොනුවේ පෙනුම, දත්ත සමුදායේ රේඛාවක්, API වෙතින් ප්‍රතිචාරයක්, ආදිය).

තවත් විශේෂිත ක්‍රියාකරුවන් ඇත: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

ඔබට ඔබේම ලක්ෂණ මත පදනම්ව ක්‍රියාකරුවන් සංවර්ධනය කර ඔබේ ව්‍යාපෘතියේ ඒවා භාවිතා කළ හැකිය. උදාහරණයක් ලෙස, අපි MongoDBToHiveViaHdfsTransfer, MongoDB සිට Hive වෙත ලේඛන අපනයනය කිරීමේ ක්‍රියාකරුවෙකු සහ වැඩ කිරීම සඳහා ක්‍රියාකරුවන් කිහිප දෙනෙකු නිර්මාණය කළෙමු. නිවස ක්ලික් කරන්න: CHLoadFromHiveOperator සහ CHTableLoaderOperator. අත්‍යවශ්‍යයෙන්ම, ව්‍යාපෘතියක් මූලික ප්‍රකාශ මත ගොඩනගා ඇති කේතය නිතර භාවිතා කළ වහාම, ඔබට එය නව ප්‍රකාශයක් බවට ගොඩනැගීම ගැන සිතිය හැක. මෙය තවදුරටත් සංවර්ධනය සරල කරනු ඇති අතර, ඔබ ව්‍යාපෘතියේ ක්‍රියාකරුවන්ගේ පුස්තකාලය පුළුල් කරනු ඇත.

ඊළඟට, මෙම සියලු කාර්යයන් ඉටු කළ යුතු අතර, දැන් අපි උපලේඛකයා ගැන කතා කරමු.

උපලේඛනකරු

වායු ප්‍රවාහයේ කාර්ය කාලසටහන ගොඩනගා ඇත සැල්දිරි. සැල්දිරි යනු පයිතන් පුස්තකාලයක් වන අතර එය ඔබට පෝලිමක් සහ අසමමුහුර්ත සහ බෙදා හරින ලද කාර්යයන් සංවිධානය කිරීමට ඉඩ සලසයි. 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. නිදහස් සේවක සැල්දිරි තිබේ නම්, කාර්යය එය වෙත යවනු ලැබේ; ඔබ ගැටලුව තුළ වැඩසටහන්ගත කළ කාර්යය ආරම්භ වන්නේ, එක් හෝ වෙනත් ක්රියාකරුවෙකු භාවිතා කරමිනි.

ප්රමාණවත් තරම් සරලයි.

Scheduler සියලු DAGs කට්ටලය සහ DAGs තුළ ඇති සියලුම කාර්යයන් මත ධාවනය වේ.

උපලේඛකයාට DAG සමඟ වැඩ කිරීම ආරම්භ කිරීමට, DAG හට කාලසටහනක් සැකසීමට අවශ්‍ය වේ:

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

සූදානම් කළ පෙරසිටුවීම් කට්ටලයක් තිබේ: @once, @hourly, @daily, @weekly, @monthly, @yearly.

ඔබට ක්‍රෝන් ප්‍රකාශන ද භාවිතා කළ හැකිය:

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

ක්රියාත්මක කිරීමේ දිනය

වායු ප්‍රවාහය ක්‍රියා කරන ආකාරය තේරුම් ගැනීමට, DAG සඳහා ක්‍රියාත්මක වන දිනය කුමක්දැයි වටහා ගැනීම වැදගත් වේ. වායු ප්‍රවාහයේදී, DAG හට ක්‍රියාත්මක කිරීමේ දින මානයක් ඇත, එනම්, DAG හි වැඩ කාලසටහන අනුව, එක් එක් ක්‍රියාත්මක කිරීමේ දිනය සඳහා කාර්ය අවස්ථා නිර්මාණය වේ. එක් එක් ක්‍රියාත්මක කිරීමේ දිනය සඳහා, කාර්යයන් නැවත ක්‍රියාත්මක කළ හැකිය - නැතහොත්, උදාහරණයක් ලෙස, DAG ක්‍රියාත්මක කිරීමේ දින කිහිපයක එකවර ක්‍රියා කළ හැකිය. මෙය මෙහි පැහැදිලිව පෙන්වා ඇත:

අවාසනාවන්ත ලෙස (හෝ සමහර විට වාසනාවකට: එය තත්ත්වය මත රඳා පවතී), DAG හි කාර්යය ක්රියාත්මක කිරීම නිවැරදි කර ඇත්නම්, පෙර ක්රියාත්මක කිරීමේ දිනයේ ක්රියාත්මක කිරීම ගැලපීම් සැලකිල්ලට ගනිමින් ඉදිරියට යනු ඇත. ඔබට නව ඇල්ගොරිතමයක් භාවිතයෙන් පසුගිය කාල පරිච්ඡේදවල දත්ත නැවත ගණනය කිරීමට අවශ්‍ය නම් මෙය හොඳයි, නමුත් එය නරකයි ප්‍රතිඵලයේ ප්‍රතිනිෂ්පාදනය නැති වී ඇති නිසා (ඇත්ත වශයෙන්ම, Git වෙතින් මූලාශ්‍ර කේතයේ අවශ්‍ය අනුවාදය ආපසු ලබා දීමට සහ ගණනය කිරීමට කිසිවෙකු ඔබට කරදර කරන්නේ නැත. ඔබට එක් වරක් අවශ්‍ය වේ, ඔබට අවශ්‍ය ආකාරයට).

කාර්යයන් උත්පාදනය කිරීම

DAG ක්‍රියාත්මක කිරීම පයිතන් හි කේතයකි, එබැවින් වැඩ කරන විට කේත ප්‍රමාණය අඩු කිරීමට අපට ඉතා පහසු ක්‍රමයක් තිබේ, උදාහරණයක් ලෙස, ෂර්ඩ් ප්‍රභවයන් සමඟ. ඔබට මූලාශ්‍රයක් ලෙස MySQL කැබලි තුනක් ඇතැයි කියමු, ඔබ එකකට නැඟී දත්ත කිහිපයක් ලබා ගත යුතුය. එපමණක්ද නොව, ස්වාධීනව හා සමාන්තරව. DAG හි ඇති Python කේතය මේ වගේ විය හැක:

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 සමඟ වැඩ කිරීම, ඔබට මෙම ලැයිස්තුවෙන් DAG හි N කාර්යයන් ජනනය කළ හැකිය, API හි ඉල්ලීම්වල සමාන්තරකරණය සංචිතයකට සීමා කර සීරීමට API වෙතින් අවශ්ය දත්ත. නම්යශීලී!

ගබඩාව

Airflow හට තමන්ගේම පසුපෙළ ගබඩාවක් ඇත, දත්ත ගබඩාවක් (MySQL හෝ Postgres විය හැක, අපට Postgres ඇත), එය කාර්යයන්, DAGs, සම්බන්ධතා සැකසුම්, ගෝලීය විචල්‍යයන් යනාදිය ගබඩා කරයි. මෙහිදී මම කැමතියි වායු ප්‍රවාහයේ ගබඩාව ඉතා සරලයි (වගු 20 ක් පමණ) සහ ඔබට එය මත ඔබේම ක්‍රියාවලියක් ගොඩනගා ගැනීමට අවශ්‍ය නම් පහසුය. මට මතකයි Informatica ගබඩාවේ ඇති වගු 100500, විමසුමක් ගොඩනඟන්නේ කෙසේද යන්න තේරුම් ගැනීමට පෙර එය බොහෝ කාලයක් අධ්‍යයනය කළ යුතුව තිබුණි.

අධීක්ෂණය

ගබඩාවේ සරල බව සැලකිල්ලට ගෙන, ඔබට පහසු කාර්ය අධීක්ෂණ ක්‍රියාවලියක් ගොඩනගා ගත හැකිය. අපි Zeppelin හි notepad එකක් භාවිතා කරමු, එහිදී අපි කාර්යයේ තත්ත්වය දෙස බලමු:

මෙය Airflow හි වෙබ් අතුරු මුහුණත ද විය හැකිය:

ගුවන් ප්‍රවාහ කේතය විවෘත මූලාශ්‍රයක් වන බැවින් අපි ටෙලිග්‍රාම් වෙත අනතුරු ඇඟවීම් එකතු කර ඇත. කාර්යයක සෑම ධාවන අවස්ථාවක්ම, දෝෂයක් සිදුවුවහොත්, සමස්ථ සංවර්ධන සහ සහායක කණ්ඩායම සමන්විත වන Telegram හි කණ්ඩායම ස්පෑම් කරයි.

අපට Telegram හරහා ඉක්මන් ප්‍රතිචාරයක් ලැබෙනු ඇත (අවශ්‍ය නම්), සහ Zeppelin හරහා අපට Airflow හි කාර්යයන් පිළිබඳ සමස්ත චිත්‍රයක් ලැබේ.

එකතුව

වායු ප්‍රවාහය මූලික වශයෙන් විවෘත මූලාශ්‍රයක් වන අතර ඔබ එයින් ආශ්චර්යයන් බලාපොරොත්තු නොවිය යුතුය. වැඩ කරන විසඳුමක් ගොඩනඟා ගැනීම සඳහා කාලය සහ ශ්රමය යෙදවීමට සූදානම්ව සිටින්න. ඉලක්කය සාක්ෂාත් කරගත හැකිය, මාව විශ්වාස කරන්න, එය වටිනවා. සංවර්ධනයේ වේගය, නම්‍යශීලී බව, නව ක්‍රියාවලි එකතු කිරීමේ පහසුව - ඔබ එයට කැමති වනු ඇත. ඇත්ත වශයෙන්ම, ඔබ ව්යාපෘතියේ සංවිධානයට විශාල අවධානයක් යොමු කළ යුතුය, වායු ප්රවාහයේම ස්ථාවරත්වය: ප්රාතිහාර්යයන් සිදු නොවේ.

දැන් අපිට Airflow දිනපතා වැඩ කරනවා කාර්යයන් 6,5 දහසක් පමණ. ඔවුන් චරිතයෙන් තරමක් වෙනස් ය. විවිධ සහ ඉතා නිශ්චිත මූලාශ්‍රවලින් ප්‍රධාන DWH වෙත දත්ත පැටවීමේ කාර්යයන් ඇත, ප්‍රධාන DWH තුළ ගබඩා ඉදිරිපස ගණනය කිරීමේ කාර්යයන් ඇත, වේගවත් DWH වෙත දත්ත ප්‍රකාශයට පත් කිරීමේ කාර්යයන් ඇත, විවිධ කාර්යයන් රාශියක් ඇත - සහ වායු ප්‍රවාහය දවසින් දවස ඒවා ඔක්කොම හපනවා. ඉලක්කම් වලින් කතා කරන්නේ නම්, මෙයයි 2,3 දහසක් DWH (Hadoop) තුළ විවිධ සංකීර්ණත්වයේ ELT කාර්යයන්, දළ වශයෙන්. දත්ත සමුදායන් 2,5 සියයක් මූලාශ්‍ර, මෙය කණ්ඩායමකි 4 ETL සංවර්ධකයින්, DWH හි ETL දත්ත සැකසීමට සහ DWH තුළ ELT දත්ත සැකසීමට බෙදා ඇති අතර ඇත්ත වශයෙන්ම තවත් එක් පරිපාලකයෙක්, සේවාවේ යටිතල පහසුකම් සමඟ කටයුතු කරන.

අනාගතය සඳහා සැලසුම්

ක්‍රියාවලි ගණන නොවැළැක්විය හැකි ලෙස වර්ධනය වෙමින් පවතින අතර, වායු ප්‍රවාහ යටිතල පහසුකම් සම්බන්ධයෙන් අප විසින් සිදු කරනු ලබන ප්‍රධානතම දෙය වන්නේ පරිමාණය කිරීමයි. අපට ගුවන් ප්‍රවාහ පොකුරක් තැනීමට, සැල්දිරි කම්කරුවන් සඳහා කකුල් යුගලයක් වෙන් කිරීමට සහ රැකියා කාලසටහන් ක්‍රියාවලීන් සහ ගබඩාවක් සහිත ස්වයං අනුපිටපත් හිසක් සෑදීමට අවශ්‍ය වේ.

එපිලේජ්

ඇත්ත වශයෙන්ම, මෙය වායු ප්‍රවාහය ගැන කියන්නට කැමති සියල්ල නොවේ, නමුත් මම ප්‍රධාන කරුණු ඉස්මතු කිරීමට උත්සාහ කළෙමි. ආහාර රුචිය පැමිණෙන්නේ ආහාර ගැනීමෙනි, එය උත්සාහ කරන්න, ඔබ එයට කැමති වනු ඇත :)

මූලාශ්රය: www.habr.com