AWSでIcebergを使ってデータウェアハウスを構築してみる / 20240810-jawsug-akita

2024/08/10 (土) [東北][秋田][JAWS-UG 秋田]勉強会 にて発表した資料

https://jaws-tohoku.connpass.com/event/321228/

Transcript

1. "84Ͱ*DFCFSHΛ࢖ͬͯ
   σʔλ΢ΣΞϋ΢εΛߏஙͯ͠ΈΔ 
   ּݪ
   ޺ +"846(౦๺
   !
   ळా
2. ࣗݾ঺հ  @kasacchiful @kasacchiful ּݪ
   ޺
   wΫϥεϝιουגࣜձࣾ
   σʔλࣄۀຊ෦
   w৽ׁݝ৽ׁࢢࡏॅ
   w޷͖ͳ"84αʔϏε
   4ɺ-BNCEBɺ4UFQ
   'VODUJPOT
   w+"846(৽ׁࢧ෦ʗ1ZUIPOػցֶशษڧձJO৽ׁʗ+B445৽ׁ ʗ48"/**ʗ$MPVE fl BSF
   .FFUVQ
   /JJHBUBʗFUD
   w"84
   $PNNVOJUZ
   #VJMEFS
   
   
   +BQBO
   "84
   "MM
   $FSUJ fi DBUJPOT
   &OHJOFFS
3. +"846(
   ৽ׁ
   ։࠵ܗࣜ  ຊฤษڧձ
   wෆఆظ։࠵ʗʙ࣌ؒఔ౓ʗΦϑϥΠϯ
   wࣄલʹςʔϚܾΊͯ։࠵
   ϓνΩϟονΞοϓձ
   wຖि໦༵໷࣌ʗ࣌ؒఔ౓ʗΦϯϥΠϯ
   wͦͷ৔ͰؾʹͳΔهࣄݟͨΓɺ͓೰Έ૬ஊड͚ͨΓ
4. ໨࣍  wσʔλϑΥʔϚοτͷ੔ཧ
   w"QBDIF
   *DFCFSHͱ͸
   w*DFCFSHͰߏங͢Δ%8)
   w·ͱΊ
5.  σʔλϑΥʔϚοτͷ੔ཧ
6. Α͋͘Δߏ੒ 
7. Α͋͘Δߏ੒  (MVFͰ&5-ॲཧ
   1BSRVFUϑΝΠϧΛ 4ʹอଘ "UIFOBͰ
   ෼ੳΫΤϦ࣮ߦ
8. σʔλϨΠΫͰΑ͘࢖ΘΕΔϑΝΠϧϑΥʔϚοτ  $47
   +40/
   wਓ͕ؒૢ࡞͠΍͍͢ɺΞϓϦέʔγϣϯؒ࿈ܞ͠΍͍͢
   "730
   ߦࢦ޲ϑΥʔϚοτ
   wσʔλͷߴ଎ॲཧ͕Մೳɺ0-"1ॲཧʹར༻Մೳ
   1BSRVFU
   ྻࢦ޲ϑΥʔϚοτ
   ˕ਪ঑
   w෼ੳ༻్ʹ࢖͍΍͍͢
9. ໰୊఺  $47
   +40/
   wਓ͕ؒૢ࡞͠΍͍͢ɺΞϓϦέʔγϣϯؒ࿈ܞ͠΍͍͢
   "730
   ߦࢦ޲ϑΥʔϚοτ
   wσʔλͷߴ଎ॲཧ͕Մೳɺ0-"1ॲཧʹར༻Մೳ
   1BSRVFU
   ྻࢦ޲ϑΥʔϚοτ
   ˕ਪ঑
   w෼ੳ༻్ʹ࢖͍΍͍͢ ಘҙ
   
   େྔσʔλͷςʔϒϧ୯ҐͷಡΈॻ͖
   ύʔςΟγϣϯϨϕϧͷಡΈॻ͖
   ෆಘҙ
   ϨίʔυϨϕϧͷಡΈॻ͖
10. ͦͷଞͷ໰୊  εϞʔϧϑΝΠϧʹΑΔύϑΥʔϚϯεྼԽ
    wϑΝΠϧ*0͕૿͑ͯಡΈࠐΈ଎౓௿Լ
    w4
    "1*ίʔϧ਺૿ՃʹΑΔར༻ྉ૿Ճ
    σʔλͷฤूɾ࡟আ͕೉͍͠
    wݸਓ৘ใͷ෦෼Λޙ͔Β࡟আͨͯ͘͠΋ʜ
    w෼ੳ࣌ͷτϥϯβΫγϣϯΛ୲อ͍͚ͨ͠Ͳ೉͍͠
    
11. ͦͷଞͷ໰୊  ಉ࣌ΞΫηε࣌ͷ੔߹ੑ୲อ͕೉͍͠
    wෳ਺ͷτϥϯβΫγϣϯॲཧΛ໰୊ͳ͘ߦ͏ʹ͸ෆ޲͖
    σʔλߋ৽͕ೖΔͱɺաڈͷঢ়ଶΛ෮ݩ͠ʹ͍͘
    wσʔλαΠΤϯε࣌ͷ࣮ݧ࠶ݱΛ͍ͨ͠
    wݱࡏͱաڈͷσʔλΛൺֱ͍ͨ͠
    wόοΫΞοϓϦετΞͩͱɺେֻ͔Γ͗͢Δ
12.  ৽ͨͳσʔλϑΥʔϚοτͰ ղܾ
13. σʔλϑΥʔϚοτ  "QBDIF
    )VEJ
    "QBDIF
    *DFCFSH
    %FMUB
    -BLF
14. σʔλϑΥʔϚοτ  "QBDIF
    )VEJ
    "QBDIF
    *DFCFSH
    %FMUB
    -BLF ࠓճ͸IcebergΛѻ͏
15.  Apache Icebergͱ͸
16. "QBDIF
    *DFCFSH  ΦʔϓϯιʔεͷσʔλςʔϒϧϑΥʔϚοτ
    wڊେͳ෼ੳςʔϒϧͷͨΊͷߴੑೳͳϑΥʔϚοτ
    wσʔλϨΠΫʹू·͍ͬͯΔσʔληοτͷૢ࡞Λ؆ૉԽ Մೳ
    wτϥϯβΫγϣϯॲཧՄೳ
    w(MVF
    W
    ΍
    "UIFOB
    &OHJOF
    W
    Ͱ
    *DFCFSH
    W
    ʹରԠ
    
    ݱࡏ
17. ओͳಛ௃  "$*%ಛੑ
    w*OTFSU
    
    %FMFUF
    
    6QEBUFͷτϥϯβΫγϣϯॲཧରԠ
    4DIFNB
    &WPMVUJPO
    wεΩʔϚมߋʹରԠɺ్தͰύʔςΟγϣϯมߋ΋Մೳ
    )JEEFO
    1BSUJUJPO
    wϢʔβ͕ҙࣝ͢Δ͜ͱͳ͘ύʔςΟγϣϯ؅ཧͰ͖Δ
    wΫΤϦ͕ςʔϒϧͷ෺ཧϨΠΞ΢τʹґଘ͠ͳ͘ͳͬͨ
    4OBQTIPU
    
    5JNFUSBWFM
    wτϥϯβΫγϣϯຖʹεφοϓγϣοτΛ࣋ͭ
    wεφοϓγϣοτόʔγϣϯ΍͋Δ࣌఺ͷঢ়ଶͷςʔϒϧΛΫΤϦͰ؆୯ʹࢀরՄೳ
18. *DFCFSHͷߏ੒  IUUQTJDFCFSHBQBDIFPSHTQFD "84Ͱ͸(MVF
    %BUB
    $BUBMPHͰ؅ཧ σʔλϑΝΠϧ͸ 1BSRVFU ϝλσʔλ͸+40/΍ "730ͳͲͰߏ੒
19. *DFCFSHͷߏ੒  IUUQTBXTBNB[PODPNKQCMPHTCJHEBUBNPOJUPSJOHBQBDIFJDFCFSHNFUBEBUBMBZFSVTJOHBXTMBNCEBBXTHMVFBOEBXTDMPVEXBUDI "84Ͱ͸(MVF
    %BUB
    $BUBMPHͰ؅ཧ σʔλϑΝΠϧ͸ 1BSRVFU ϝλσʔλ͸+40/΍ "730ͳͲͰߏ੒
20. *DFCFSHͷߏ੒  *DFCFSHςʔϒϧ͸ɺ
    EBUB
    ͱ
    NFUBEBUB
    Ͱߏ੒
21. *DFCFSHͷߏ੒  NFUBEBUB
    ͸
    +40/ͱBWSPϑΝΠϧͰ؅ཧ
22. *DFCFSHͷߏ੒  EBUB
    ͸
    1BSRVFUϑΝΠϧͰ֨ೲ
23. &YQSFTTJWF
    42-  w*OTFSU
    
    %FMFUF
    
    6QEBUF
    w.FSHF
    ͳͲɺॊೈͳ42- ΫΤϦʹରԠ
    w%FMFUF͸࡟আϑΝΠϧͰϑ ϥά؅ཧ
    w6QEBUF͸*OTFSU %FMFUFͷ ৼΔ෣͍
24. &YQSFTTJWF
    42-  w"UIFOBͰΫΤϦ ࣮ߦ͢Δͷ͕؆୯
25. εΩʔϚఆٛ  w"UIFOBͰఆٛ͢Δͷ͕؆ ୯
    w"MUFSจ΋࢖͑Δ
26. $3&"5&
    5"#-&  CREATE TABLE `sample_partition_100_db`.`partition100_date_iceberg` ( id string, uuid string, target_date date ) PARTITIONED BY (day(target_date)) LOCATION 's3://sample-partition-100-db/partition100_date_iceberg' TBLPROPERTIES ( 'table_type'='ICEBERG' ); `UBCMF@UZQF`
    
    `*$&#&3(`
    Λࢦఆ
27. εφοϓγϣοτ
    
    λΠϜτϥϕϧ  wτϥϯβΫγϣϯຖʹε φοϓγϣοτΛ࣋ͭ
    wεφοϓγϣοτόʔδϣ ϯΛࢦఆͯ͠ɺաڈ࣌఺ͷ σʔλΛࢀরՄೳ
28. εφοϓγϣοτ
    
    λΠϜτϥϕϧ  wτϥϯβΫγϣϯຖʹε φοϓγϣοτΛ࣋ͭ
    wεφοϓγϣοτόʔδϣ ϯΛࢦఆͯ͠ɺաڈ࣌఺ͷ σʔλΛࢀরՄೳ
29. εφοϓγϣοτ
    
    λΠϜτϥϕϧ  wτϥϯβΫγϣϯຖʹε φοϓγϣοτΛ࣋ͭ
    wλΠϜελϯϓΛࢦఆ͠ ͯɺաڈ࣌఺ͷσʔλΛࢀ রՄೳ
30. 0QUJNJ[F
    
    7BDVVN  wෆཁͳϑΝΠϧ࡟আ౳Λ࣮ࢪ͠ ͯ࠷దԽ
    w0QUJNJ[F
    w࠷దͳϑΝΠϧϨΠΞ΢τʹ ॻ͖׵͑Δ
    w7BDVVN
    wεφοϓγϣοτ༗ޮظݶ੾ ΕٴͼΦʔϓϯϑΝΠϧͷ࡟ আ
    ϝλσʔλ࠷దԽ
31.  IcebergͰߏங͢ΔDWH
32. Α͋͘Δߏ੒  (MVFͰ&5-ॲཧ
    1BSRVFUϑΝΠϧΛ 4ʹอଘ "UIFOBͰ
    ෼ੳΫΤϦ࣮ߦ
33. ྫ͑͹͜Μͳײ͡Ͱ*DFCFSHςʔϒϧʹ࿈ܞ  (MVFͰશྔऔಘ
    PS
    ࠩ෼औಘ
    $47ग़ྗͱԾఆ "UIFOBͰ
    ෼ੳΫΤϦ࣮ߦ (MVFͰ*DFCFSH
    ςʔϒϧʹߋ৽
34. (MVFδϣϒͰ.&3(&  ## ..<省略>.. ## Read Csv from Catalog df = glueContext.create_dynamic_frame.from_catalog( database="sample_partition_100_db", table_name="partition100_date_csv") ## Transform Type df = ApplyMapping.apply( frame=df, mappings=[ ("id", "string", "id", "string"), ("uuid", "string", "uuid", "string"), ("target_date", "string", "target_date", "date") ]).toDF() ## Store tmp table df.createOrReplaceTempView("tmp_partition100_date_csv") ## Merge spark.sql(''' MERGE INTO glue_catalog.sample_partition_100_db.partition100_date_i ceberg t USING tmp_partition100_date_csv s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.uuid = s.uuid, t.target_date = s.target_date WHEN NOT MATCHED THEN INSERT * ''' ) job.commit()
35. ࣮ࡍͷࣄྫ 
36. ࣮ࡍͷࣄྫ  ೔࣍Ͱσʔλऔಘ
    ʢશྔ
    PS
    ࠩ෼ʣ
37. ࣮ࡍͷࣄྫ  *DFCFSHςʔϒϧʹ
    Ϛʔδ
    -BNCEBͷ৔߹ "UIFOBܦ༝Ͱ
38. τϥϯβΫγϣϯຖʹ࿈ܞ
    wมߋσʔλΩϟϓνϟ $%$ Λར༻
    wมߋ෼Λૹͬͯ*DFCFSHม׵ Ԡ༻ 
39.  ·ͱΊ
40. ·ͱΊ  ·ͱΊ
    w"QBDIF
    *DFCFSH͸ڊେͳ෼ੳςʔϒϧͷͨΊͷߴੑೳͳ ϑΥʔϚοτ
    wσʔληοτͷૢ࡞Λ؆ૉԽͰ͖Δ
    wදܗࣜͷσʔλͰ͋Ε͹ɺ*DFCFSHςʔϒϧͰอଘ͢Δ͜ ͱ΋ݕ౼͓ͯ͘͠ͱྑ͍