PowerPoint Presentation

Transcription

1 AWS Black Belt Online Seminar Kinesis アマゾンウェブサービスジャパン株式会社 ソリューションアーキテクト半場光晴

2 自己紹介 名前 半場 光晴 はんば みつはる 所属 アマゾン ウェブ サービス ジャパン 株式会社 技術統括本部 メディア エンターテインメント ソ リューション部 ソリューション アーキテクト 好きなAWSサービス Kinesis S3 EMR

3 内容についての注意点 本資料では 2017 年 10 月 11 日時点のサービス内容および価格についてご説明しています 最新の情報は AWS 公式ウェブサイト ( にてご確認ください 資料作成には十分注意しておりますが 資料内の価格と AWS 公式ウェブサイト記載の価格に相違があった場合 AWS 公式ウェブサイトの価格を優先とさせていただきます 価格は税抜表記となっています 日本居住者のお客様が東京リージョンを使用する場合 別途消費税をご請求させていただきます AWS does not offer binding price quotes. AWS pricing is publicly available and is subject to change in accordance with the AWS Customer Agreement available at Any pricing information included in this document is provided only as an estimate of usage charges for AWS services based on certain information that you have provided. Monthly charges will be based on your actual use of AWS services, and may vary from the estimates provided.

4 Kinesis Firehose が 東京リージョンでも 利用可能になりました available-in-asia-pacific-tokyo-eu-frankfurt-and-us-east-ohio-regions/

5 このセミナーの内容 1. Kinesis の特徴 1. Kinesis Streams 2. Kinesis Firehose 3. Kinesis Analytics 2. Kinesis の適用例 3. まとめ

6 このセミナーの内容 1. Kinesis の特徴 1. Kinesis Streams 2. Kinesis Firehose 3. Kinesis Analytics 2. Kinesis の適用例 3. まとめ

7 Kinesis の特徴

8 Kinesis プラットフォームストリームデータを収集 処理するためのフルマネージドサービス群 Kinesis Streams ストリームデータを処理するためのアプリケーションを独自に構築 Kinesis Firehose ストリームデータを S3, Redshift, ES へ簡単に配信 Kinesis Analytics ストリームデータを標準的な SQL クエリでリアルタイムに分析

9 ストリーミングデータの利用シナリオ STEP 1 STEP 2 STEP 3 シナリオデータの収集と ETL 継続的なメトリクス計算リアルタイム分析と応答 データタイプ アプリケーションログ / コンピューターメトリクス / クリックストリーム / センサーデータ / マーケットデータ... 広告 / マーケティング 配信や入札データの収集 コンバージョンレート / 収益 / カバレージの計算 ユーザー行動に応じた配信 入札エンジンの最適化 IoT テレメトリーデータの収集 デバイスやサーバーの稼働状態の計算 稼働状態予測やアラート 通知の発行 ゲーム 顧客行動データの収集 成功率 / トランジションレート / クリックレートの計算 リーダーボードやマッチメイクの最適化 コンシューマー向け 顧客行動データの収集 ページビュー / クリックレートの計算 レコメンデーションエンジンの最適化 運用セキュリティ システムメトリクスの収集システムログ分析異常検知

10 Kinesis プラットフォームの全体的な特徴リアルタイムなデータストリームを簡単に捉え 配信し 処理できます 簡単なプロビジョニング デプロイ 管理 弾力のある拡張性とサーバーレス リアルタイムなレイテンシー 先払いなしの従量課金 ユースケースに合わせた適切なサービスの選択が可能

11 Kinesis の特徴 Kinesis Streams

12 Kinesis Streams ストリームデータを処理するためのアプリケーションを独自に構築 1 管理が容易 必要なキャパシティをセットしてストリームを作成するだけで利用可能 スループットやデータ量の変化に応じてスケール 2 独自のリアルタイムアプリケーション Kinesis Client Library, Apache Spark/Flink, AWS Lambda 等を利用してストリーム処理を実装 3 低コスト あらゆるスケールのワークロードで高いコスト効果

13 Kinesis Streams アーキテクチャ概要 順序つきイベントストリームとして複数のアプリケーションから同時アクセス可能 集約して S3 にアーカイブ 数百万のソースが 1 時間あたり数百 TB のデータを生成 リアルタイムダッシュボード / アラート Front End point end 認証 認可 AZ AZ AZ 機械学習 / スライディングウィンドウ分析 3 アベイラビリティゾーンの永続ストレージに強い整合性でデータを複製 データウェアハウスにロード

14 Kinesis Streams 主要なコンセプト データレコード Kinesis Streams エンドポイント ストリームシャード 0 シャード 1 シャード..N S3 DynamoDB Redshift EMR データ送信側 データの種類や処理の用途に応じて ストリーム を作成 ストリームは 1 つ以上の シャード で構成保存されるデータの単位を データレコード と呼び 保持期間はデフォルトで 24 時間 / 最長で 7 日間 1 データレコードの最大サイズは 1 MB データ送信側のキャパシティは 1 シャードあたり秒間 1 MB もしくは 1,000 PUT レコードデータ処理側のキャパシティは 1 シャードあたり秒間 2 MB もしくは 5 回の読み取りトランザクションストリーム内のシャード数を増減することでスループットをコントロール データ処理側

15 Kinesis Streams データレコードの分散 データ データレコード パーティションキー パーティションキーのハッシュで分散 * ストリーム シャード 0 "HashKeyRange" { "EndingHashKey": " ", "StartingHashKey": "0" } シャード 1 "HashKeyRange" { "EndingHashKey": " ", "StartingHashKey": " " } 0 ハッシュキーの範囲 2^128-1 * ハッシュキーを直接指定することも可能 データ入力時に指定するパーティションキー ( 最長 256 文字 ) で保存先のシャードが決定 MD5 ハッシュ関数でパーティションキーを 128 ビット整数値のハッシュキーに変換 ハッシュキーの範囲に対応したシャードにデータレコードをマップ

16 Kinesis Streams データレコードの順序性 ストリーム SequenceNumbe r 32 SequenceNumbe r 26 SequenceNumbe r 25 SequenceNumbe r 17 シャード 0 データレコード データレコード SequenceNumbe r 35 SequenceNumbe r 15 SequenceNumbe r 12 SequenceNumbe r 11 シャード 1 データレコード 全てのデータレコードにはシーケンス番号がアサインされる ストリーム内の全シャード間でユニーク シャード内で時間の経過とともに単調増加 ( シャード間では単調増加しない ) データ処理時にシーケンス番号でレコード取得開始ポジションを指定可能

17 Kinesis Streams をサポートするプロデューサー コンシューマー プロデューサー ( データ送信側 ) コンシューマー ( データ処理側 ) AWS SDK Kinesis Producer Library Kinesis Agent AWS IoT Kinesis Log4j Appender Fluentd Kinesis Firehose Get* API Kinesis Client Library Kinesis Analytics AWS Lambda EMR CloudWatch Events/Logs Apache Storm

18 Kinesis Agent プロデューサー ( データ送信側 ) Kinesis サービスにデータを簡単に収集して取り込む OSS のスタンドアロン Java アプリケーション エージェントでモニタリングするファイルのパターンと送信先ストリームを指定 ファイルのローテート処理 ポインティングのチェック 失敗時の再試行を管理 フォーマット変換やログパースなどの前処理機能を提供 送信前のバッファリングが可能 すべてのデータを信頼性の高いタイムリーかつシンプルな方法で配信 CloudWatch へメトリクスを送信 Kinesis Streams と Kinesis Firehose どちらへも送信可能 /etc/aws-kinesis/agent.json { } "kinesis.endpoint": " "firehose.endpoint": " "flows": [ { "filepattern": "/tmp/app1.log*", "kinesisstream": "yourkinesisstream" }, { "filepattern": "/tmp/app2.log*", "deliverystream": "yourfirehosedeliverystream" } ]

19 Kinesis Producer Library (KPL) プロデューサー (データ送信側) Kinesis Streams にデータを送信 する OSS の補助ライブラリ KPL は C++ で書かれており メインプ ロセスの子プロセスとして実行される Aggregation 複数件のデータを 1 データ レコードに集約して送信可能 Collection 複数のレコードをバッファリ ングして送信 キャパシティ超過時のリトライをサポート タイムアウト時間や送信時に利用するコネ クション数を設定可能 パフォーマンスメトリクスを CloudWatch に自動送信

20 Fluent plugin for Kinesis V2 出た! プロデューサー ( データ送信側 ) Kinesis Streams と Kinesis Firehose にイベントを送信する OSS の Fluentd 出力プラグイン KPL 集約レコードフォーマットもサポート fluent-plugin-kinesis gem に以下 3 つの出力プラグインが含まれいる 1. Kinesis_streams 2. Kinesis_firehose 3. Kinesis_streams_aggregated Fluentd をログ収集に使っているなら このプラグインを追加するだけで Kinesis へのデータ投入がすぐにできる

21 Kinesis Data Generator (KDG) プロデューサー (データ送信側) Kinesis Data Generator (KDG) を利用して Kinesis Streams または Kinesis Firehose にテストデータを簡単に送 信できる HTML と JavaScript で実装されたOSSのテスト用プロデューサーUI GithubにホストされたUIを利用することが可能 S3 静的ウェブサイトホスティングを利用するなども可能

22 Kinesis Client Library (KCL) コンシューマー ( データ処理側 ) KCL を利用して Kinesis アプリケーションを作成できる Java Ruby Python Node.js の開発に利用できる OSS のクライアントライブラリ EC2 インスタンスなどにデプロイして利用可能 KCL アプリは 3 つのコンポーネントを含んでいる 1. Record Processor Factory - 2. のレコードプロセッサーを作る 2. Record Processor - Kinesis Streams のシャードから取り出したデータを処理するプロセッサーの単位 3. Worker - 個々のアプリケーションインスタンスとマッピングする処理単位

23 Kinesis Client Library のステート管理 DynamoDB Shard-0 Worker -A shardid Shard-0 check point TRIM _HORIZO N lease Owner Worker-A lease Counter Shard-1 Worker -B Shard-1 4 Worker-B DynamoDB テーブルを利用してアプリケーションの状態を追跡シャードとワーカーのマッピングを調整処理されたレコードのチェックポイントを作成ワーカーインスタンスの増減やシャードの分割 / 結合に追従

24 ストリームデータの利用シナリオに沿った AWS での事例 STEP 1 データの収集と ETL STEP 2 継続的なメトリクス計算 STEP 3 リアルタイム分析と応答 引用元

25 Kinesis Streams 料金 東京リージョンの場合 料金構成 料金 シャード時間 (1 MB/sec 取込, 2MB/sec 放出 ) $ PUT ペイロードユニット (25 KB), 1,000,000 PUTs 毎 $ 延長データ保持期間 ( 最長 7 日間 ), シャード時間毎 $0.026

26 Kinesis の特徴 Kinesis Firehose

27 Kinesis Firehose ストリームデータを S3 Redshift ES へ簡単に配信 1 管理不要 アプリケーションの実装やインフラストラクチャーの管理を一切行わずに S3 / Redshift / ES にデータを配信可能 2 データストアとダイレクトに統合 シンプルな設定でストリームデータのバッチ化 圧縮 暗号化が可能最短 60 秒でデータを配信 3 シームレスにスケール データのスループットに応じて自動的にスケール 4 サーバレス ETL Lambda を利用したストリームデータの変形が可能

28 Kinesis Firehose 主要なコンセプト データ送信側 データレコード Kinesis Firehose エンドポイント S3 配信ストリーム Redshift 配信ストリーム ES 配信ストリーム S3 Redshift ES 配信先に応じて 配信ストリーム を作成 シャードの作成やパーティションキーの指定不要 1 データレコードの最大サイズは 1 MB 制限なしにスケールするよう設計 米国東部 ( バージニア北部 オハイオ ) 米国西部( オレゴン ) 欧州( アイルランド フランクフルト ) アジアパシフィック ( 東京 ) リージョンで利用可能

29 Kinesis Firehose から S3 へ プロデューサー ソースレコード Firehose 配信ストリーム 変換済みレコード 変換失敗 配信先 S3 バケット ソースレコード バックアップ S3 バケット

30 Kinesis Firehose から Redshift へ プロデューサー ソースレコード Firehose 配信ストリーム 変換済みレコード 変換失敗 中間 S3 バケット 変換済みレコード COPY Redshift クラスター 配信失敗 ソースレコード バックアップ S3 バケット

31 Kinesis Firehose から Elasticsearch Service へ プロデューサー ソースレコード Firehose 配信ストリーム 変換済みレコード Elasticsearch クラスター 変換失敗 配信失敗 ソースレコード バックアップ S3 バケット

32 Kinesis Streams から Firehose へ直接ストリームデータを送ることが可能に BEFORE AFTER Data producer Kinesis Streams Data consumer Kinesis Firehose Data producer Kinesis Streams Kinesis Firehose Kinesis Streams をデータソースに指定すると 他のデータソースを 同一の Kinesis Firehose 配信ストリームに送ることはできない Kinesis Streams をデータソースに指定すると その Kinesis Firehose 配信ストリームに対して PutRecord や PutRecordBatch ができなくなる データソースに指定した Kinesis Streams に PutRecord や PutRecords すれば良い 一方で 1 つの Kinesis Streams に対しては 複数の Kinesis Firehose 配信ストリームやその他のコンシューマーからデータレコードを読むことができる Kinesis Firehose 配信ストリームは Kinesis Streams の LATEST からデータを読み始める Kinesis Producer Library を利用して Kinesis Firehose 配信ストリームにデータレコードを送りたい場合のワークアラウンドにも ( より簡単に ) なる

33 Kinesis Firehose 料金 東京リージョンの場合 料金構成 最初の 500 TB / 月 ( データ取込分 ) 次の 1.5 PB / 月次の 3 PB / 月以降 5 PB / 月 料金 $0.036 $0.031 $0.025 応相談 料金は Kinesis Firehose に取り込まれたデータの量に基づく データの量は このサービスに送信したデータレコードの数に 直近の 5 KB の倍数に切り上げた各レコードのサイズを乗算した値として計算される 例 - 各データレコードが 42 KB の場合は 取り込まれたデータとしては 45 KB としてそれぞれカウントされる

34 Kinesis の特徴 Kinesis Analytics

35 Kinesis Analytics ストリームデータを標準的な SQL クエリでリアルタイムに分析 1 標準 SQL 複雑な処理フレームワークやプログラミング言語の学習不要 2 リアルタイム分析アプリケーション 秒以下のレイテンシーでストリームデータを連続的に分析 3 弾力的にスケール データのスループットに応じて処理能力を伸縮 オペレーションの介入不要

36 Kinesis Analytics 主要なコンセプトアプリケーション アプリケーション内部 ( 出力 ) ストリーム 前処理 SQL 参照テーブル ストリーミングソース (Kinesis Streams または Kinesis Firehose) アプリケーション内部 ( 入力 ) ストリーム 分析単位に アプリケーション を作成し 入力 / 出力となる ストリーミングソース / デスティネーション を設定 ストリーミングソース / デスティネーションをアプリケーション内部の 入力 / 出力ストリーム にマッピング SQL クエリ実行の前処理として Lambda 関数の指定が可能 アプリケーション内部の入力ストリームを分析し 結果を出力ストリームへ出力する SQL を記述 1 入力行の最大サイズは 50 KB/ 参照ソースの最大サイズは 1 GB クエリの複雑さとデータのスループットに応じて処理能力 (KPU Kinesis Processing Units) を自動伸縮 米国東部 ( バージニア北部 )/ 米国西部 ( オレゴン )/ 欧州 ( アイルランド ) リージョンで利用可能 参照ソース アプリケーション内部エラーストリーム ストリーミングデスティネーション (Kinesis Streams または Kinesis Firehose)

37 Lambda による前処理 SQL クエリ実行前に 指定した Lambda 関数による前処理が実行される データレコードの情報の補完やフォーマットの変換処理などの用途が例としてあげられる 補完処理の例 - S3 以外のリソースにあるデータソースを参照して データレコードの情報を補完する 変換処理の例 - GZIP や KPL フォーマットなどから CSV や JSON へ変換する 前処理の結果は 3 種類のステータスに分類できる Ok - 前処理に成功 処理済みのデータレコードが SQL クエリへ流れる Dropped - 前処理によって意図的にデータレコードを排除 SQL クエリの対象にしない ProcessingFailed - 前処理に失敗 未処理のデータレコードがエラーストリームへ流れる この前処理用途の汎用的なLambda 関数のブループリントをいくつか提供している

38 アプリケーション内部ストリームとポンプ 内部 ( 入力 ) ストリーム ポンプ 内部 ( 出力 ) ストリーム SQL SOURCE_SQL_STREAM_001 STREAM_PUMP DESTINATION_SQL_STREAM CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), sector VARCHAR(12), change REAL, price REAL); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM SELECT STREAM ticker_symbol, sector, change, price FROM "SOURCE_SQL_STREAM_001 ;

