Youtube APIデータをBigQueryへロードする作業を自動化する

## 初めに
初学者である私が[こちらの記事](https://qiita.com/myotkh/items/dfab8229f45e277db95d)をかなり参考にして
データ基盤を構築した際の備忘録となっています。
初めてデータ基盤を構築したい方のためにできるだけ詳細に記載していますので、
周りくどい内容であることをご承知ください。

##前提条件
・[Google Cloud Platform](https://console.cloud.google.com/home/dashboard?project=thematic-fort-321903&hl=ja)のアカウントを取得している
・YouTubeAPIを取得している<br>

YouTube APIの取得に関しては以下の記事でわかりやすく説明されています！
[【Python】YouTube Data API を使って、いろんな情報を取得してみた！](https://qiita.com/ryoya41/items/dd1fd4c1427ece787eea)

## 0.設計したアーキテクチャ

![GCP アーキテクチャ.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/68a4b53f-5da1-8f54-cd1e-45c36458f433.png)

1. **Cloud Scheduler**でメッセージを発行する
2. **Pub/SubがCloud Functions**にメッセージを通知
3. **Cloud Functions**でデータを取得し、**Cloud Storage**に保存
4. **Cloud Storage**にファイルがアップロードされたのをトリガーに、<br>**Cloud Functions**が実行され**BigQuery**にデータをロード
5. **BigQuery**と連携しデータポータルでレポート作成

以下でそれぞれのコンポーネントについて詳しく説明していきます。
※GCPを使用するため、プロジェクトを作成しておく必要があります。

## 1.Pub/Subのトピックおよびサブスクリプションを作成
順番が前後しますが、Cloud Schedulerの設定のために
[Pub/Sub](https://cloud.google.com/pubsub?hl=ja)のトピックとサブスクリプションを作成する必要があります。


[公式の手順](https://cloud.google.com/pubsub/docs/quickstart-cli?hl=ja)に沿って作成していきます。
####1.1.GCP上でShellを起動
下図の赤丸の部分をクリックするとShellが起動されます。
![スクリーンショット 2021-08-10 13.33.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/6512a34d-529b-3747-419b-56a88e3c1364.png)

####1.2.トピックの作成
my-topic を含むトピックを作成されます。

```
gcloud pubsub topics create <my-topic>
```

####1.3.サブスクリプションの作成
my-sub を含むサブスクリプションを作成し、先ほど作成した my-topic に添付します。

```
gcloud pubsub subscriptions create <my-sub> --topic=<my-topic>
```
これでトピックおよびサブスクリプションの準備が完了です！

## 2.Cloud Schedulerでメッセージを発行

####2.1.ジョブを作成
<img width='450' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/1f789025-fe7b-f4b5-8349-ed41ca6f337d.png'>

※頻度に関しては[こちら](https://cloud.google.com/scheduler/docs/configuring/cron-job-schedules?hl=ja&_ga=2.266691406.-1904697997.1623659898#defining_the_job_schedule)を参照してください。

<img width='450' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/017a6c0d-632a-c517-6e74-eb795a8d7796.png'>

Cloud Pub/Sub トピックは「1.2.トピックの作成」で作成したものを選択してください。

## 3.Cloud FunctionsでYouTubeデータを取得する関数を作成
####3.1.1. 事前に行っておくこと
1.　GCS（Google Cloud Storage）に**バケットを作成**
2.　作成したバケットの中に**「'subscriberCount', 'viewCount', 'date'」**
のカラム名が入った空データの**csvファイルを保存**（このcsvファイルにデータが保存されていく）

###3.1.「チャンネルの登録者数、総再生回数」を取得する関数
関数を作成から以下の通りに記載します（関数名は好きなもので結構です）。
<img width='450' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/dbee9815-486c-0153-3ac1-58d26281033e.png'>
Cloud Pub/Sub トピックは「1.2.トピックの作成」で作成したものを選択してください。

ランチタイム：python3.9（Pythonであればなんでも大丈夫です）
ソースコードは：インラインエディタ
エントリポイント：youtube（main.pyにある関数を定義します）
ちなみに関数youtubeの引数にはPub/Subから通知されたメッセージが入ります。

Cloud Functions内に以下の5つのファイルを作成します
・main.py
・requirements.txt
・get_channel.py
・load_to_gcs.py
・settings.py

```python:main.py
from apiclient.discovery import build
from get_channel import GetChannel
import settings

def youtube(event, context):
    # 設定
    API_KEY = settings.API_KEY
    YOUTUBE_API_SERVICE_NAME = 'youtube'
    YOUTUBE_API_VERSION = 'v3'    
    channel_name = '<ここに取得したいチャンネル名を記載>'

    youtube = build(
        YOUTUBE_API_SERVICE_NAME,
        YOUTUBE_API_VERSION,
        developerKey=API_KEY,
        cache_discovery=False
    )
    channel = GetChannel(youtube, channel_name)
    channel.get_channel_details()
```

```requirements.txt
google-api-core==1.23.0
google-api-python-client==1.12.8
google-cloud-storage==1.33.0
google-auth==1.23.0
pandas==1.1.5
python-dotenv==0.15.0
```

```python:get_channel.py
from datetime import datetime
from apiclient.discovery import build
import pandas as pd
import load_to_gcs

class GetChannel(object):
    def __init__(self, youtube, channel_name):
        self.youtube = youtube    
        self.channel_name = channel_name

    def get_channel_basic(self):

        response = self.youtube.search()\
                       .list(q=self.channel_name, part='id,snippet', maxResults=1).execute()

        for item in response.get('items', []):
            if item['id']['kind'] == 'youtube#channel':
                return item['id']['channelId']

    def get_channel_details(self):   
        response = self.youtube.channels().list(
            part = 'snippet,statistics',
            id = self.get_channel_basic()
            ).execute()

        for item in response.get("items", []):
            if item["kind"] == "youtube#channel":
                series = pd.Series([item['statistics']['subscriberCount'], 
                                   item['statistics']['viewCount'],
                                   datetime.now().strftime('%Y-%m-%d')],
                                   ['subscriberCount', 'viewCount', 'date'])
                load_to_gcs.read_from_gcs_and_append(series)

```

```python:load_to_gcs.py
import datetime
from io import BytesIO
from google.cloud import storage as gcs
import pandas as pd
import settings

def read_from_gcs_and_append(series):
    project_id = settings.project_id
    bucket_name = settings.bucket_name
    gcs_path = '<GCSに保存したファイル名>.csv' # csv形式

    client = gcs.Client(project_id)
    bucket = client.get_bucket(bucket_name)

    blob_read = bucket.blob(gcs_path)
    content_read = blob_read.download_as_string()

    df = pd.read_csv(BytesIO(content_read))
    df = df.append(series, ignore_index=True)

    blob_write = bucket.blob(gcs_path)
    blob_write.upload_from_string(df.to_csv(index=False, header=True), content_type='text/csv')
```

```python:settings.py
API_KEY = '<自分のYouTube API　を入力>'
project_id = '<自分のproject_idを入力>'
bucket_name = '<load_to_gcs.pyでcsvファイルを保存したbucket_nameを入力>'
```
###3.２.「アップされた動画の詳細情報」を取得する関数
Cloud Pub/Sub トピックは「1.2.トピックの作成」で作成したものを選択してください。
ランチタイム：python3.9（Pythonであればなんでも大丈夫です）
ソースコードは：インラインエディタ
エントリポイント：youtube（main.pyにある関数を定義します）
ちなみに関数youtubeの引数にはPub/Subから通知されたメッセージが入ります。

今回は以下の6つのファイルを作成します
・main.py
・requirements.txt
・get_channel.py
・load_to_gcs.py
・settings.py
・get_video.py

※ settings.py、requirements.pyは上記と同じコードなので記載してません。

```python:main.py
from apiclient.discovery import build
from get_channel import GetChannel
from get_video import GetVideo
import settings

def youtube(event, context):
    # 設定
    API_KEY = settings.API_KEY
    YOUTUBE_API_SERVICE_NAME = 'youtube'
    YOUTUBE_API_VERSION = 'v3'    
    channel_name = '<ここに取得したいチャンネル名を記載>'

    youtube = build(
        YOUTUBE_API_SERVICE_NAME,
        YOUTUBE_API_VERSION,
        developerKey=API_KEY,
        cache_discovery=False
    )

    videos = GetVideo(youtube, channel_name)
    videos.get_video_details_csv()
```

```python:get_channnel.py
from apiclient.discovery import build

class GetChannel(object):
    def __init__(self, youtube, channel_name):
        self.youtube = youtube    
        self.channel_name = channel_name

    def get_channel_basic(self):
        response = self.youtube.search().list(q=self.channel_name, 
                                              part='id,snippet',
                                              maxResults=1).execute()

        for item in response.get('items', []):
            if item['id']['kind'] == 'youtube#channel':
                return item['id']['channelId']
```

```python:load_to_gcs.py
import datetime
from google.cloud import storage as gcs
import pandas as pd
import settings

def load_to_gcs_csv(df):
    now = datetime.datetime.now()
    project_id = settings.project_id
    bucket_name = settings.bucket_name
    gcs_path = '{}-test.csv'.format(now.strftime('%Y-%m-%d-%H:%M'))

    client = gcs.Client(project_id)
    bucket = client.get_bucket(bucket_name)

    blob = bucket.blob(gcs_path)
    blob.upload_from_string(df.to_csv(index=False, header=True), content_type='text/csv')
```

```python:get_video.py
from apiclient.discovery import build
import pandas as pd
from google.cloud import storage as gcs
from get_channel import GetChannel
from load_to_gcs import load_to_gcs_csv

class GetVideo(GetChannel):
    def get_video_basic(self):
        nextPagetoken = None
        video_ids = []
        while True:
            if nextPagetoken != None:
                nextpagetoken = nextPagetoken
            response = self.youtube.search().list(
                             part = "snippet",
                             channelId = self.get_channel_basic(),
                             maxResults = 50,
                             order = "date", #日付順にソート
                             pageToken = nextpagetoken
                       ).execute()

            for item in response.get("items", []):
                if item["id"]["kind"] == "youtube#video":
                    video_ids.append(item["id"]["videoId"])
            try:
                nextPagetoken = response["nextPageToken"]
            except:
                break
        return video_ids

    def get_video_details_csv(self):
        videos = []
        for video_id in self.get_video_basic():
            response = self.youtube.videos().list(
                part = 'snippet,statistics',
                id = video_id
            ).execute()
            for item in response.get("items", []):
                if item["kind"] == "youtube#video":
                    videos.append(
                    [item["snippet"]["title"],
                     item["statistics"]["viewCount"],
                     item["statistics"]["likeCount"],
                     item["statistics"]["dislikeCount"],
                     item["statistics"]["favoriteCount"] if "favoriteCount" in item["statistics"].keys() else None,
                     item["statistics"]["commentCount"] if "commentCount" in item["statistics"].keys() else None,
                     item["snippet"]["tags"] if "tags" in item["snippet"].keys() else None,
                     item["snippet"]["publishedAt"]
                     ])
        df_video = pd.DataFrame(videos, 
                                columns=['title', 'viewCount', 'likeCount', 'dislikeCount', 'favoriteCount', 'commentCount', 'tags', 'publishAt'])
        load_to_gcs_csv(df_video)
```

###3.3.「GCSからBigQueryへロード」する関数
####事前に行っておくこと
BigQueryにdetasetおよびtable（2つ）を作成します。

Cloud Shell上から以下のコマンドでdatasetとtableを作成できます

```
PROJECT=<自分のproject ID>
DATASET=<作成するDetaset名>
bq --project_id ${PROJECT} mk ${DATASET}
bq --project_id ${PROJECT} mk --time_partitioning_type=DAY ${DATASET}.<作成するtable名1>
bq --project_id ${PROJECT} mk --time_partitioning_type=DAY ${DATASET}.<作成するtable名2>
```

作成できれば、以下のようにtableのスキーマを設定します
######・<上記で作成したtable名1>のスキーマ
<img width='350' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/120b6c9d-d14e-b227-ce1a-552935dc52d5.png'>
######・<上記で作成したtable名2>のスキーマ
<img width='350' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/f7a1e9ba-ea76-ad8d-51f5-1e5d3ec52d33.png'>

####関数の定義
Storageへの保存をトリガーにするため、
以下の通りに設定する。
<img width='350' src='https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1371880/06d4d1b2-551a-1d6f-df72-e34eadd54313.png'>

ランチタイム：python3.9（Pythonであればなんでも大丈夫です）
ソースコードは：インラインエディタ
エントリポイント：youtube（main.pyにある関数を定義します）
ちなみに関数youtubeの引数にはPub/Subから通知されたメッセージが入ります。

今回は以下の3つのファイルを作成します
・main.py
・requirements.txt
・settings.py

```python:main.py
from google.cloud import bigquery
import settings

def load_data(data, context):
    # check content-type
    if data['contentType'] != 'text/csv':
        print('Not supported file type: {}'.format(data['contentType']))
        return
    # get file info
    bucket_name = data['bucket']
    file_name = data['name']
    uri = 'gs://{}/{}'.format(bucket_name, file_name)

    dataset_id = settings.dataset_id
    if file_name == '<GCSに保存してあるcsvファイル名>':
        table_id = settings.<上記で作成したtable名1>
    else:
        table_id = settings.<上記で作成したtable名2>
    client = bigquery.Client()
    dataset_ref =client.dataset(dataset_id)

    # Set Load Config
    job_config = bigquery.LoadJobConfig()
    job_config.autodetect = True
    job_config.source_format = bigquery.SourceFormat.CSV
    job_config.write_disposition = 'WRITE_TRUNCATE'

    # Load data
    load_job = client.load_table_from_uri(
        uri, dataset_ref.table(table_id), job_config=job_config
    )
    print("Starting job {}".format(load_job.job_id))
    load_job.result()
    print("Job finished.")
```

```requirements.txt
google-cloud-bigquery==1.24.0
python-dotenv==0.15.0
```

```python:settings.py
dataset_id = 'test_dataset'
table_view_subscriber_id = 'table_view_subscriber_id'
table_video_id = 'table_video_id'
```

##最後に
これでBigQueryへのロードまでを自動化することができたと思います！

データを取得するタイミングをずらしたい場合は、
Pub/Subのトピック・サブスクリプションを１つ追加して、
Cloud Schedulerの時間を指定することで簡単に変更することができます。

また、得られたデータは[Cloud Data Portal](https://marketingplatform.google.com/intl/ja/about/data-studio/)で簡単に可視化することができます。
興味のある方は是非チャレンジしてください！

初学者であるが故に至らぬ点が多くあると思いますので
ご指摘・修正点があればコメントください！
また、ご不明点があれば、お気軽にコメントしてください！