React + API Gateway + Lambda + DynamoDB で動画の再生回数を取得する仕組みを作ってみた

こんにちは、大前です。

普段からメディア系のプリセールスに参加する事が多いのですが、「動画の視聴回数を取得したい」というご要望を頂く事が多いです。

視聴回数を取得する為にはフロント側の実装が不可欠である為、AWS だけでは実現する事が出来ないのですが、一度自分で作ってみようと思いやってみましたので、ブログにしていきます。

構成

AWS 側としては API Gateway + Lambda + DynamoDB の構成となります。

フロント側で動画の再生をトリガーにして API Gateway に対して API をコールし、Lambda から DynamoDB に格納している値（再生回数）を更新します。

フロントは React で実装します。

やってみた

AWS 側構築

DynamoDB

まず、DynamoDB のテーブルを 2つ作成します。

• ViewCount
  • 動画の再生回数を格納するテーブル
  • プライマリキー ... video_name（ビデオ名）

• RequestHistory
  • 同一 IP からリロードを繰り返された際にいたずらに再生回数がカウントされる事を防ぐ為の情報を格納するテーブル
  • プライマリキー ... video_name（ビデオ名）
  • ソートキー ... ip_address（IP アドレス）

Lambda

以下のソースコードを実装しました。ランタイムは Python3.8 です。

処理失敗時のロールバック等は実装していませんので、あくまで参考程度にご利用ください。

「TABLENAME_COUNT」、「TABLENAME_HISTORY」には作成した DynamoDB のテーブル名をそれぞれ指定ください。

import os
import boto3
from boto3.dynamodb.conditions import Key
import json
import base64
from datetime import datetime, timedelta, timezone
import logging

TABLENAME_COUNT = os.environ['TABLENAME_COUNT']
TABLENAME_HISTORY = os.environ['TABLENAME_HISTORY']
COUNT_INTERVAL = os.environ['COUNT_INTERVAL']

logger = logging.getLogger()
logger.setLevel(logging.INFO)

JST = timezone(timedelta(hours=+9), 'JST')

dynamoDB = boto3.resource('dynamodb')

def excute_countup(target):
    firstCount = True
    
    # ViewCountテーブル情報を取得
    table_get = dynamoDB.Table(TABLENAME_COUNT)
    result = table_get.query(
        KeyConditionExpression = Key('video_name').eq(target),
        )
    logger.debug('[excute_countup] query result: ' + str(result))
    
    # 既存のカウント数を保持
    if len(result['Items']) != 0:
        oldCount = result['Items'][0]['view_count']
        firstCount = False
        
    # ViewCountテーブルを更新
    table_update = dynamoDB.Table(TABLENAME_COUNT)
    if firstCount:
        table_update.put_item(
            Item = {
                'video_name' : target,
                'view_count' : 1,
                'last_updated' : str(datetime.now(JST))
            })
    else:
        newCount = oldCount + 1
        logger.debug('[excute_countup] newCount: ' + str(newCount))
        table_update.update_item(
            Key = {
                'video_name' : target
            },
            UpdateExpression="set view_count = view_count + :val, last_updated = :lu",
            ExpressionAttributeValues = {
                ':val' : 1,
                ':lu' : str(datetime.now(JST))
            }
        )
    
    # どちらか失敗したらロールバック
    return

def update_history(target, sourceIp, timeEpoch):
    
    # RequestHistoryの最新情報を取得
    table_history_get = dynamoDB.Table(TABLENAME_HISTORY)
    result = table_history_get.query(
        KeyConditionExpression = Key('video_name').eq(target) & Key('ip_address').eq(sourceIp),
        )
    logger.debug('[excute_countup] history query result: ' + str(result))
    
    # 項目が存在しなければ作成
    table_history_update = dynamoDB.Table(TABLENAME_HISTORY)
    if len(result['Items']) == 0:
        table_history_update.put_item(
            Item = {
                'video_name' : target,
                'ip_address' : sourceIp,
                'timeEpoch' : timeEpoch
            })
    else:
        table_history_update.update_item(
            Key = {
                'video_name' : target,
                'ip_address' : sourceIp
            },
            UpdateExpression="set timeEpoch = :te",
            ExpressionAttributeValues = {
                ':te' : timeEpoch
            }
        )

    return

