技術検証：AWS Lambdaを使ってYouTube APIから動画データをDynamoDBに挿入する実践手順

はじめに

ITスクールのハッカソンで他のチームが取り組んでいたAWSを使用したサーバーレスアーキテクチャに興味を持ち、自分でも実際に手を動かして技術検証を行いました。

今回の検証内容は、AWS Lambdaを使用してYouTube APIから動画データを取得し、そのデータをDynamoDBに挿入する仕組みの構築です。

今回はLambda関数を手動で実行し、技術検証を進めますが、次回の記事ではEventBridgeを活用して、定期的に実行されるスケジュールを設定し、さらに効率化を図ります！

知識整理

AWS Lambdaとは

AWS Lambdaは、サーバーを管理せずにコードを実行できるサービスです。

イベントに基づいて自動的にコードを実行し、スケーリングやインフラ管理をAWSが担当するため、開発者はビジネスロジックの実装に集中できます。

Lambdaに関する技術検証を行った記事もありますので、興味のある方はぜひ参考にしてください。

YouTube APIとは

YouTube APIは、YouTubeの動画、プレイリスト、チャンネルなどにアクセスできるAPIです。

このAPIを使うことで、動画の再生回数やいいね数、コメントなどの情報を取得でき、さまざまなアプリケーションで活用できます。

DynamoDBとは

DynamoDBは、AWSが提供するフルマネージド型のNoSQLデータベースサービスです。

高いスループットと低レイテンシを実現しており、大規模なデータを扱うアプリケーションに最適です。

DynamoDBについても詳しく解説している記事がありますので、興味のある方は参考にしてください！

実際にやってみた

YouTube APIの設定

まず、Google Cloud Consoleにアクセスし、新しいプロジェクトを作成します。

プロジェクト名を入力し、「作成」をクリックします。

次に、「APIとサービス」 > 「ライブラリ」を選択し、検索バーに「YouTube Data API v3」を入力して検索します。APIを選択した後、「有効にする」をクリックします。

その後、「APIとサービス」 > 「認証情報」を開き、「認証情報を作成」 > 「APIキー」を選択します。

生成されたAPIキーをメモしておきます。

DynamoDBテーブルの作成

ここでは、RDSではなく「DynamoDB」を使用して技術検証を進めます。

サービス検索バーに「DynamoDB」と入力し、DynamoDBの画面で左側のメニューから「テーブル」をクリックします。

テーブル作成画面が表示されますので、以下のように入力します。テーブル名は「YouTubeVideos」を使用します。

パーティションキーを入力し、データ型として「文字列」を選択します。テーブル設定は、デフォルトのまま進めていきます。

数秒でテーブルが作成され、作成完了後、DynamoDBの「テーブル」セクションに表示されます。以下は、テーブルに挿入されたデータの一例です。

video_id	title	views	likes	dislikes	published_at
abc123	サンプル動画1	100000	1000	50	2024-01-01T00:00:00Z
xyz456	サンプル動画2	50000	500	10	2024-02-01T00:00:00Z

ローカル環境でgoogleapiclient モジュールの作成

あらかじめ、Lambda関数で使うモジュール含めて必要なPythonファイル含めて作成していきます。

今回は、ローカルで必要なモジュールを導入した上で、Lambda上にZipファイルをアップロードします。

任意のディレクトリを作成し、その中に lambda_function.py という空のファイルを作成します。

mkdir test
cd test
touch lambda_function.py

必要なライブラリをローカル環境にインストールします。ここではMacBookを使用していますが、他の環境でも同様にインストールできます。

pip install google-api-python-client -t ./lambda_package
pip install google-auth google-auth-httplib2 google-auth-oauthlib -t ./lambda_package
cp lambda_function.py ./lambda_package/

次に、Lambda関数のコード準備を行います。lambda_function.py を lambda_package フォルダにコピーし、lambda_package フォルダ全体をZIP形式で圧縮します。

cd lambda_package
zip -r ../lambda_function.zip .
cd ..

最後に、ls -l コマンドで lambda_function.zip ファイルが作成されたことを確認します。正常に完了していれば、このファイルが表示されます。

Lambda関数の作成

AWS Lambdaサービスを開き、「関数の作成」を選択します。関数名（例: YouTubeDataFetcher）を入力し、ランタイムとして「Python 3.13」を選択します。

Lambda関数には、DynamoDBへのアクセス権限が必要ですが、この後でIAMロールを設定します。ここでは一旦デフォルトの設定で進めます。

YouTubeDataFetcher 関数を選択し、コードタブから「アップロード元」をクリックします。先ほどローカルで作成したZIPファイルをアップロードします。

ローカルで作成したファイルのパス先まで移動して選択して、アップロードしてください。

アップロード後、lambda_function.py の空ファイルが表示されます。ここに以下のコードをコピー＆ペーストし、「デプロイ」をクリックして変更を保存します。

lambda_function.py

import json

def lambda_handler(event, context):
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }
import json
import os
import boto3
from googleapiclient.discovery import build

# 環境変数からAPIキーを取得
YOUTUBE_API_KEY = os.environ['YOUTUBE_API_KEY']
YOUTUBE_API_SERVICE_NAME = "youtube"
YOUTUBE_API_VERSION = "v3"

