Hello, Habr! Maqaalkan waxaan rabaa inaan ka hadlo hal qalab oo weyn oo loogu talagalay horumarinta hababka xogta dufcada, tusaale ahaan, kaabayaasha DWH ee shirkadda ama DataLake kaaga. Waxaan ka hadli doonaa Apache Airflow (kadib loo yaqaan socodka hawada). Waxaa si aan caddaalad ahayn looga reebay dareenka Habré, qaybta ugu muhiimsanna waxaan isku dayi doonaa inaan ku qanciyo in ugu yaraan hawadu ay mudan tahay in la eego markaad dooranayso jadwaliyaha hababkaaga ETL / ELT.
Hadda ka hor, waxaan qoray taxane maqaallo ah oo ku saabsan mawduuca DWH markii aan ka shaqeeyay Tinkoff Bank. Hadda waxaan ka mid noqday kooxda Mail.Ru Group waxaanan horumarinayaa madal loogu talagalay falanqaynta xogta ee aagga ciyaaraha. Dhab ahaantii, sida wararka iyo xalalka xiisaha leh ay u muuqdaan, kooxdayda iyo aniga waxaan halkan kaga hadli doonaa madalkeena falanqaynta xogta.
Hordhac
Markaa, aan bilowno. Waa maxay socodka hawadu? Tani waa maktabad (ama ) in la horumariyo, qorsheeyo lana kormeero hababka shaqada. Habka ugu muhiimsan ee socodka hawada: Python code waxaa loo isticmaalaa in lagu qeexo (hormarinta) hababka. Tani waxay leedahay faa'iidooyin badan oo loogu talagalay abaabulka mashruucaaga iyo horumarkaaga: nuxur ahaan, mashruucaaga (tusaale ahaan) ETL waa mashruuc Python ah, waxaadna u habeyn kartaa sidaad rabto, adoo tixgelinaya waxyaabaha gaarka ah ee kaabayaasha, tirada kooxda iyo shuruudaha kale. Qalab ahaan wax walba waa sahlan yihiin. Isticmaal tusaale ahaan PyCharm + Git. Waa wax cajiib ah oo aad u habboon!
Hadda aan eegno hay'adaha ugu muhiimsan ee socodka hawada. Markaad fahamto nuxurkooda iyo ujeedadooda, waxaad si fiican u habayn kartaa hab-dhismeedkaaga hab-dhismeedka. Waxaa laga yaabaa in cidda ugu weyni ay tahay Garaafka Acyclic Directed (hadda dambe waxaa loo yaqaan DAG).
DAG
DAG waa urur semantic ah oo ka mid ah hawlahaaga oo aad rabto in aad ku dhammaystirto si adag oo taxane ah oo waafaqsan jadwal gaar ah. Hawo-marintu waxay bixisaa is-dhexgal shabakadeed ku habboon oo lagula shaqeynayo DAGs iyo hay'adaha kale:
DAG wuxuu u ekaan karaa sidan:
Horumariyuhu, marka uu naqshadeynayo DAG, wuxuu dejiyaa hawl-wadeenno kuwaas oo hawlaha DAG gudahooda lagu dhisi doono. Halkan waxaan ku nimid qayb kale oo muhiim ah: Hawlwadeenka socodka hawada.
Hawlwadeennada
Hawlwadeenku waa hay'ad ku salaysan xaalado shaqo oo la abuuray, taasoo sharraxaysa waxa dhici doona inta lagu jiro fulinta tusaale shaqo. waxay hore ugu jireen hawlwadeeno diyaar u ah in la isticmaalo. Tusaalooyinka:
- BashOperator - hawlwadeenka fulinta amarka bash.
- PythonOperator - hawlwadeenka wacaya koodka Python.
- EmailOperator - hawlwadeenka dirida iimaylka.
- HTTPOperator - hawlwadeen ka shaqeeya codsiyada http.
- SqlOperator - hawlwadeenka fulinta koodka SQL.
- Dareemuhu waa hawlwadeen sugidda dhacdo ( imaatinka wakhtiga loo baahan yahay, muuqaalka faylka loo baahan yahay, laynka kaydka xogta, jawaabta API, iwm., iwm.).
Waxaa jira hawl wadeeno gaar ah: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.
Waxa kale oo aad samayn kartaa hawlwadeeno ku salaysan sifooyinkaaga oo aad u isticmaasho mashruucaaga. Tusaale ahaan, waxaanu abuurnay MongoDBToHiveViaHdfsTransfer, hawlwadeen ka dhoofinta dukumeentiyada MongoDB una socda Hive, iyo hawlwadeeno badan oo la shaqeeya : CHLoadFromHiveOperator iyo CHTableLoaderOperator. Asal ahaan, isla marka mashruucu si joogto ah u isticmaalo kood ku dhisan weedhaha aasaasiga ah, waxaad ka fikiri kartaa inaad u dhisto qoraal cusub. Tani waxay fududeyn doontaa horumar dheeraad ah, waxaadna ballaarin doontaa maktabadaada hawlwadeenada mashruuca.
Marka xigta, dhammaan dhacdooyinkan hawlaha waxay u baahan yihiin in la fuliyo, hadda waxaan ka hadli doonaa jadwalka.
Jadwalka
Jadwalaha hawsha socodka hawada waa lagu dhisay . Celery waa maktabad Python ah oo kuu ogolaanaysa inaad abaabusho saf iyo asynchronous iyo fulinta hawlaha qaybsan. Dhinaca socodka hawada, dhammaan hawlaha waxa loo qaybiyaa barkado. Barkadaha gacanta ayaa loo abuuray. Caadi ahaan, ujeedadoodu waa inay xaddidaan culayska shaqada ee la shaqaynta isha ama inay ku qoraan hawlaha gudaha DWH. Barkadaha waxaa lagu maamuli karaa interneedka shabakadda:
Barkad kastaa waxay leedahay xad tirada boosaska. Marka la abuurayo DAG, waxaa la siiyaa barkad:
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__Barkadda lagu qeexay heerka DAG waa la dhaafi karaa heerka hawsha.
Nidaam gaar ah, Jadwalka, ayaa mas'uul ka ah jadwalka dhammaan hawlaha socodka hawada. Dhab ahaantii, Jadwalku wuxuu la macaamilaa dhammaan makaanikada dejinta hawlaha fulinta. Hawshu waxa ay martaa dhawr marxaladood ka hor inta aan la fulin:
- Hawlihii hore waxay ku dhammaadeen DAG; mid cusub ayaa saf loogu geli karaa.
- Safka waxaa loo kala soocaa iyadoo ku xiran mudnaanta hawlaha (mudnaanta sidoo kale waa la xakameyn karaa), iyo haddii ay jirto meel bannaan oo bilaash ah oo barkadda ah, hawsha waa la qaadi karaa.
- Haddii ay jirto celery shaqaale bilaash ah, hawsha ayaa loo diraa; Shaqadii aad barnaamijka u dejisay dhibaatadu waxay bilaabmaysaa, adoo isticmaalaya hal ama mid kale.
Fudud ku filan.
Jadwalka wuxuu ku shaqeeyaa dhammaan DAG-yada iyo dhammaan hawlaha gudaha DAGs.
Jadwalka si uu u bilaabo la shaqaynta DAG, DAG waxay u baahan tahay inay dejiso jadwal:
dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')Waxaa jira hab-u-sameeyayaal diyaarsan: @once, @hourly, @daily, @weekly, @monthly, @yearly.
Waxa kale oo aad isticmaali kartaa tibaaxaha cron:
dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')Taariikhda Dilka
Si loo fahmo sida socodka hawadu u shaqeeyo, waxaa muhiim ah in la fahmo waxa Taariikhda Fulintu u tahay DAG. Socodka Hawada, DAG waxay leedahay cabbir Taariikhda Fulintu tahay, tusaale ahaan, iyadoo ku xidhan jadwalka shaqada ee DAG, tusaalooyin hawleed ayaa loo sameeyay Taariikhda Fulin kasta. Taariikh kasta oo la fulinayo, hawlaha dib ayaa loo fulin karaa - ama, tusaale ahaan, DAG waxay u shaqayn kartaa isku mar dhowr Taariikhood oo Fulin ah. Tan si cad ayaa halkan loogu muujiyay:
Nasiib darro (ama laga yaabee nasiib wanaag: waxay kuxirantahay xaaladda), haddii fulinta hawsha DAG la saxo, ka dibna fulinta taariikhda hore ee fulinta ayaa sii socon doonta iyadoo la tixgelinayo hagaajinta. Tani way fiicantahay haddii aad u baahan tahay inaad dib u xisaabiso xogta wakhtiyadii hore adoo isticmaalaya algorithm cusub, laakiin way xun tahay sababtoo ah soo saarista natiijada ayaa lumay (dabcan, qofna kuma dhibayo inaad soo celiso nooca loo baahan yahay ee koodhka isha ee Git oo xisaabiso waxa waxaad u baahan tahay hal mar, sida aad ugu baahan tahay).
Shaqo abuur
Hirgelinta DAG waa koodka Python, sidaas darteed waxaan haynaa hab aad u habboon oo lagu dhimi karo qadarka koodka marka la shaqeynayo, tusaale ahaan, ilo la jajabiyey. Aynu nidhaahno inaad haysatid saddex MySQL oo ah isha, waxaad u baahan tahay inaad mid kasta fuulo oo aad soo qaadato xog qaar. Waxaa intaa dheer, si madax-bannaan oo barbar socda. Koodhka Python ee DAG wuxuu u ekaan karaa sidan:
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 waxay u egtahay sidan:
Xaaladdan oo kale, waxaad ku dari kartaa ama ka saari kartaa jeexjeex adiga oo si fudud u hagaajinaya goobaha oo aad cusbooneysiisay DAG. Raaxo
Waxa kale oo aad isticmaali kartaa jiil code oo aad u adag, tusaale ahaan, la shaqayn ilo qaab database ah ama sharax qaab-dhismeedka miiska, algorithm ka shaqeeya miiska, iyo, iyada oo la tixgelinayo sifooyinka kaabayaasha DWH, abuuraan habsocodka. si loogu shubo miisaska N ee kaydintaada. Ama, tusaale ahaan, la shaqaynta API-ga oo aan taageerin la shaqaynta halbeegga qaab liiska, waxaad ka soo saari kartaa hawlaha N ee DAG liiskan, xaddid isbarbardhigga codsiyada API-ga barkad oo xoq xogta lagama maarmaanka ah ee API. Dabacsan!
kayd
Hawadu waxay leedahay kayd gadaal u gaar ah, database (waxay noqon kartaa MySQL ama Postgres, waxaan leenahay Postgres), kaas oo kaydiya xaaladaha hawlaha, DAGs, goobaha isku xirka, doorsoomayaasha caalamiga ah, iwm., iwm. Halkan waxaan jeclaan lahaa inaan dhihi karo kaydinta socodka hawadu waa mid aad u fudud (qiyaastii 20 miisas) waana ku haboon tahay haddii aad rabto inaad ku dhisto mid ka mid ah hababkaaga korkiisa. Waxaan xasuustaa 100500 oo miis oo ku yaal kaydka Informatica, kuwaas oo ay ahayd in la barto wakhti dheer ka hor inta aan la fahmin sida loo dhiso su'aal.
Kormeerka
Marka la eego fududaanta kaydka, waxaad dhisi kartaa hab socodka la socodka hawsha oo adiga kugu habboon. Waxaan ku isticmaalnaa warqad xusuus qor gudaha Zeppelin, halkaas oo aan ku eegno heerka hawlaha:
Tani waxay sidoo kale noqon kartaa isku xirka shabakada ee Airflow laftiisa:
Koodhka hawadu waa il furan, markaa waxaanu ku darnay digniinta Telegram. Tusaale kasta oo hawl ah, haddii uu khalad dhaco, waxay ku dhejisaa kooxda ku jirta Telegram, halkaas oo dhammaan kooxda horumarinta iyo taageerada ay ka kooban yihiin.
Waxaan jawaab degdeg ah ka helnaa Telegram (haddii loo baahdo), iyo Zeppelin waxaan ka helnaa sawirka guud ee hawlaha socodka hawada.
Wadarta
Socodka hawadu waa il furan, mana aha in aad mucjisooyin ka filan. U diyaargarow inaad waqti iyo dadaal geliso si aad u dhisto xal shaqeynaya. Hadafka waa la gaari karaa, i rumee, waa u qalantaa. Xawaaraha horumarka, dabacsanaanta, fududaynta ku darista hababka cusub - waad jeclaan doontaa. Dabcan, waxaad u baahan tahay inaad fiiro gaar ah u yeelato abaabulka mashruuca, xasiloonida Hawada lafteeda: mucjisooyinku ma dhacaan.
Hadda waxaan haynaa socodka hawada oo shaqeeya maalin kasta qiyaastii 6,5 kun oo hawlood. Aad bay u kala duwan yihiin dabeecad ahaan. Waxaa jira hawlo lagu soo shubayo xogta DWH-ga ugu weyn oo laga keenay ilo badan oo kala duwan oo aad u gaar ah, waxaa jira hawlo lagu xisaabinayo bakhaarka gudaha DWH-ga ugu weyn, waxaa jira hawlo lagu daabacayo xogta DWH degdeg ah, waxaa jira hawlo badan oo badan oo kala duwan - iyo Airflow. maalinba maalinta ka danbaysa wuu ruugayaa dhamaantood. Ku hadalka tirada, tani waa 2,3 kun Hawlaha ELT ee kakanaanta kala duwan ee DWH (Hadoop), qiyaastii. 2,5 boqol oo database ilo, tani waa koox ka socota 4 horumariyayaal ETL, kuwaas oo loo kala qaybiyay habaynta xogta ETL ee DWH iyo ELT xogta gudaha DWH iyo dabcan qaar kale oo badan hal admin, kuwaas oo ka shaqeeya kaabayaasha adeegga.
Qorshayaasha mustaqbalka
Tirada geeddi-socodyadu si lama huraan ah ayey u kordhayaan, waxa ugu weyn ee aan ka qaban doono marka la eego kaabayaasha socodka Hawada waa kor u kaca. Waxaan rabnaa inaan dhisno koox socodka hawada, oo aan u qoondeyno labo lugood oo loogu talagalay shaqaalaha Selery, oo aan sameyno madax isku duuban oo leh habab jadwal shaqo iyo kayd.
Epilogue
Tani, dabcan, maaha wax kasta oo aan jeclaan lahaa inaan ka sheego socodka hawada, laakiin waxaan isku dayay inaan iftiimiyo qodobbada ugu muhiimsan. Cunto-cunista waxay la socotaa cunto cunid, tijaabi oo waad jeclaan doontaa :)
Source: www.habr.com