39 様々なタイムスタンプ イベント時刻 収集時刻 処理時刻 your_own_event_time_column approximate_arrival_time rowtime CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM SELECT STREAM your_own_event_time_column, approximate_arrival_time, rowtime FROM "SOURCE_SQL_STREAM_001";

40 ウィンドウ問い合わせ タンブリングウィンドウ FROM "SOURCE_SQL_STREAM_001" GROUP BY ticker_symbol, FLOOR("SOURCE_SQL_STREAM_001".rowtime TO MINUTE); スライディングウィンドウ FROM "SOURCE_SQL_STREAM_001" WINDOW last_hour AS (PARTITION BY ticker_symbol RANGE INTERVAL '1' HOUR PRECEDING), last_two_rows AS (PARTITION BY ticker_symbol ROWS 2 PRECEDING);

41 シンプルなアラート CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), avg_change DOUBLE); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM SELECT STREAM ticker_symbol, avg_change FROM ( SELECT STREAM ticker_symbol, AVG(change) OVER w1 AS avg_change FROM "SOURCE_SQL_STREAM_001 WINDOW w1 AS (PARTITION BY ticker_symbol RANGE INTERVAL '10' SECOND PRECEDING)) WHERE ABS(avg_change) > 1; 内部 ( 出力 ) ストリーム "DESTINATION_SQL_STREAM" を宣言ポンプ "STREAM_PUMP" を宣言 10 秒のスライディングウィンドウから ticker_symbol ごとに変化量の平均値を算出 変化量の平均値の絶対値が 1 を超える行のみを抽出

42 参照テーブルの結合 CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), company VARCHAR(20), sector VARCHAR(12), change DOUBLE, price DOUBLE); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM ticker_symbol, c."company", sector, change, price FROM "SOURCE_SQL_STREAM_001" LEFT JOIN "CompanyName" c ON "SOURCE_SQL_STREAM_001".ticker_symbol = c."ticker"; ( 参照テーブル "CompanyName" をアプリケーションに事前追加 ) 内部 ( 出力 ) ストリーム "DESTINATION_SQL_STREAM"/ ポンプ "STREAM_PUMP" を宣言内部 ( 入力 ) ストリーム "SOURCE_SQL_STREAM_001" に参照テーブルを外部結合ティッカーシンボルが一致した場合に参照テーブルから "Company" カラムの値を出力

