એરફ્લો એ બેચ ડેટા પ્રોસેસિંગ પ્રક્રિયાઓને અનુકૂળ અને ઝડપથી વિકસાવવા અને જાળવવા માટેનું એક સાધન છે

હેલો, હેબ્ર! આ લેખમાં હું બેચ ડેટા પ્રોસેસિંગ પ્રક્રિયાઓ વિકસાવવા માટેના એક મહાન સાધન વિશે વાત કરવા માંગુ છું, ઉદાહરણ તરીકે, કોર્પોરેટ DWH અથવા તમારા ડેટાલેકના ઇન્ફ્રાસ્ટ્રક્ચરમાં. અમે અપાચે એરફ્લો (ત્યારબાદ એરફ્લો તરીકે ઓળખાય છે) વિશે વાત કરીશું. તે Habré પર ધ્યાનથી અયોગ્ય રીતે વંચિત છે, અને મુખ્ય ભાગમાં હું તમને ખાતરી આપવાનો પ્રયત્ન કરીશ કે તમારી ETL/ELT પ્રક્રિયાઓ માટે શેડ્યૂલર પસંદ કરતી વખતે ઓછામાં ઓછું એરફ્લો જોવા યોગ્ય છે.

અગાઉ, જ્યારે મેં Tinkoff બેંકમાં કામ કર્યું ત્યારે મેં DWH વિષય પર શ્રેણીબદ્ધ લેખો લખ્યા હતા. હવે હું Mail.Ru ગ્રૂપ ટીમનો ભાગ બની ગયો છું અને ગેમિંગ ક્ષેત્રમાં ડેટા વિશ્લેષણ માટે એક પ્લેટફોર્મ વિકસાવી રહ્યો છું. વાસ્તવમાં, જેમ જેમ સમાચાર અને રસપ્રદ ઉકેલો દેખાય છે, તેમ તેમ મારી ટીમ અને હું ડેટા એનાલિટિક્સ માટેના અમારા પ્લેટફોર્મ વિશે અહીં વાત કરીશું.

પ્રસ્તાવના

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

હવે ચાલો એરફ્લોની મુખ્ય સંસ્થાઓ જોઈએ. તેમના સાર અને હેતુને સમજીને, તમે તમારી પ્રક્રિયા આર્કિટેક્ચરને શ્રેષ્ઠ રીતે ગોઠવી શકો છો. કદાચ મુખ્ય એન્ટિટી ડાયરેક્ટેડ એસાયક્લિક ગ્રાફ છે (ત્યારબાદ DAG તરીકે ઓળખાય છે).

ડૅગ

DAG એ તમારા કાર્યોનું અમુક અર્થપૂર્ણ જોડાણ છે જેને તમે ચોક્કસ શેડ્યૂલ અનુસાર સખત રીતે વ્યાખ્યાયિત ક્રમમાં પૂર્ણ કરવા માંગો છો. એરફ્લો DAGs અને અન્ય સંસ્થાઓ સાથે કામ કરવા માટે અનુકૂળ વેબ ઈન્ટરફેસ પ્રદાન કરે છે:

DAG આના જેવું દેખાઈ શકે છે:

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

ઓપરેટરો

ઑપરેટર એ એક એન્ટિટી છે જેના આધારે જોબના દાખલા બનાવવામાં આવે છે, જે વર્ણવે છે કે નોકરીના દાખલાના અમલ દરમિયાન શું થશે. GitHub માંથી એરફ્લો રિલીઝ થાય છે પહેલેથી ઉપયોગ કરવા માટે તૈયાર ઓપરેટર્સનો સમૂહ ધરાવે છે. ઉદાહરણો:

• BashOperator - bash આદેશ ચલાવવા માટે ઓપરેટર.
• PythonOperator - Python કોડ કૉલ કરવા માટે ઓપરેટર.
• ઈમેલ ઓપરેટર — ઈમેલ મોકલવા માટે ઓપરેટર.
• HTTPOperator - HTTP વિનંતીઓ સાથે કામ કરવા માટે ઓપરેટર.
• SqlOperator - SQL કોડ ચલાવવા માટે ઓપરેટર.
• સેન્સર એ ઇવેન્ટની રાહ જોવા માટે ઓપરેટર છે (જરૂરી સમયનું આગમન, જરૂરી ફાઇલનો દેખાવ, ડેટાબેઝમાં એક લાઇન, API તરફથી પ્રતિસાદ, વગેરે. વગેરે).