def get_target(body, sourceIp, timeEpoch):
    
    target = body[1]
    logger.info('[get_target] target: ' + target)
    
    # DynamoDBのRequestHistoryテーブルから同一リクエストを検索
    table = dynamoDB.Table(TABLENAME_HISTORY)
    result = table.query(
        KeyConditionExpression = Key('video_name').eq(target) & Key('ip_address').eq(sourceIp),
        ScanIndexForward = False
        )
    logger.debug('[get_target] query result: ' + str(result))
    
    # 過去にリクエストがなければカウント対象
    if len(result['Items']) == 0:
        logger.info('[get_target] new request')
        return target
    
    # 1分以内の再リクエストだったらカウントしない
    oldTimeEpoch = result['Items'][0]['timeEpoch']
    logger.debug('[get_target] timeEpoch diff(minutes) : ' + str((float(timeEpoch/1000) - float(oldTimeEpoch/1000)) / 60))
    if (float(timeEpoch/1000) - float(oldTimeEpoch/1000)) / 60 < float(COUNT_INTERVAL):
        logger.info('don\'t count')
        return
        
    return target

def return_response(message):
    return {
        'statusCode': 200,
        'headers': {
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Headers": "*"
        },
        'body': json.dumps({"result": message})
    }

def lambda_handler(event, context):
    logger.debug('[main] event: ' + str(event))
    
    body = base64.b64decode(event['body']).decode('utf-8').split('=')
    sourceIp = event['requestContext']['http']['sourceIp']
    timeEpoch = event['requestContext']['timeEpoch']
    logger.debug('[main] sourceIp: ' + str(sourceIp) + ', timeEpoch : ' + str(timeEpoch))
    
    target = get_target(body, sourceIp, timeEpoch)
    if target is None:
        return return_response('Didn\'t count')
    
    excute_countup(target)
    update_history(target, sourceIp, timeEpoch)
    
    return return_response('count success')

ロジック

ロジックは以下です。

1. リクエストから 動画URL、IP アドレス、リクエスト時刻（UNIX 時間）を取得
2. 動画 URL と IP アドレスを用いて DynamoDB の RequestHistory テーブルを検索し、指定された時間内に再生がなかったかチェック（環境変数 COUNT_INTERVAL に設定した値（分））
3. 2.のチェックで問題なければ DynamoDB の ViewCount テーブルを更新（インクリメント）

アタッチする IAM ロール

この Lambda 関数では以下操作を行うので、必要なポリシーはアタッチしてください。

• DynamoDB
  • query
  • put item
  • update item

API Gateway

最後に、API Gateway を作成します。

Lambda のデザイナー画面より、「＋トリガーを追加」から以下を設定し、「追加」をクリックします。

認証等が必要なシステムであれば設定は適宜調整してください。

• API Gateway
• Create an API
• HTTP API
• セキュリティ ... オープン

API Gateway が作成されたら、特に設定する事はありません。Lambda のデザイナー画面等からエンドポイントを確認できるので、これをメモしておきます。

CORS について

フロントのアプリケーションがホストされる環境と API Gateway に設定されるドメインが異なる場合には、CORS の設定を行う必要があります。

CORS については弊社ブログを含め世に多くの情報がありますので、それらをご覧ください。

API Gateway の Lambda プロキシ統合のCORS対応をまとめてみる

API Gateway + LambdaでCORSを有効にする

フロント実装

続いて、フロント側の実装をしていきます。

npm と node のバージョンは以下です。

$ npm --version 
6.14.5
$ node --version
v13.7.0

ローカル PC 上に作業用ディレクトリを用意し、以下コマンドを実行していきます。

React が未インストールの方はこちらなどを参照にインストールを済ませてください。

今回は videocount という名前でアプリケーションを作成します。

$ npx create-react-app videocount

生成されたディレクトリに移動し、必要なパッケージをインストールします。