43 ウィンドウやタイムスタンプを利用した ストリーム処理のAWSでの事例 STEP 1 データの収集と ETL STEP 3 STEP 2 継続的なメトリクス計算 リアルタイム分析と応答 引用元

44 Kinesis Analytics 料金 オレゴンリージョンの場合 料金構成 料金 KPU の平均量 / 時間 $0.110 KPU (Kinesis Processing Unit) 4 GB のメモリ 1 vcpu のコンピューティング 対応するネットワークリソースで構成されるストリーム処理能力の単位 クエリの複雑性やメモリと計算能力の要求が応答に応じて異なるため Kinesis Analytics では分析を完了するために必要な KPU が自動的 かつ 伸縮自在にスケールされる

45 このセミナーの内容 1. Kinesis の特徴 1. Kinesis Streams 2. Kinesis Firehose 3. Kinesis Analytics 2. Kinesis の適用例 3. まとめ

46 Kinesis の適用例

47 Kinesis とストリーム処理 はじめの一二三歩 Lambda function 1 可視化 Data producer Kinesis Firehose Elasticsearch Service 2 仮説検証 Lambda function S3 Athena 3 自動化 2017, S3 Web Services, Inc. or its Affiliates. Kinesis All rights reserved. (Reference data source) Analytics Kinesis Streams Lambda function SNS