# DynamoDBの設定
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('YouTubeVideos')

def get_youtube_data(video_id):
    """
    YouTube APIから動画データを取得する関数
    """
    youtube = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, developerKey=YOUTUBE_API_KEY)

    # YouTube動画情報の取得
    request = youtube.videos().list(
        part="snippet,statistics",
        id=video_id
    )
    response = request.execute()
    
    return response

def insert_data_into_db(video_data):
    """
    取得したYouTube動画データをDynamoDBに挿入する関数
    """
    table.put_item(Item={
        'video_id': video_data['id'],
        'title': video_data['snippet']['title'],
        'views': video_data['statistics']['viewCount'],
        'likes': video_data['statistics'].get('likeCount', '0'),
        'dislikes': video_data['statistics'].get('dislikeCount', '0'),
        'published_at': video_data['snippet']['publishedAt']
    })

def lambda_handler(event, context):
    """
    Lambdaのエントリーポイント
    イベントから動画IDを取得してYouTube APIからデータを取得し、DynamoDBに保存
    """
    video_id = event.get('video_id')
    if not video_id:
        return {
            'statusCode': 400,
            'body': json.dumps('video_id is required')
        }

    try:
        # YouTube APIからデータを取得
        video_data = get_youtube_data(video_id)
        
        if 'items' in video_data and len(video_data['items']) > 0:
            video_info = video_data['items'][0]
            insert_data_into_db(video_info)

            return {
                'statusCode': 200,
                'body': json.dumps({
                    'message': 'Data successfully inserted into DynamoDB',
                    'video_id': video_id,
                    'title': video_info['snippet']['title'],
                    'views': video_info['statistics']['viewCount']
                })
            }
        else:
            return {
                'statusCode': 404,
                'body': json.dumps('Video not found')
            }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps(f'Error fetching data: {str(e)}')
        }

追加対応として、作成したLambda関数の「設定」タブから環境変数を設定します。事前に取得しておいたAPIキーを環境変数として設定しましょう。

環境変数を設定した後、コード内では os.environ['YOUTUBE_API_KEY'] を使ってAPIキーを取得できます。

IAMロールの設定

Lambda関数には、DynamoDBのPutItem操作を実行する権限が必要です。以下のカスタムポリシーを作成し、IAMロールにアタッチしてください。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "dynamodb:PutItem",
      "Resource": "arn:aws:dynamodb:ap-northeast-1:340823193247:table/YouTubeVideos"
    }
  ]
}

この設定により、Lambda関数はDynamoDBのYouTubeVideosテーブルに対してPutItem操作を実行できるようになります。

Lambda関数を実行してみた

次に、テストイベント用のJSONファイルを作成し、video_idを含むデータを入力します。

JSONデータは、テストしたいYouTube動画のIDを含む任意のデータを追加してください。

{
  "video_id": "dQw4w9WgXcQ"  // テストしたいYouTube動画のID
}

ここでは、通っているITスクールのYouTube動画のデータをLambdaで取得し、そのままDynamoDBにデータを挿入する処理を行います。

Lambda関数の実行確認

Lambda関数の「Test」ボタンを使ってテスト実行を行い、以下のように200ステータスコードが出力されたことを確認できれば、正常に処理が完了しています。

test.log

Response:
{
  "statusCode": 200,
  "body": "{\"message\": \"Data successfully inserted into DynamoDB\", \"video_id\": \"sdBro6Vl15E\", \"title\": \"\\u672a\\u7d4c\\u9a13\\u304b\\u3089\\u30a8\\u30f3\\u30b8\\u30cb\\u30a2\\u3078\\uff01\\u8077\\u7a2e\\u30fb\\u5e74\\u53ce\\u30fb\\u30aa\\u30b9\\u30b9\\u30e1\\u5ea6\\u3092\\u5fb9\\u5e95\\u89e3\\u8aac\\uff01\", \"views\": \"46\"}"
}

次に、DynamoDBで作成したYouTubeVideosテーブルの中身を確認すると、以下のように各項目のデータが正しく取得されていることが分かります。

中身の属性や値も、想定通り取得されていることが確認でき、検証は大成功です！！

Lambda関数を手動で実行し、YouTube APIから取得した動画データがDynamoDBに正常に挿入されたことが確認できました。

video_id	title	views	likes	dislikes	published_at
sdBro6Vl15E	未経験からエンジニアへ！職種・年収・オススメ度を徹底解説！	46	5	0	2024-12-04T09:00:55Z

まとめ

この記事では、AWS Lambdaを使用してYouTube APIから動画データを取得し、DynamoDBに保存する方法を解説しました。

YouTube APIの設定、Lambda関数の作成、データ挿入、IAMロールの設定まで、サーバーレスで動画データ管理システムを構築するためのステップを紹介しました。

次回の記事では、EventBridgeを使用して定期的にLambda関数を実行する方法を紹介する予定です。この記事が誰かの技術的な支えになれば幸いです！

おまけ：私はLambda関数が大好き

私はLambda関数が大好きで、自宅でもいろいろな検証を行っています。興味のある方は、ぜひ読んで技術検証に挑戦してみてください！