Python on Google Cloud Functions で作るバッチ処理

https://jxpress.connpass.com/event/215106/ での登壇

Transcript

1. 1ZUIPO
   PO
   (PPHMF
   $MPVE
   'VODUJPOT
   Ͱ࡞Δόονॲཧ
   5BJUP
   4V[VLJ
2. ࣗݾ঺հ
   5BJUP
   4V[VLJʢླ໦
   ହేʣ
   όοΫΤϯυΤϯδχΞ
   ʢʴϑϩϯτΤϯυΤϯδχΞݟश͍ʣ
   ޷͖ͳݴޠ
   1ZUIPO
   (P
   5ZQF4DSJQU
   +9௨৴ࣾʹೖࣾͯ͠೥ʢʁʣܦ͔ͬͨͳɾɾɾ
3. +9ͷงғؾ ʮࣗ༝ʯ
   εʔύʔϑϨοΫεʢίΞλΠϜͳ͠ͷϑϨοΫεʣɻ
   ׬શϦϞʔτɻձࣾͷݻఆ੮͸͋ͬͯͳ͍Α͏ͳ΋ΜͰ͢ɻ
   ΈΜͳϚΠϖʔεͰ։ൃΛਐΊ͍ͯΔͱ͍͏ҹ৅ɻ
   ʮεϐʔυײʯ
   ͲΜͲΜ৽͍͠΋ͷΛग़͍ͯͧ͘͠ʂͱ͍͏งғؾ͕޷͖ɻ
   ʮਆอொʯ
   ձࣾ͸ਆอொʹ͋ΔɻனٳΈ͸ΧϨʔ԰ʹߦͬͯॻళʹߦཱͬͯͪಡΈͯ͠ؼͬͯ͘Δɻ
   ʮถॏ͞Μʢฐࣾ$&0ʣʯ
   ໘ന͍ਓɻϐβ͕޷͖ɻ
4. എܠ 01
5. 2,ݱঢ়෼ੳͱ໰୊఺ ࠓ೔ͷ͓࿩
   1ZUIPO
   PO
   (PPHMF
   $MPVE
   'VODUJPOT
   Ͱ࡞Δ
   όονॲཧ
   ͳͥ͜ͷ࿩Λ͢Δʁ
   ձࣾͷงғؾΛ஌ͬͯ΋Β͏ͨΊʹ͸ɺ࠷ۙͷ։ൃͷ࿩Λ͢Δͷ͕Ұ൪ʂͱ͍͏͜ͱͰ 1,എܠ
6. 2,ݱঢ়෼ੳͱ໰୊఺ ࠷ۙ
   (PPHMF
   $MPVE
   'VODUJPOT
   ࢖ͬͯ·͢
   "84
   -BNCEB΋Α͘࢖͏
   ͳͥʁίʔυྔ͕গͳ͘ࡁΉ͔ΒɻΑΓૣ͘։ൃ͕ऴΘΔ͠ɺͦͷޙͷอक΋ָɻ 1,എܠ
7. 2,ݱঢ়෼ੳͱ໰୊఺ ࠷ۙ࡞ͬͨ΋ͷ
   (PPHMF
   $MPVE
   'VODUJPOTʢҎԼ($'ʣ
   Λ࢖ͬͨόονॲཧ
   σʔλΛσʔλϕʔε͔Βநग़͠ɺ($'্ͰՃ޻͠ɺετϨʔδ΁ग़ྗ
   σʔλϕʔε͔Βநग़͞ΕΔσʔλྔ͕ͦͦ͜͜ͷେ͖͞ʢສϨίʔυʣ
   όονͷ࢖༻ස౓͸਺೔ʹ̍౓͙Β͍ 1,എܠ σʔλϕʔε ($'
   ͪΐͬͱͨ͠σʔλՃ޻ ετϨʔδ
8. Ͳ͏࡞͔ͬͨ
    ૉ๿ͳ࣮૷
    ࠷௿ݶͷ࣮૷
    ฒྻॲཧʹΑΔ࣮૷
    'BOPVUʹΑΔ࣮૷ 02
9. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ૉ๿ͳ࣮૷ σʔλϕʔε σʔλͷ
   நग़ͱՃ޻ ετϨʔδ def transfer_data(event, context): with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn: current_data = None while True: data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break write_data_to_storage(storage_conn, edit_data(data)) current_data = data[-1]
10. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ૉ๿ͳ࣮૷ σʔλϕʔε σʔλͷ
    நग़ͱՃ޻ ετϨʔδ def transfer_data(event, context): with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn: current_data = None while True: data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break write_data_to_storage(storage_conn, edit_data(data)) current_data = data[-1] σʔλநग़ σʔλՃ޻ʢʴసૹʣ
11. 2,ݱঢ়෼ੳͱ໰୊఺ ՝୊ɾɾɾͱ͍͏͔શવμϝ
    શͯͷσʔλΛॲཧ͢Δલʹؔ਺͕ڧ੍ऴྃ
    ($'͸ɺؔ਺ͷ࠷େ࣮ߦ࣌ؒ͸෼͔ͩΒʢͪͳΈʹ
    "84
    -BNCEB
    ͸෼ʣ 2,Ͳ͏࡞͔ͬͨ - ૉ๿ͳ࣮૷
12. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ࠷௿ݶͷ࣮૷ σʔλϕʔε σʔλͷ
    நग़ͱՃ޻ ετϨʔδ ετϨʔδ
    ్தܦաΛ อଘ def transfer_data(event, context): with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2: current_data = load(storage_conn2) while True: datas = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break write_data_to_storage(storage_conn1, edit_data(data)) current_data = data[-1] save(storage_conn2, current_data)
13. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ࠷௿ݶͷ࣮૷ σʔλϕʔε σʔλͷ
    நग़ͱՃ޻ ετϨʔδ ετϨʔδ
    ్தܦաΛ อଘ def transfer_data(event, context): with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2: current_data = load(storage_conn2) while True: datas = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break write_data_to_storage(storage_conn1, edit_data(data)) current_data = data[-1] save(storage_conn2, current_data) ॲཧͷ్தܦաΛ
    TUPSBHF͔Βऔಘ ॲཧͷ్தܦաΛ
    TUPSBHF΁อଘ
14. 2,ݱঢ়෼ੳͱ໰୊఺ ՝୊
    શͯͷσʔλΛॲཧ·Ͱͷ͕࣌ؒσΧ͍
    ސ٬ͷཁ๬Λड͚͔ͯΒճ͢όονͳͷͰɺͳΔ΂͘଎͘׬ྃͯ͠ཉ͍͠ 2,Ͳ͏࡞͔ͬͨ - ࠷௿ݶͷ࣮૷
15. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ฒྻॲཧʹΑΔ࣮૷ σʔλϕʔε σʔλͷ
    நग़ͱՃ޻ ετϨʔδ ετϨʔδ
    ్தܦաΛ อଘ def transfer_data(event, context): # σʔλͷநग़ jobs = {} max_jobs = 10 with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2: current_data = load(storage_conn2) while True: sweep_terminated_jobs(jobs) if len(jobs) > max_jobs: continue data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break job = new_job(target=transfer_data_each, args=(data, )) job.start() jobs[job.name] = job current_data = data[-1] save(storage_conn2, current_data) def transfer_data_each(data): # σʔλͷՃ޻ with new_storage_conn() as storage_conn1: write_data_to_storage(storage_conn1, edit_data(data))
16. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - ฒྻॲཧʹΑΔ࣮૷ σʔλϕʔε σʔλͷ
    நग़ͱՃ޻ ετϨʔδ ετϨʔδ
    ్தܦաΛ อଘ def transfer_data(event, context): # σʔλͷநग़ jobs = {} max_jobs = 10 with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2: current_data = load(storage_conn2) while True: sweep_terminated_jobs(jobs) if len(jobs) > max_jobs: continue data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break job = new_job(target=transfer_data_each, args=(data, )) job.start() jobs[job.name] = job current_data = data[-1] save(storage_conn2, current_data) def transfer_data_each(data): # σʔλͷՃ޻ with new_storage_conn() as storage_conn1: write_data_to_storage(storage_conn1, edit_data(data)) ऴྃͨ͠+PCΛ
    ϓʔϧ͔Βআ֎ +PCͷϓʔϧ +PC࡞੒ ͜͜Ͱ͍͏+PCͱ͸
    ϓϩηεɺ΋͘͠͸εϨου
17. 2,ݱঢ়෼ੳͱ໰୊఺ ՝୊
    શͯͷσʔλΛॲཧ·Ͱͷ͕࣌ؒɺׂͱσΧ͍
    ࢖༻Ͱ͖ΔܭࢉϦιʔε͕͋·Γ૿΍ͤͳ͍ɻ
    ($'ͷϝϞϦΛ্͛ͪΌ͑ʢ͍ΘΏΔεέʔϧΞοϓʣʂ
    ɾɾɾͱࢥ͚ͬͨͲɺ($'ͷϝϞϦͪΐͬͱ͔͠૿΍ͤͳ͍΍Μɻ
    ཁ͸ɺ͜ͷํ๏Ͱ͸͋·ΓվળͰ͖ͳ͔ͬͨ
    ฒྻॲཧͷίʔυ͸อक͠೉͍
    ιʔείʔυ͕ෳࡶʹͳΓ͕ͪɻ
    Ϣχοτςετॻ͖೉͍ɻ
    ͔ͤͬ͘ͷ($'ͷϝϦοτʢίʔυྔগͳ͍ނͷอक͠΍͢͞ʣΛࡴͯ͠͠·͏ɻ 2,Ͳ͏࡞͔ͬͨ - ฒྻॲཧʹΑΔ࣮૷
18. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - Fan-outʹΑΔ࣮૷ σʔλϕʔε σʔλͷநग़ ετϨʔδ ɾɾɾ σʔλͷՃ޻ def transfer_data(event, context): # σʔλͷநग़ with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2,\ new_queue_conn() as queue_conn: current_data = load(storage_conn2) while True: data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break enqueue(queue_conn, data) current_data = data[-1] save(storage_conn2, current_data) def transfer_data_each(event, context): # σʔλͷՃ޻ data = get_data_from_event(event) with new_storage_conn() as storage_conn: write_datas_to_storage(storage_conn, edit_data([data]))
19. 2,ݱঢ়෼ੳͱ໰୊఺ 2,Ͳ͏࡞͔ͬͨ - Fan-outʹΑΔ࣮૷ σʔλϕʔε σʔλͷநग़ ετϨʔδ ɾɾɾ σʔλͷՃ޻ def transfer_data(event, context): # σʔλͷநग़ with new_db_conn() as db_conn,\ new_storage_conn() as storage_conn1,\ new_storage_conn() as storage_conn2,\ new_queue_conn() as queue_conn: current_data = load(storage_conn2) while True: data = read_data_from_db( db_conn, current_data, 1000, ) if len(data) <= 0: break enqueue(queue_conn, data) current_data = data[-1] save(storage_conn2, current_data) def transfer_data_each(event, context): # σʔλͷՃ޻ data = get_data_from_event(event) with new_storage_conn() as storage_conn: write_datas_to_storage(storage_conn, edit_data([data])) ΩϡʔʹೖΕΔ͚ͩ ΩϡʔͷσʔλΛೖྗ ͱͯ͠ىಈ͢Δ
    ΋͏Ұͭͷ
    ($'
    JOTUBODF
20. 2,ݱঢ়෼ੳͱ໰୊఺ ՝୊
    ෳ਺ͷ($'Λ؅ཧ͠ͳ͚Ε͹ͳΒͳ͍ 2,Ͳ͏࡞͔ͬͨ - Fan-outʹΑΔ࣮૷
21. 2,ݱঢ়෼ੳͱ໰୊఺ ίωΫγϣϯྨ͸άϩʔόϧม਺ʹ֨ೲ͠࠶ར༻
    IUUQTDMPVEHPPHMFDPNGVODUJPOTEPDTCFTUQSBDUJDFTUJQT
    ($'ͷ࠷େΠϯελϯε਺͸ɺؔ࿈ίϯϙωϯτͷϦιʔε্ݶʹ഑ྀ͠ͳ͕Β
    IUUQTDMPVEHPPHMFDPNGVODUJPOTEPDTNBYJOTUBODFT
    ෳ਺($'Λ؅ཧ͢Δࡍʹ࢖ͬͯΔπʔϧʢ"84
    -BNCEBͰ΋࢖͑·͢ʣ
    IUUQTXXXTFSWFSMFTTDPN
    ($'ͷ(SBDFGVMγϟοτμ΢ϯ
    ($'͕λΠϜΞ΢τ͢Δલʹؔ਺Λऴྃ͢ΔΑ͏ʹ
    $16ό΢ϯυͳॲཧ͸ɺ($'ʹ͓͍ͯ͸ɺϚϧνϓϩηε͸ݫͦ͠͏ɻʢ'BOPVUͨ͠΄͏͕ྑ͍ʣ
    $16ͷίΞ਺ͱ͔͕Α͘Θ͔ΒΜɻ($'ͷϝϞϦΛ૿΍ͤ͹$16΋૿͑ΔΒ͍͕͠ɾɾɾ 2,Ͳ͏࡞͔ͬͨ - ࿩͍ͯ͠ͳ͍͜ͱ
22. 2,ݱঢ়෼ੳͱ໰୊఺ σʔλϕʔε
    1PTUHSFT
    ීஈ͸ɺ3%#ͱͯ͠.Z42-ͱ1PTUHSFTΛซ༻ɻ/P42-ͱͯ͠%ZOBNP%#ɻ
    1ZUIPO
    ύοέʔδ؅ཧ
    QPFUSZ
    ීஈ͸ɺQPFUSZ
    PS
    1JQFOWͰ͢ɻ
    σʔλϕʔε
    TRMBMDIFNZ
    ීஈ͸ɺ%KBOHPΛ࢖͍ͬͯΔՕॴͰ͸%KBOHPͷ03.Ͱ͢ɻͦ͏Ͱͳ͍ՕॴͰ͸TRMBMDIFNZ
    ϩʔΧϧͷ։ൃ؀ڥ
    .BD্Ͱ%PDLFSίϯςφΛ্ཱͪ͛ͯɺίϯςφ্ͰΞϓϦέʔγϣϯΛಈ͔͍ͯ͠·͢ɻ
    ʢࠓճͷ։ൃ࿩ͱ͸ؔ܎ͳ͍Ͱ͕͢ɾɾɾʣ͍ͭͰʹݴ͓ͬͯ͘ͱ
    ීஈ͸΢Σϒαʔόʔͱͯ͠ɺ%KBOHP
    PS
    'BTU"1*Ͱ͢ɻ'BTU"1*Λ࢖͏͜ͱ͕ଟ͘ͳ͍ͬͯ·͢ɻ 2,Ͳ͏࡞͔ͬͨ - ࢖༻ٕͨ͠ज़·ͱΊ
23. ·ͱΊ 03
24. ,·ͱΊ ࡞Γํ
    ·ͣ͸࠷௿ݶͷཁ݅Λຬͨ͢΋ͷΛૉૣ͘࡞Γɺͦͷޙͷվળ͸গͣͭ͠ɻ
    ($'Λ࢖ͬͨόονॲཧ
    ؔ਺ͷ࠷େ࣮ߦ࣌ؒʢ෼ʣʹޚ༻৺ɻ
    ύϑΥʔϚϯε޲্͍ͨ͠ͳΒɺεέʔϧΞοϓͰ͸ͳ͘εέʔϧΞ΢τΛҙࣝɻͭ·Γ'BOPVUɻ
    ࠓ೔ͷ͓࿩͸($'Ͱ͕ͨ͠ɺ"84
    -BNCEBͰ΋ಉ͡ϊ΢ϋ΢͕࢖͑Δ͸ͣɻ
25. Thank you!
26. Agenda 1, എܠ 2, Ͳ͏࡞͔ͬͨ 3, ·ͱΊ
27. εϥΠυͷςϯϓϨʔτ 03
28. 2,ݱঢ়෼ੳͱ໰୊఺ ϒϥ΢βͷΈͰνϟοτ΍ϑΝΠϧͷ΍ΓͱΓΛ࣮ݱ ɾ$PNFUͱ͍͏ٕज़Λར༻
    ɾHPPHMF͕$PNFUαʔόʔΛఏڙ͢Δ༧ఆ ϑΝΠϧ͸ʮ"NB[PO4ʯʹΞοϓϩʔυ ɾαʔόʔ্ʹ͋ΔͷͰ͍ͭͰ΋μ΢ϯϩʔυՄ
    ɾ௿ίετ͔ͭ҆ఆͨ͠ӡ༻͕Մೳ
29. 2,ݱঢ়෼ੳͱ໰୊఺ w ϝʔϧɺి࿩ΑΓѹ౗ తʹޮ཰͕͍͍ ਤ w ใ࿈૬ɺٞ࿦ɺґཔͳ Ͳ׆༻ͷ෯͕޿͍ w ແྉͰखܰʹ࢖͑Δ
30. 2,ݱঢ়෼ੳͱ໰୊఺ ਤ ྑ͍ػೳ͚ͩͰ͸࢖ΘΕͳ͍