ત્યાં વધુ ચોક્કસ ઓપરેટરો છે: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

તમે તમારી પોતાની લાક્ષણિકતાઓના આધારે ઓપરેટરો પણ વિકસાવી શકો છો અને તમારા પ્રોજેક્ટમાં તેનો ઉપયોગ કરી શકો છો. ઉદાહરણ તરીકે, અમે MongoDBToHiveViaHdfsTransfer, MongoDB થી Hive માં દસ્તાવેજો નિકાસ કરવા માટે એક ઓપરેટર અને સાથે કામ કરવા માટે ઘણા ઓપરેટરો બનાવ્યાં છે. ક્લીકહાઉસ: CHLoadFromHiveOperator અને CHTableLoaderOperator. અનિવાર્યપણે, જેમ જેમ કોઈ પ્રોજેક્ટમાં મૂળભૂત નિવેદનો પર બનેલા કોડનો વારંવાર ઉપયોગ થાય છે, ત્યારે તમે તેને નવા નિવેદનમાં બનાવવા વિશે વિચારી શકો છો. આ વધુ વિકાસને સરળ બનાવશે, અને તમે પ્રોજેક્ટમાં ઓપરેટરોની તમારી લાઇબ્રેરીને વિસ્તૃત કરશો.

આગળ, કાર્યોના આ તમામ ઉદાહરણોને એક્ઝિક્યુટ કરવાની જરૂર છે, અને હવે આપણે શેડ્યૂલર વિશે વાત કરીશું.

શેડ્યૂલર

એરફ્લોનું કાર્ય શેડ્યૂલર બિલ્ટ ઓન છે સેલરી. સેલેરી એ પાયથોન લાઇબ્રેરી છે જે તમને કતાર ગોઠવવા ઉપરાંત અસુમેળ અને કાર્યોના વિતરિત અમલીકરણની મંજૂરી આપે છે. એરફ્લો બાજુ પર, તમામ કાર્યોને પૂલમાં વિભાજિત કરવામાં આવે છે. પૂલ મેન્યુઅલી બનાવવામાં આવે છે. સામાન્ય રીતે, તેમનો હેતુ સ્ત્રોત સાથે કામ કરવાના વર્કલોડને મર્યાદિત કરવાનો અથવા 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 સ્તરે વ્યાખ્યાયિત થયેલ પૂલને કાર્ય સ્તરે ઓવરરાઇડ કરી શકાય છે.
એક અલગ પ્રક્રિયા, શેડ્યૂલર, એરફ્લોમાં તમામ કાર્યોને સુનિશ્ચિત કરવા માટે જવાબદાર છે. વાસ્તવમાં, શેડ્યૂલર એક્ઝેક્યુશન માટે કાર્યો સેટ કરવાના તમામ મિકેનિક્સ સાથે વ્યવહાર કરે છે. કાર્ય અમલમાં મૂકતા પહેલા ઘણા તબક્કાઓમાંથી પસાર થાય છે:

1. અગાઉના કાર્યો DAG માં પૂર્ણ કરવામાં આવ્યા છે; એક નવું કતારબદ્ધ કરી શકાય છે.
2. કાર્યોની પ્રાધાન્યતાના આધારે કતારને સૉર્ટ કરવામાં આવે છે (અગ્રતાઓને પણ નિયંત્રિત કરી શકાય છે), અને જો પૂલમાં મફત સ્લોટ હોય, તો કાર્યને ઓપરેશનમાં લઈ શકાય છે.
3. જો ત્યાં મફત કાર્યકર સેલરી હોય, તો કાર્ય તેને મોકલવામાં આવે છે; તમે સમસ્યામાં પ્રોગ્રામ કરેલ કાર્ય એક અથવા બીજા ઓપરેટરનો ઉપયોગ કરીને શરૂ થાય છે.

પર્યાપ્ત સરળ.

શેડ્યૂલર તમામ DAGs અને DAG ની અંદરના તમામ કાર્યોના સેટ પર ચાલે છે.