48 Kinesis とラムダアーキテクチャ SERVING APPs SPEED Kinesis Streams Kinesis Analytics Kinesis Streams AWS Lambda ES DynamoDB SNS Data producer Kinesis Firehose EMR QuickSight BATCH S3 AWS Glue Redshift RDS ElastiCache Any on EC2 Athena

49 1 Kinesis と IoT センサーデータ データ保管 対象シーン センサーデータなどのバックアップ 機械学習などのモデル作成に向けた準備 ルールエンジン 条件 128 KB 以内のメッセージ IoT AWS 2 他システム連携 対象シーン AWS 内外のシステムと API 連携するパターン 条件 センサーデータをバッファリングして他システムと連携する デバイズ IoT Kinesis Streams Kinesis Firehose AWS Lambda S3 他システム 3 ストリーム処理 対象シーン センサーデータに対して異常検知 スライディングウィンドウ処理などをストリームで処理したい 条件 要件に応じて Spark on EMR なども検討 Kinesis Analytics

50 このセミナーの内容 1. Kinesis の特徴 1. Kinesis Streams 2. Kinesis Firehose 3. Kinesis Analytics 2. Kinesis の適用例 3. まとめ

51 まとめ

52 Kinesis プラットフォームストリームデータを収集 処理するためのフルマネージドサービス群 Kinesis Streams ストリームデータを処理するためのアプリケーションを独自に構築 Kinesis Firehose ストリームデータを S3, Redshift, ES へ簡単に配信 Kinesis Analytics ストリームデータを標準的な SQL クエリでリアルタイムに分析

53 Kinesis プラットフォームの全体的な特徴リアルタイムなデータストリームを簡単に捉え 配信し 処理できます 簡単なプロビジョニング デプロイ 管理 弾力のある拡張性とサーバーレス リアルタイムなレイテンシー 先払いなしの従量課金 ユースケースに合わせた適切なサービスの選択が可能

54 Kinesis とストリーム処理 はじめの一二三歩 Lambda function 1 可視化 Data producer Kinesis Firehose Elasticsearch Service 2 仮説検証 Lambda function S3 Athena 3 自動化 2017, S3 Web Services, Inc. or its Affiliates. Kinesis All rights reserved. (Reference data source) Analytics Kinesis Streams Lambda function SNS

55 さいごに Kinesis が提供しているもの ストリームデータを取り込み 継続的に処理するアプリケーションを作るためのマネージドサービス Kinesis でデータを集めよう! Kinesis についてもっと知りたい方はこちら 公式ドキュメント - AWS クラウドサービス活用資料集 - AWS ブログ - What's New in Kinesis - AWS Big Data Blog (Category: Kinesis) - Kinesis Forum -

56