Embulkに足りない5つのこと

embulk meetup tokyoで話しました！
ユースケースが書かれているので是非参考にして下さい。

Transcript

1. Embulkʹ଍Γͳ͍ 5ͭͷ͜ͱ 2015-12-15 Embulk Meetup Tokyo #2 @Civitaspo
2. ͓͜ͱΘΓ
3. ࡞ͬͨπʔϧͷ࿩͕ ग़͖ͯ·͕͢
4. ϦϦʔε͠·ͤΜʂ
5. ͝ΊΜͳ͍͞(><)
6. ཧ༝ɿ͋Δਓ͔ΒͷҰ੠
7. ࠓ͜ͷπʔϧΛ ϦϦʔε͢Δͱ ੈͷதΛࠞཚͷӔʹ ר͖ࠐΉ͜ͱ ʹͳͬͯ͠·͏ͩΖ͏ɻ
8. ࠞཚͷछʹͳΔͩΖ͏ػೳ͸ ʮEmbulk Projectʹظ଴͢Δ͜ͱʯ ͱ͍͏߲໨Ͱઆ໌͠·͢
9. Ͱ͸࢝Ί·͢ʂ
10. Embulkʹ଍Γͳ͍ 5ͭͷ͜ͱ 2015-12-15 Embulk Meetup Tokyo #2 @Civitaspo
11. ࣗݾ঺հ தࢁوത (@Civitaspo) • DeNA: ೖࣾ3೥໨ • ෼ੳܥΠϯϑϥΤϯδχΞ • σʔλճऩ؀ڥߏங • Hadoopӡ༻…etc. • Perl / Ruby / Java • ৽ଔͰձࣾʹೖͬͯॳΊͯί ϯιʔϧ։͍ͨস
12. ࡞ͬͨEmbulk Pluginୡ embulk-input-hdfs embulk-output-hdfs embulk-output-sftp embulk-filter-join-file embulk-filter-json-key embulk-filter-expand-json embulk-filter-flatten-json embulk-filter-distinct ࣭໰΍ཁ๬ɺΞυόΠε౳͋Ε͹twitterͰʂ
13. ͪͳΈʹ શ෦Java PluginͰ͕͢
14. ๻͸Intellij࢖ͬͯॻ͍ͯ·͢ vim!! vim!!
15. ΞδΣϯμ • ฐࣾͷBulkdataࣄ৘ • ߏஙͨ͠Embulkج൫ • ଍Γͳ͔͔ͬͨΒิͬͨػೳ̑ͭ • Embulk Projectʹظ଴͢Δ͜ͱ
16. ΞδΣϯμ • ฐࣾͷBulkdataࣄ৘ • ߏஙͨ͠Embulkج൫ • ଍Γͳ͔͔ͬͨΒิͬͨػೳ̑ͭ • Embulk Projectʹظ଴͢Δ͜ͱ
17. ฐࣾͷBulkdataࣄ৘
18. HDFS -> Vertica
19. ฐࣾͷBulkdataࣄ৘ • ݎ࿚ͳϩάճऩج൫ • શͯͷαʔϏεͷϩά͸HDFSʹ֨ೲ͞ΕΔ • ϑΥʔϚοτ΋౷Ұ͞Ε͍ͯΔ • TSV + JSONͷࠞ߹ϑΥʔϚοτ
20. ฐࣾͷBulkdataࣄ৘ • ෼ੳͰ࢖༻͢ΔओཁετϨʔδ͸Vertica • HDFSͷϩάΛͦͷ··Verticaʹ֨ೲ͍ͨ͠ • ͨͩ͠streaming insert͸Ͱ͖ͳ͍… • Vertica͸ߴස౓ͳσʔλ౤ೖʹඇৗʹ ऑ͍
21. Vertica? •ྻࢦ޲ܕߴ଎ूܭσʔλϕʔε •঎༻ιϑτ΢ΣΞϥΠηϯε(1TB·Ͱ͸ແྉ) •ඇৗʹ๛෋ͳ෼ੳؔ਺ •twitter΍facebookͰ΋࢖ΘΕͯΔ •ฐࣾͷ෼ੳ؀ڥͷओ࣠
22. ߴ·Δχʔζ • GCS, S3, BigQueryͳͲcloud storageͷར༻֦େ • HDFS -> Vertica ͚ͩͰ͸ͳ͍Bulkloadχʔζ
24. Embulk
25. ͜Ε͕΄͔ͬͨ͠Μ΍ʂ
26. ͬͦ͘͞ಋೖ
27. ΞδΣϯμ • ฐࣾͷBulkdataࣄ৘ • ߏஙͨ͠Embulkج൫ • ଍Γͳ͔͔ͬͨΒิͬͨػೳ̑ͭ • Embulk Projectʹظ଴͢Δ͜ͱ
28. Wrapper ߏஙͨ͠Embulkج൫
29. σʔλ֨ೲ·ͰͷྲྀΕ
30. Wrapper 1. GoogleSpreadSheet͔Β εΩʔϚ৘ใͱBulkloadλΠϓΛநग़
31. Wrapper 2. Ϋϥελ৘ใ΍εέδϡʔϧ৘ใͱ ඥ෇͚ͯMySQLʹ֨ೲ
32. Wrapper 3. εέδϡʔϧʹैͬͯRedis΁ Enqueue
33. Wrapper 4. ඥͮ͘BulkloadλΠϓ΍ɺΫϥελ ৘ใ͔Βconfig.ymlΛࣗಈੜ੒
34. Wrapper 5. Embulk WrapperΛkick
35. Wrapper 6. ֨ೲઌͷσʔλΛ࡟আ
36. Wrapper 7. Embulk run
37. Wrapper 8. ॲཧ݁ՌΛMySQLʹอଘ
38. Wrapper 9. ݁Ռදࣔ
39. ৄࡉͳ࣮૷ʹؔͯ͠ ؾʹͳΔํ͸࠙਌ձͰʂ
40. ࠓ೔࿩͢෦෼
41. Wrapper ࠓ೔࿩͢෦෼
42. Wrapper ࠓ೔࿩͢෦෼ ͳΜͰspreadsheet࢖ͬͯΔͷʁ
43. Wrapper ࠓ೔࿩͢෦෼ Job Queue؅ཧͲ͏΍ͬͯΔͷʁ
44. Wrapper ࠓ೔࿩͢෦෼ ͳΜͰwrapper͔·ͯ͠Δͷʁ
45. Wrapper ࠓ೔࿩͢෦෼ embulkʹ໰୊͸ͳ͔ͬͨͷʁ
46. ͳͲʹ͍ͭͯ࿩͠·͢
47. ΞδΣϯμ • ฐࣾͷBulkdataࣄ৘ • ߏஙͨ͠Embulkج൫ • ଍Γͳ͔͔ͬͨΒิͬͨػೳ̑ͭ • Embulk Projectʹظ଴͢Δ͜ͱ
48. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
49. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
50. ଍Γͳ͍͜ͱ̍
51. YAML؅ཧ
52. YAML؅ཧ • Embulkͷ࢓༷͚ͩͲɻɻɻ • 1ͭͷBulkloadʹରͯ͠1ͭͷYAMLϑΝΠϧ
53. ฐࣾͷࣄ৘ • εΩʔϚ৘ใ͸ΞφϦετ͕ఆٛ • εΩʔϚ৘ใҎ֎ͷઃఆ͸෼ੳج൫͕؅ཧ => ͻͱͭͷconfig.ymlੜ੒ʹෳ਺ਓ͕ؔΘΔ͜ͱʹ…
54. ฐࣾͷࣄ৘ • ର৅ͷBulkload͸1αʔϏε͋ͨΓ20~50 • ৗʹ20Ҏ্ͷαʔϏε͕ฒྻͰՔಇ͠ɺҠΓม ΘΓ΋ܹ͍͠ => େྔͷyaml͕ੜ੒͞Εͯ؅ཧ͕ͭΒ͍(઀ଓ৘ใͱ͔)
56. ج൫؅ཧ ΞφϦετ؅ཧ
58. ൒෼͘Β͍ʹͳͬͨ
59. Α͘มΘΔ ΄ͱΜͲมΘΒͳ͍
61. ݁ہΑ͘มΘΔͷ͸ εΩʔϚ৘ใ͘Β͍
62. ͭͬͨ͘ • Embulk Config Generator & UI • ઀ଓ৘ใΛDBͰҰݩ؅ཧ • BulkloadύλʔϯΛந৅Խͯ͠DBͰ؅ཧ • εΩʔϚ৘ใ͸spreadsheetͰ؅ཧ
63. ͭͬͨ͘ • config.yml͸࣮ߦ௚લʹੜ੒͞ΕΔͷͰ࢓༷ม ߋͳͲʹ΋ॊೈʹରԠͰ͖Δ • ࢒೦ͳ͕Βguess͸࢖͑ͳ͘ͳΓ·ͨ͠ɻ • ಛʹࠔͬͯͳ͍ • ϩάΛఆٛͨ͠ਓ͕εΩʔϚΛఆٛͯ͠Δ
64. Embulk Projectʹظ଴͢Δ͜ͱ • େن໛ར༻ʹͱͬͯͷyaml • fileͰશͯΛ؅ཧ͢Δͷ͸ͭΒ͍ • ಉ͡yaml಺Ͱ΋؅ཧऀ͕ҟͳΔ • ಉ͡yaml಺Ͱ΋มߋස౓͕ҟͳΔ
65. Embulk Projectʹظ଴͢Δ͜ͱ • templateػೳ͚ͩͰͳ͘ɺΑΓଟ࣍ݩʹ؅ཧͰ ͖Δ࢓૊Έ͕ඞཁͳͷͰ͸ʁ • ҰํͰɺ݁ہɺۀ຿ϑϩʔʹΑΔͷͰΈΜͳࣗ ࡞͢Δͷ͔΋͠Εͳ͍ͱ΋ࢥͬͯΔ
66. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
67. ଍Γͳ͍͜ͱ2
68. ୯ମॲཧ ϑϨʔϜϫʔΫ
69. Embulk͸Bulkload͢Δ͚ͩ • ઃఆ௨Γʹɺ͋Δσʔλιʔε͔Βɺ͋Δσʔ λιʔε΁σʔλΛϩʔυ͢Δ • Input / Outputͷঢ়ଶʹ͸ڵຯ͕ແ͍
70. ͔ͩΒEmbulkͷ୲อ͢Δႈ౳ੑ ͸͜͏ͳΔ • 1ճ࣮ߦ͠Α͏ͱͨ͠Bulkloadʹର͢Δႈ౳ੑΛ ୲อ͠Α͏ͱ͢ΔʢPlugin࣍ୈʣ • ౰વ͚ͩͲɺ֨ೲઌͷσʔλͷ੔߹ੑΛอͭ΋ ͷͰ͸ͳ͍
71. ฐࣾͷࣄ৘ • HDFS͔ΒVertica΁ͷ֨ೲ͸ৗʹΧϥϜΛߜΔʢ͓ ۚΣ…ʣ • ෳࡶͳKPI͕ݟͨ͘ͳͬͨ࣌ʹΧϥϜ௥ՃΛߦ͏ • HDFS্ͷσʔλ͕ඞཁʹͳͬͨλΠϛϯάͰ σʔλͷ࠶ϩʔυ͕ൃੜ => ࣄલʹ֨ೲൣғͷσʔλ࡟আ͕ඞཁ
72. ฐࣾͷࣄ৘ • File֨ೲ͕׬͍ྃͯ͠Δ͔Ͳ͏͔Λ `_SUCCESS` ͱ͍͏ϑΝΠϧΛಉ֊૚ʹஔ͘͜ͱͰ؅ཧ • `_SUCCESS` ͕ͳ͚Ε͹ॲཧதͷσʔλͱ ೝࣝ͞ΕΔ => ࣄલʹϑΝΠϧͷଘࡏ֬ೝͱࣄޙʹϑΝΠϧͷ put͕ඞཁ
73. ͱ͍͏͔ • ձࣾʹΑͬͯϧʔϧ͸ҧ͑Ͳɺࣄલॲཧࣄޙॲ ཧ͸ઈର͍Δ͸ͣ
74. ͭͬͨ͘ • Embulk Wrapper • Embulkʹ଍Γͳ͍ࣄલɾࣄޙͷ୯ମॲཧΛ αϙʔτ͢ΔWrapper • YAMLͷઃఆϑΝΠϧʹج͍࣮ͮͯߦ͞ΕΔ • action × storage ͱ͍͏୯ҐͰpluginΛॻ͘
75. ͪΐͬͱ͚ͩ Embulk Wrapper঺հ આ໌༻ͷYaml͕ ؒԆͼ͢ΔͷͰ ΞϯΧʔ࢖͍·͢ɻ
76. actionͱ͍ actionͱ͍͏୯ҐͰ ॲཧΛఆٛ
77. actionΛarrayͰఆٛ͠ γʔέϯγϟϧʹ࣮ߦ
78. embulkͷ࣮ߦ
79. ࡞ͬͨActions • vertica#delete • vertica#check_null • vertica#glance (νϥݟ) • s3#poll • s3#remove • ….
80. Embulk Projectʹظ଴͢Δ͜ͱ • Embulkʹ͸୯ମॲཧϑϨʔϜϫʔΫඞཁෆՄܽ • ґଘؔ܎ͷͳ͍୯ମॲཧΛߦ͏ϓϥάΠϯػߏ ͕Embulkʹ಺ଂ͞ΕΕ͹ɺEmbulkͷΈͰߦ͑Δ ͜ͱͷ෯͕͔ͳΓ޿͕ΔͷͰ͸ͳ͍ͩΖ͏͔ʁ
81. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
82. ଍Γͳ͍͜ͱ3
83. δϣϒΩϡʔ
84. Embullk ͷ࣮ߦΛ Ͳ͏؅ཧ͠Α͏͔ • େن໛ར༻͢ΔͳΒδϣϒ؅ཧͷ࢓૊Έ͕ඞਢ • ౰વ͚ͩͲ Embulk ʹͦΜͳػೳ͸ͳ͍ • ࣗ෼Ͱ࡞Δ͔طଘͷผͷԿ͔Λ࢖͏͔͠ແ͍
85. ݁ہɺͭͬͨ͘ • Redis/Sidekiq Λ࢖ͬͨfifoͳδϣϒΩϡʔ • εέδϡʔϧ࣮ߦ͸ɺεέδϡʔϥ͕࣌ؒʹͳΔ ͱ job Λ enqueue • Adhoc࣮ߦ͸ɺϢʔβʔ͕δϣϒ࣮ߦϘλϯΛԡ ͢ͱ job Λ enqueue => ͍͢͝ී௨ͷδϣϒΩϡʔ͆
86. Workerαʔόʔ(Embulk࣮ߦ؀ڥ) • 1αʔόʔ͋ͨΓɺEmbulk 1 process • Embulk͸CPUΛ࢖͑Δ΄Ͳߴ଎ʹͳΔͨΊ • 24 CPU, memory 60 GB αʔόʔ 6୆Ͱฒྻ࣮ߦ
87. ࠓͷͱ͜Ζࠔͬͯͳ͍ͷ͕ͩ • ഉଞॲཧ͕ͪΌΜͱͰ͖͍ͯͳ͍ • ༏ઌॲཧ͕ग़͖ͯͦ͏ • graceful restartͰ͖ͯͳ͍ • … => Ұॠߟ͑Δ͚ͩͰग़͖ͯͦ͏ͳ໰୊͕͍ͬͺ͍
88. ͜Ε͸ྲྀੴʹEmbulkʹ͸ ظ଴͸ͯ͠ͳ͍ • ॻ͍͚ͨͲEmbulkʹ଍Γͳ͍͜ͱͱ͍͏ΑΓ Batch؀ڥʹ଍Γͳ͍͜ͱ͔ͬͯΜ͡Ͱͨ͠ • ͱ͸͍͑ɺEmbulkΛར༻͢Δ্Ͱඞཁͳ͜ͱͳ ͷͰϕετϓϥΫςΟε୳ͯ͠·͢
89. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
90. ଍Γͳ͍͜ͱ4
91. ฒྻ਺੍ޚ
92. Embulkͷฒྻ਺੍ޚ • ݱঢ়LocalExecutor͸InputͷϑΝΠϧ਺Ͱશମͷ ฒྻ਺Λ੍ޚ͍ͯ͠Δ • FileInputPluginܧঝͰͳ͚Ε͹ฒྻ਺͸جຊతʹ 1ͭ
93. ฐࣾͷࣄ৘ • HDFSʹ֨ೲ͞Ε͍ͯΔϑΝΠϧ͸1ϑΝΠϧ͕ ڊେ • namenodeͷϝλ৘ใΛۃྗগͳ͘͢ΔͨΊ • ϑΝΠϧ͕෼͔Ε͍ͯͳ͍ͨΊɺϚϧνεϨου Ͱಈ͔ͣɺCPUΛશવ࢖͑ͳ͍
94. ͭͬͨ͘ • embulk-input-hdfs • ಉ͡ϑΝΠϧͷinput streamΛฒྻ਺෼ੜ੒ ͠ɺඞཁൣғͷΈread͢Δ࢓૊Έ • configʹॻ͔Εͨ਺ʹϑΝΠϧΛ෼ׂ • ೚ҙͷฒྻ਺Ͱembulk࣮ߦ͕Մೳʹʂ
95. ͜ΕͰCPU ϑϧʹ࢖͑Δͧʂ
96. ͱࢥͬͨΒ໰୊͕ • OutputͰ͋ΔVertica • Session਺͕૿͑͗͢ΔͱVertica͕ෆ҆ఆʹ • ࠷େ20͘Β͍ʹߜͬͯ͘ΕͱVerticaνʔϜ ͔Βґཔ • InputͰฒྻ਺نఆ͞ΕΔͷʹͲ͏͢Ε͹͑͑Μ ΍Ͱ
97. ͭͬͨ͘ • embulk-output-vertica • ॲཧલʹσʔλ֨ೲ༻ͷthreadΛผ్੾Δ • ֤εϨου͸σʔλ֨ೲ༻ͷthreadʹpageΛ enqueue • σʔλ֨ೲ༻thread͸pageΛdequeue͠ίϐʔͯ͠ ͍͘ • ͜ΕͰ1ճͷίϐʔͰ1session͔͠࢖ΘͣʹࡁΉͧʂ
98. Embulk Projectʹظ଴͢Δ͜ͱ • ฒྻ਺੍ޚ͸΍ͬͺΓExecutorʹ΍ͬͯ΄͍͠ • input / filters / outputόϥόϥʹઃఆ͍ͨ͠ • https://github.com/embulk/embulk/issues/ 232 • ϑΝΠϧ෼ׂͷAPIͱ͔FileInputPluginͷAPIͱ͠ ͍͍ͯ͋ͬͯΜ͡Όͳ͍͔͠Β
99. ଍Γͳ͍͜ͱϥΠϯφοϓ 1. YAML؅ཧ 2. ୯ମॲཧϑϨʔϜϫʔΫ 3. δϣϒΩϡʔ 4. ฒྻ਺੍ޚ 5. ଓ͖͔Β࣮ߦ
100. ଍Γͳ͍͜ͱ5
101. ଓ͖͔Β࣮ߦ
102. Embulkͷػೳͱͯ͠ͷ ʮଓ͖͔Β࣮ߦʯ • Embulk࣮ߦ࣌ʹɺ࣍ʹEmbulkΛ࣮ߦ͢Δ࣌ʹ࢖ ༻͢ΔconfigϑΝΠϧΛੜ੒͓ͯ͘͠ • Input Plugin୯ମͷػೳͱͯ͠ఏڙ͞Ε͍ͯΔ embulk run /path/to/next-config.yml \ -o /path/to/next-config.yml
103. ฐࣾͷࣄ৘ • HDFS্ͷϑΝΠϧʹσʔλ͕௥ه͞Εͯߦ͘͜ ͱ͕ଟ͍ • ϑΝΠϧύε͸มΘΒͳ͍ɺͰ΋ɺඞཁͳσʔ λ͕૿͑ͯΔ • σʔλͷ಺༰͔ΒFilter͢Δ͔Ͳ͏͔൑அ͢ Δඞཁ͕͋Δ
104. ΍Γ͔ͨͬͨ͜ͱ • Outputઌ(Vertica) ΁֨ೲ͞Ε͍ͯΔσʔλΛ֬ ೝ͠ɺInputσʔλΛFiltering͢Δ • SELECT max(id) FROM table; ͷ݁Ռ͔Β • embulk-filter-row ͷconditionੜ੒
105. ͭͬͯ͘ͳ͍ • ୯ମॲཧϑϨʔϜϫʔΫͱ͸ҧ͍ɺॲཧؒͷґ ଘؔ܎͕ଘࡏ͢ΔͨΊɺͬ͘͞ͱ͸Ͱ͖·ͤΜ Ͱͨ͠ٽ • Embulk WrapperͰର৅ൣғΛࣄલʹফͯ͠શͯ ϩʔυ͢Δͱ͍͏ྗٕͰରԠͨ͠
106. Embulk Projectʹظ଴͢Δ͜ͱ • ΄Μͱ͏ͷҙຯͰ੔߹ੑΛอͭʹ͸Outputଆͷ σʔλΛݟʹߦ͘ඞཁ͕͋Δ • ࣮ࡍʹॻ͜͏ͱ͢ΔͱEmbulkͷྖҬΛ௒͑Δ • ࡢ೔ฉ͍͚ͨͲʮdigdagʯͱ͍͏πʔϧ͕ग़ΔΒ ͍͠ΐ • https://github.com/treasure-data/digdag-docs
107. ·ͱΊ
108. ·ͱΊ • Embulkಋೖͯ͠ϝονϟḿͬͯΔ • ৽͍͠σʔλιʔε͕ग़͖ͯͯ΋ίετ௿͘ ಋೖͰ͖Δʂ
109. ·ͱΊ • ͨͩɺ࣮ࡍproductionӡ༻͠Α͏ͱ͢Δͱࡉ͔͍ ͜ͱ͕ؾʹͳͬͯ͘Δ • YAMLͷ࿩ͱ͔ • δϣϒΩϡʔͷ࿩ͱ͔ • લॲཧɺޙॲཧͲ͏͢Δͷͱ͔
110. ·ͱΊ • ࣮ࡍɺEmbulk͚ͩͰղܾ͢ΔΑ͏ͳ࿩Ͱ͸ͳ͍ • ΈΜͳͰ஌ݟͨΊͯϕετϓϥΫςΟε୳͍ͯ͠ ͖·͠ΐ͏ • AdventCalendarۭ͍ͯΔͷͰॻ͖·͠ΐ͏
111. ͝ਗ਼ௌ͋Γ͕ͱ͏ ͍͟͝·ͨ͠ʂ