$ cd videocount
$ npm install --save axios
$ npm install --save react-player

各種ソースを格納するディレクトリを作成します。

$ pwd
~/PATH/videocount
$ mkdir -p ./src/component/js
$ mkdir -p ./src/component/css

ファイルは以下 3ファイルを作成・更新します。

src/index.js

生成時に記載されている App に関する記述は削除し、代わりに RespoonsivePlayer 部分を追加しています。

src には再生可能な HLS のインデックスファイルを指定します。

import React from 'react';
import ReactDOM from 'react-dom';
import './index.css';
+ import ResponsivePlayer from './component/js/ResponsivePlayer'
import * as serviceWorker from './serviceWorker';

ReactDOM.render(
  <React.StrictMode>
+    <ResponsivePlayer src = 'https://d2zihajmogu5jn.cloudfront.net/bipbop-advanced/bipbop_16x9_variant.m3u8' />
  </React.StrictMode>,
  document.getElementById('root')
);

// If you want your app to work offline and load faster, you can change
// unregister() to register() below. Note this comes with some pitfalls.
// Learn more about service workers: https://bit.ly/CRA-PWA
serviceWorker.unregister();

src/component/js/ResponsivePlayer.js

以下ソースコードを作成しました。

const server 部分には作成した API Gateway のエンドポイントを指定します。

API のコールには axios を使用しています。

動画の再生を onStart でハンドルし、API Gateway に通信を行う様にしています。

動画の再生周りは ReactPlayer を使用していますが、実装にあたってはこちら を参考にさせて頂きました。

import React, { Component } from 'react';
import ReactPlayer from 'react-player';
import '../css/ReactPlayer.css';
import axios from 'axios';

const server = 'https://xxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/default/xxxxxxxx'

class ResponsivePlayer extends Component {
    onStart = () => {
        console.log('onStart')

        let params = new URLSearchParams();
        params.append('target', this.props.src);
        axios.post(server, params)
        .then((res) => {
            console.log(res)
        })
        .catch((error) => {
            console.log(error)
        });
    }

    render() {
        return (
            <div className='player-wrapper'>
                <ReactPlayer
                    url={this.props.src}
                    className='react-player'
                    controls
                    width='100%'
                    height='100%'
                    onStart={this.onStart}
                />
            </div>
        )
    }
}

export default ResponsivePlayer;

src/component/css/ReactPlayer.css

.player-wrapper {
    position: relative;
    padding-top: 56.25% /* Player ratio: 100 / (1280 / 720) */
}

.react-player {
    position: absolute;
    top: 0;
    left: 0;
}

動かしてみた

動画を再生し、DynamoDB にデータが格納されるか確認してみます。

ローカルでフロント環境を実行します。

$ npm start

特に問題がなければ、ブラウザで localhost:3000 としてアプリケーションが起動するので、動画を再生してみます。

Chrome の開発者ツールで Console を確認すると、API Gateway へのリクエストが行われている事がわかります。（レスポンスメッセージが count success）

1分以内にブラウザをリロードし再度動画を再生したところ、API Gateway へのリクエストは行われていますがロジック通り値の更新はされていなさそうです。（レスポンスメッセージが Didn't Count）

DynamoDB を確認してみると、共に想定どおり値が更新されている事がわかります。（途中で IP も変えて試してみました）

ViewCount

RequestHistory

無事に、動画の再生回数を取得する仕組みを作成する事ができました！

おわりに

React + API Gateway + Lambda + DynamoDB で動画の再生回数を取得する仕組みを作成してみました。

全体的に詳しく無いカテゴリだったので苦戦しながら進めましたが、無事やりたい事が出来て良かったです。

この記事がどなたかの参考になれば幸いです。

以上、AWS 事業本部の大前でした。

参考

• API Gateway
• Lambda
• DynamoDB
• React
• API Gateway の Lambda プロキシ統合のCORS対応をまとめてみる
• API Gateway + LambdaでCORSを有効にする
• Reactのインストールとhello worldを表示する手順
• axios
• ReactPlayer
• ReactPlayer: ReactJS open sources media playerの使用法