શેડ્યૂલર માટે 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 માં કાર્યના અમલીકરણમાં સુધારો કરવામાં આવે, તો અગાઉની એક્ઝેક્યુશન તારીખમાં અમલીકરણ ગોઠવણોને ધ્યાનમાં લઈને આગળ વધશે. જો તમારે નવા અલ્ગોરિધમનો ઉપયોગ કરીને પાછલા સમયગાળામાં ડેટાની પુનઃગણતરી કરવાની જરૂર હોય તો આ સારું છે, પરંતુ તે ખરાબ છે કારણ કે પરિણામની પુનઃઉત્પાદનક્ષમતા ખોવાઈ ગઈ છે (અલબત્ત, કોઈ તમને ગિટમાંથી સ્રોત કોડનું આવશ્યક સંસ્કરણ પરત કરવા અને તેની ગણતરી કરવા માટે પરેશાન કરતું નથી. તમારે એક સમયની જરૂર છે, જે રીતે તમને તેની જરૂર છે).

કાર્યો પેદા કરી રહ્યા છીએ

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 માંથી જરૂરી ડેટા. લવચીક!

ભંડાર

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

મોનીટરીંગ

રીપોઝીટરીની સરળતાને જોતાં, તમે તમારા માટે અનુકૂળ કાર્ય મોનીટરીંગ પ્રક્રિયા બનાવી શકો છો. અમે ઝેપ્પેલીનમાં નોટપેડનો ઉપયોગ કરીએ છીએ, જ્યાં આપણે કાર્યોની સ્થિતિ જોઈએ છીએ:

આ એરફ્લોનું વેબ ઈન્ટરફેસ પણ હોઈ શકે છે:

એરફ્લો કોડ ઓપન સોર્સ છે, તેથી અમે ટેલિગ્રામમાં ચેતવણી ઉમેરી છે. કાર્યની દરેક ચાલી રહેલ ઘટના, જો કોઈ ભૂલ થાય, તો ટેલિગ્રામમાં જૂથને સ્પામ કરે છે, જ્યાં સમગ્ર વિકાસ અને સહાયક ટીમનો સમાવેશ થાય છે.

અમે ટેલિગ્રામ (જો જરૂરી હોય તો) દ્વારા તાત્કાલિક પ્રતિસાદ પ્રાપ્ત કરીએ છીએ, અને ઝેપ્પેલીન દ્વારા અમને એરફ્લોમાં કાર્યોનું એકંદર ચિત્ર પ્રાપ્ત થાય છે.

કુલ

એરફ્લો મુખ્યત્વે ઓપન સોર્સ છે અને તમારે તેનાથી ચમત્કારોની અપેક્ષા ન રાખવી જોઈએ. કામ કરે છે તે ઉકેલ બનાવવા માટે સમય અને પ્રયત્નો કરવા તૈયાર રહો. ધ્યેય પ્રાપ્ત કરી શકાય તેવું છે, મારા પર વિશ્વાસ કરો, તે મૂલ્યવાન છે. વિકાસની ગતિ, લવચીકતા, નવી પ્રક્રિયાઓ ઉમેરવાની સરળતા - તમને તે ગમશે. અલબત્ત, તમારે પ્રોજેક્ટના સંગઠન, એરફ્લોની સ્થિરતા પર ઘણું ધ્યાન આપવાની જરૂર છે: ચમત્કારો થતા નથી.

હવે અમારી પાસે એરફ્લો દરરોજ કાર્યરત છે લગભગ 6,5 હજાર કાર્યો. તેઓ પાત્રમાં તદ્દન અલગ છે. મુખ્ય DWH માં ઘણાં વિવિધ અને ખૂબ જ ચોક્કસ સ્ત્રોતોમાંથી ડેટા લોડ કરવાના કાર્યો છે, મુખ્ય DWH ની અંદર સ્ટોરફ્રન્ટ્સની ગણતરી કરવાના કાર્યો છે, ઝડપી DWH માં ડેટા પ્રકાશિત કરવાના કાર્યો છે, ત્યાં ઘણા બધા વિવિધ કાર્યો છે - અને એરફ્લો દિવસ પછી તે બધાને ચાવે છે. સંખ્યામાં બોલતા, આ છે 2,3 હજાર DWH (Hadoop) ની અંદર વિવિધ જટિલતાના ELT કાર્યો, આશરે. 2,5 સો ડેટાબેઝ સ્ત્રોતો, આ એક ટીમ છે 4 ETL વિકાસકર્તાઓ, જે DWH માં ETL ડેટા પ્રોસેસિંગ અને DWH ની અંદર ELT ડેટા પ્રોસેસિંગમાં વિભાજિત છે અને અલબત્ત વધુ એક એડમિન, જે સેવાના ઇન્ફ્રાસ્ટ્રક્ચર સાથે વ્યવહાર કરે છે.

ભવિષ્યની યોજનાઓ

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

ઉપસંહાર

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

સોર્સ: www.habr.com