Youtube Data APIを利用してSnowflakeにデータ収集してみた

この投稿は、2024年JINSのアドベントカレンダー14日目の記事です。

はじめに

9月にJINSに入社しました、ITデジタル部のじゅんぱぱ（@junpapa）です。
データエンジニアとしてSnowflakeのデータ基盤運用に関わっておりますが、まだまだSnowflakeのお作法を学んでいるところなので、ちょっとしたデータ収集のテストをしてみようと思います。

Youtube Data APIから動画の視聴データを収集し、Snowflakeに格納するPythonコードを書いてみました。

アプローチ

データ取得の流れを下図にまとめます。

APIから取得されるRAWデータはJSON形式です。
YoutubeAPIの提供項目はとても多いので、最初から必要なものを決めてテーブルを作ると、後々やっぱりあれも必要などと言われたりして項目の追加などで苦しめられそうな予感がします。
柔軟性を持たせるため、JSONをそのままSnowflakeのRAW層のテーブルに格納した上で、ビューでJSONをパースして必要な項目を参照するようにします。

事前準備

YoutubeAPIの種類と用途

YoutubeAPIは下記の3種類が提供されております。

APIの種類	用途
YouTube Data API	公開されているチャネルおよび動画について、視聴数や良いね数、コメント欄の投稿など、オープンなデータを取得する
YouTube Analytics API	コンテンツ管理権限を持つチャンネルおよび動画について、視聴数の時系列推移やトラフィックソース、性年代など、より詳細な情報を取得する
YouTube Reporting API	コンテンツ管理権限を持つチャネルおよび動画について、一括レポートを取得する

