Amazon Kinesisでストリーム解析 - Node学園12時限目

Node学園12時限目で話します．
Amazon Kinesisをストリーム解析に使えるか検証した時の話です．
http://plaid.co.jp

Transcript

1. Amazon KinesisͰ ετϦʔϜղੳ (ג)ϓϨΠυ @nashibao
2. 0. Who are you?
3. @nashibaoͱ͍͍·͢ ࣲࢁ௚थ ৘೤త෼ࢄܕػցֶशతΤϯδχΞ(31) ! ~201208: େֶͱ͔Ͱ໱ʑͱ͢ΔɽMapReduceͱߦྻ෼ղ͕޷෺ɽ 201209: Node.jsʹ໨֮ΊΔ 201304: ݩָఱ(31)ͱىۀ σʔλղੳ&CRMαʔϏε 201309: ത࢜υϩοϓΞ΢τ
4. (ג)ϓϨΠυͱ͍͍·͢ http://plaid.co.jp EC޲͚ͷղੳ&CRMαʔϏεΛεςϧε։ൃதɽ ! ݱࡏ։ൃ͸ͨͬͨ2ਓ(t-nakamura)ͳͷͰɺ ԠืΛຊ౰ʹ͓଴͓ͪͯ͠Γ·͢ʂ ෲյ͢·Ͱ೑͕৯͑ΔձࣾʂͦΕ͕ϓϨΠυʂ ׻ܴձ͸γϡϥείͰ͢ʂ
5. Amazon KinesisΛετϦʔϜ ղੳʹ࢖͑Δ͔ݕূ͓ͨ͠࿩
6. 1. Kinesisͱ͸ʁ 2. Node.js͔ΒKinesisΛ࢖͏ 3. ύϑΥʔϚϯεݕূ 4. ײ૝
7. 1. Kinesisͱ͸ʁ
8. Amazon Kinesis
9. “a fully managed service for real-time processing of streaming data at massive scale.” ͲͰ͔͍σʔλ͔Β ϦΞϧλΠϜॲཧ͕ग़དྷΔ ϑϧϚωʔδυͳαʔϏε
10. 5JCTF 5JCTF 5JCTF -# -# -# 2TQFWEGT -KPGUKU#RRNKECVKQPU 5VTGCO 2TQFWEGT 2TQFWEGT 2TQFWEGT 2TQFWEGT 哋┮؃䠉㏰׵ا揉ⓧ
11. 5JCTF 5JCTF 5JCTF -# -# -# WEGT -KPGUKU#RRNKECVKQPU 5VTGCO WEGT WEGT WEGT EGT ϑϧϚωʔδυͳ෦෼ ! • 24࣌ؒอ࣋ • ࠶औಘՄʢex: Τϥʔ࣌ʣ • 1000tps/shard put • 5tps/shard get
12. 5JCTF 5JCTF 5JCTF -# -# -# WEGT -KPGUKU#RRNKECVKQPU 5VTGCO WEGT WEGT WEGT EGT ϑϧϚωʔδυͳ෦෼ ! • 24࣌ؒอ࣋ • ࠶औಘՄʢex: Τϥʔ࣌ʣ • 1000tps/shard put • 5tps/shard get • hashKey(ࣗ༝)ͰৼΓ෼͚
13. 5JCTF 5JCTF 5JCTF -# -# -# WEGT -KPGUKU#RRNKECVKQPU 5VTGCO WEGT WEGT WEGT EGT େྔʹॻ͖ࠐΜͰ ·ͱΊͯ औಘ͢ΔͨΊͷΩϡʔ? ॻ͖ࠐΈΛΩϟογϡ͢Δ ड͚ࡼͷΑ͏ͳΠϝʔδ
14. ෛՙूதͷड͚ࡼʹ Ͱ͖Δʁ
15. 5KVG# 5KVG$ &$ &$ ! 5VTGCO #PCN[VKEU 5GOKDCVEJ/CR4GFWEG ! .QIIGT Ý ݱঢ় YGDUQEMGV
16. 5KVG# 5KVG$ &$ &$ ! 5VTGCO #PCN[VKEU 5GOKDCVEJ/CR4GFWEG ! .QIIGT Ý ݱঢ় ෛՙ͕ूத͢Δͱ͞͹͘ͷେมʂ YGDUQEMGV
17. 5KVG# 5KVG$ &$ &$ ! 5VTGCO #PCN[VKEU 5GOKDCVEJ/CR4GFWEG ! .QIIGT Ý Kinesis? YGDUQEMGV
18. 5KVG# 5KVG$ &$ ! 5VTGCO #PCN[VKEU 5GOKDCVEJ/CR4GFWEG Kinesis? -KPGUKU ! Ý YGDUQEMGV
19. 2.Node.js͔Βѻ͏
20. plaidev/kinesis-wrapper.js # install npm install kinesis-wrapper.js # stream ϕʔε mhart/kinesis ! # ΫϥγΧϧͳͷ͕ཉ͍͠ͷɽɽ͔ͩΒॻ͍ͨͷɽɽ plaidev/kinesis-wrapper.js
21. plaidev/kinesis-wrapper.js # install npm install kinesis-wrapper.js # stream ϕʔε mhart/kinesis ! # bacon.js ϕʔε andyhorng/kinesis ! # ΫϥγΧϧͷ͕ཉ͍͠ͷɽɽ͔ͩΒॻ͍ͨͷɽɽ plaidev/kinesis-wrapper.js
22. 1. putRecordͰ౤͛ࠐΉ var kinesis = require('..'); ! # ετϦʔϜͱͭͳ͛Δ var stream = kinesis.stream('hoge'); ! # ‘key’͸MD5ͰϋογϡԽ͞ΕͯγϟʔυʹৼΓ෼͚ΒΕΔ stream.putRecord('key', {date: new Date()}, function(err, res){ console.log(err, res); });
23. 2. getRecordsͰϙʔϦϯά var kinesis = require('..'); ! # ετϦʔϜͱͭͳ͛Δ var stream = kinesis.stream('hoge'); ! # ϙʔϦϯάͯ͠औಘɽγϟʔυ͝ͱ΋ग़དྷ·͢ stream.getRecords(function(err, records){ console.log(new Date(), records[0].val.date); });
24. ͍ͭͰʹ ಉظతʹฒྻԽ͢Δ • async.jsʁ • ͜Ε͔Β͸yieldʂ • node-thunkify • nashibao/co-gate΋ѱ͘ͳ͍Αʂ
25. co-gateϞδϡʔϧ ैདྷͷίʔϧόοΫܗࣜͷؔ਺Λͦͷ··ฒ ྻԽɾಉظతʹॻ͚Δ ICVG ECNNDCEM ECNNDCEM ECNNDCEM ECNNDCEM KP QWV TGUWNVU
26. var co = require('co'); var Gate = require('co-gate'); ! co(function *(){ ! var gate = new Gate(); ! async_function(gate.in()); ! var results = yield gate.out(); ! })(); co-gateϞδϡʔϧ
27. 3. େྔʹ౤͛ࠐΉʢ࠶ʣ co(function *(){ var gate = new Gate(); ! for (var i = 0; i < 100; i++){ stream.putRecord('key', {date: new Date()}, gate.in()); } ! var res = yield gate.out(); }
28. 4. ϙʔϦϯάʢ࠶ʣ co(function *(){ var gate = new Gate(); ! stream.getRecords(gate.in()); ! var records = yield gate.out(); })();
29. 3. KinesisͷύϑΥʔϚϯε
30. ݕূ؀ڥɾ৚݅ • EC2(c3-4xlarge)1୆͔Β౤͛ͯݕূɽ • ΫϥΠΞϯτଆͷλΠϜελϯϓͰtpsΛܭଌ • 1~3shard • ৄ͘͠͸ɺ΋͏গ͠ݕূඞཁ
31. putRecord / shard UQTͱ͔ೖͬͨΓ͢Δɽɽ
32. put & getRecords UQTఔ౓Ͱݻ·ͬͯऔಘͰ͖Δ
33. ࢥͬͨΑΓॻ͖ࠐΊΔɽͰ΋ ͪΐͬͱ஗͍? • max3000~ tps/shard put • ॻ͖ࠐΈ͗͢Δͱ݁ہऔΓग़͢·Ͱʹ͔͔࣌ؒΔɽ • ݁ہ400~700tpsఔ౓·Ͱ͍͚Δ͜ͱ͸֬ೝ • avg1sec get • ϙʔϦϯάࠐΈͰ͜ͷఔ౓ • 200msҎ಺ʹฦͬͯ͘ΔΘ͚͡Όͳ͍ • ॻ͖ࠐΜͰऔΓग़͢·Ͱʹͳ͔ͥϥά͕͋ͬͨΓͳ͔ͬͨΓ
34. ײ૝ • ϑϧϚωʔδυʁ • KA͸ࣗ෼Ͱ؅ཧɽγϟʔυʹϫʔΧʔ৐ͤΒΕͳ͍ͷ͔ɽɽ ! • ϦΞϧλΠϜʁ • ϙʔϦϯά • औΓग़͢ͷʹϥά͕͚ͬ͜͏͋Δ ! • ύϑΥʔϚϯε • write͸ߴ͍ɽγϟʔυ૿΍ͤ͹͔ͳΓ͍͚ͦ͏
35. ͋Γ͕ͱ͏͍͟͝·ͨ͠ɽ
36. http://plaid.co.jp