詳細は [公式ドキュメント](https://developers.google.com/youtube/reporting?hl=ja をご参照ください。

今回は「YouTube Data API」を利用します。
Youtubeに公開されている動画であれば、チャンネル権限がなくてもベーシックな情報が取得できるため、どなたでも試すことができます。

より詳細な情報を取得したいのであれば「YouTube Analytics API」または「YouTube Reporting API」の利用をご検討ください。利用に際しては、チャンネルの所有者に相談し、チャンネル権限を付与してもらう必要があります。

Youtube Data APIの初期設定

Youtube Data APIは無料で利用できます。（1日あたり10,000クォータの上限あり）
公式リファレンス に沿ってGoogle API Console上でAPIの初期設定を行ってください。

• Youtube Data API v3 の有効化
• Youtube Data APIキーの取得

情報を参照したいYoutubeチャンネルIDを取得

YoutubeチャンネルのURLにチャンネルIDが含まれていますので取得します。
JINS公式Youtube のチャンネルIDは「UCKs1fDW7UrrlGIWzQIoFsOw」です。

チャンネル登録もぜひ！

Pythonライブラリのインストール

コードテストは以下の環境で行っています。

• Python 3.11.6
• Youtube Data API v3

snowflake-connector-python　pandas-tools

pandasでごにょごにょ加工したやつをテーブルに突っ込めると便利なのでこれを使います。
公式リファレンス

pip3 install "snowflake-connector-python[pandas]"

不足しているライブラリがあれば適宜追加してください。

Snowflakeテーブル・ビューの作成

Snowflakeに2つのテーブル・ビューを作成します。

TABLE

APIから取得したJSONをそのまま格納するテーブル

CREATE OR REPLACE TABLE RAW.YOUTUBE_VIDEOS (
	date_part DATE NOT NULL COMMENT '日付',
	json_data VARIANT COMMENT 'JSON'
)
;

VIEW

JSONをパースして、動画別の指標を参照するビュー

CREATE OR REPLACE VIEW STAGING.YOUTUBE_VIDEOS_VIEW
(
    date_part COMMENT '日付',
    video_id COMMENT '動画ID',
    title COMMENT 'タイトル',
    description COMMENT '説明',
    category_id COMMENT 'カテゴリID',
    channel_id COMMENT 'チャンネルID',
    channel_title COMMENT 'チャンネルタイトル',
    published_at COMMENT '公開日時',
    view_count COMMENT '視聴回数',
    like_count COMMENT 'いいね数',
    comment_count COMMENT 'コメント数'
)
AS

SELECT
date_part::DATE AS date_part,
items.value:id::VARCHAR AS video_id,
-- snippet
items.value:snippet.title::VARCHAR AS title,
items.value:snippet.description::VARCHAR AS description,
items.value:snippet.categoryId::NUMBER AS category_id,
items.value:snippet.channelId::VARCHAR AS channel_id,
items.value:snippet.channelTitle::VARCHAR AS channel_title,
items.value:snippet.publishedAt::TIMESTAMP AS published_at,
-- statistics
items.value:statistics.viewCount::NUMBER AS view_count,
items.value:statistics.likeCount::NUMBER AS like_count,
items.value:statistics.commentCount::NUMBER AS comment_count
FROM RAW.YOUTUBE_VIDEOS
,LATERAL FLATTEN(INPUT => json_data:items) AS items
;

取得可能な動画情報の詳細は公式リファレンスをご参照ください。

サンプルコード

実行環境に応じて【】の箇所を置換してテストしてください。

import requests
import datetime
import pandas as pd
import snowflake.connector
from snowflake.connector.pandas_tools import write_pandas

# --------------------------------------------------------
# こちらのブロックを環境に応じて書き換えてください
# --------------------------------------------------------
# APIキー
YOUTUBE_DATA_API_KEY = '【YOUTUBE_DATA_API_KEY】'

# 検索したいチャンネルのchannelId
CHANNEL_ID = '【CHANNEL_ID】'

# Snowflake接続情報
SNOWFLAKE_HOST     = '【HOST】'
SNOWFLAKE_ACCOUNT  = '【ACCOUNT】'
SNOWFLAKE_USER     = '【USER】'
SNOWFLAKE_PASSWORD = '【PASSWORD】'
SNOWFLAKE_DATABASE = '【DATABASE】'
SNOWFLAKE_SCHEMA   = '【SCHEMA】'
# --------------------------------------------------------

# SearchリソースのリクエストURL指定
REQUEST_URL_SEARCH = 'https://www.googleapis.com/youtube/v3/search'

# VideosリソースのリクエストURL指定
REQUEST_URL_VIDEOS = 'https://www.googleapis.com/youtube/v3/videos'

page_token = None
video_id_list = []

while True:
    # パラメータ設定
    params = {
        'key': YOUTUBE_DATA_API_KEY, # APIキー
        'part': 'id',                # videoIdを含むプロパティ
        'channelId': CHANNEL_ID,     # チャンネルID
        'order': 'date',             # 結果を日付順に表示
        'type': 'video',             # 動画のみを検索
        'maxResults': 50,            # 1リクエストあたりの結果（5-50）
        'pageToken': page_token      # ページトークン
    }
    
    # リクエスト実行
    search_result = requests.get(REQUEST_URL_SEARCH, params=params)
    
    # json形式の結果を格納
    search_result_json = search_result.json()
    
    # videoIdの抽出
    for item in search_result_json['items']:
        video_id_list.append(item['id']['videoId'])
    
    # 検索結果を全て表示できていない場合はpageTokenを更新
    if 'nextPageToken' in search_result_json:
        page_token = search_result_json['nextPageToken']
    else:
        break

# video_id_listを50件ずつ分割
video_id_list_splitted = []
for i in range(0, len(video_id_list), 50):
    video_id_list_splitted.append(video_id_list[i:i+50])

# 分割グループごとにリクエスト実行
json_data = []
for video_id_grp in video_id_list_splitted:
    # パラメータ設定
    params = {
        'key': YOUTUBE_DATA_API_KEY,
        'part': 'snippet,statistics',
        'id': ','.join(video_id_grp)
    }
    
    # リクエスト実行
    videos_result = requests.get(REQUEST_URL_VIDEOS, params=params)
    # リストに追加
    json_data.append(videos_result.json())

# dataframeにJSONデータを追加
df_json = pd.DataFrame({'json_data':json_data})

# dataframeにデータ取得日列を追加
dt_now = datetime.datetime.now()
date_part = dt_now.strftime('%Y-%m-%d')
df_json['date_part'] = date_part

# Snowflakeへ接続
session = snowflake.connector.connect(
    host = SNOWFLAKE_HOST,
    account = SNOWFLAKE_ACCOUNT,
    user = SNOWFLAKE_USER,
    password = SNOWFLAKE_PASSWORD,
    database = SNOWFLAKE_DATABASE,
    schema = SNOWFLAKE_SCHEMA
)

# Snowflakeへ書き込み
# TABLEへのDataFrameの書き込み
# success 処理の成否
# nchunks チャンクの数
# nrows 挿入した行数
# output 内部で発行しているCOPY INTO <TABLE>の出力
# 列は順不同でも問題ないが、存在しない列が含まれているとエラーになる
success, nchunks, nrows, output = write_pandas(
    conn = session,
    df = df_json,
    table_name = 'YOUTUBE_VIDEOS',
    overwrite = True,          # 洗い替える
    quote_identifiers = False, # pandasの列名を "" でエンクローズしない
    auto_create_table = False  # テーブルを自動でCREATEしない
)

# 更新結果を表示
print(output)

こちらの保存ファイル名を「main.py」としています。

テスト

サンプルコードを実行してみます。

python3 main.py

正常にロードされました。

[('*******/file0.txt', 'LOADED', 2, 2, 1, 0, None, None, None, None)]

Snowflakeに保存されたJSONデータを確認してみます。
SQL Worksheetから下記のクエリを実行してください。

SET DATABASE 【データベース名】;
SELECT * FROM RAW.YOUTUBE_VIDEOS;

パースされたJSONデータを確認してみます。

SELECT * FROM STAGING.YOUTUBE_VIDEOS_VIEW;

記事執筆時点（2024/12/10）で68本の動画情報が取得できました。

まとめ

今回は、Youtube Data APIを例にデータ取得、Snowflakeへの格納、利用までの一連の手続きをまとめてみました。他のAPIをテストする際などに、ちょこっとでも参考になる部分があれば幸いです。