こんにちは、今日は、Pythonを使って、Google Search ConsoleのデータをBigQueryにロードする方法についてまとめました。
当ブログの運用でもGoogle Search Consoleを利用しており、本格的に運用を始めてから半年以上が経過したので、普段良く使うBigQueryとSQLを用いて集計をしたいと思い、BigQueryへの連携の自動化を目指し実装しました。
本プログラムは、Windows10 + Python 3.8環境で検証しています。
- Google Search Console APIを有効にする
- サービスアカウントの作成とGCP権限設定
- Googleサーチコンソールの権限設定
- BigQueryにロード先テーブルを作成する
- Pythonの必要なライブラリをインストールする
- Pythonで取得する
- まとめ
Google Search Console APIを有効にする
以下のURLからGCPのマーケットプレイスにアクセスし、Google Search Console APIを有効にします。
https://console.cloud.google.com/iam-admin/serviceaccounts
サービスアカウントの作成とGCP権限設定
GCPサービスアカウントを作成して、秘密鍵を発行します。 更に以下のロールを付与しておきます。
- BigQuery ジョブユーザー
- BigQuery データ編集者
サービスアカウントについては以下の記事でも詳細を解説しています。
Googleサーチコンソールの権限設定
Googleサーチコンソールの画面にアクセスし、
設定 > ユーザーと権限 > ユーザーを追加
から先程作成したサービスアカウントを追加します。権限は「フル」にしましょう。
BigQueryにロード先テーブルを作成する
BigQueryのテーブルを作成します。
今回の例では、search_console.データセットに、analytics_dataというテーブルを作成しています。
以下のスキーマで作成しましょう。
また、dateを日別のパーティションフィールドとして指定しておきましょう。
[ {"name":"date","type":"DATE","mode":"NULLABLE"}, {"name":"query","type":"STRING","mode":"REPEATED"}, {"name":"page","type":"STRING","mode":"NULLABLE"}, {"name":"device","type":"STRING","mode":"NULLABLE"}, {"name":"country","type":"STRING","mode":"NULLABLE"}, {"name":"clicks","type":"INTEGER","mode":"NULLABLE"}, {"name":"impressions","type":"INTEGER","mode":"NULLABLE"}, {"name":"ctr","type":"FLOAT","mode":"NULLABLE"}, {"name":"position","type":"FLOAT","mode":"NULLABLE"} ]
Pythonの必要なライブラリをインストールする
ローカルのPython環境に必要なライブラリをインストールします。
$ pip install google-api-python-client==2.41.0 $ pip install oauth2client==4.1.3 $ pip install google-auth==2.6.2 $ pip install google-cloud-bigquery==2.34.2
Pythonで取得する
以下のプログラムで、Google Search Console APIとGoogle BigQuery APIを実行して、データ取得からBigQueryへのロードまでを行います。
from datetime import datetime from urllib.parse import unquote from google.cloud import bigquery from google.oauth2 import service_account from googleapiclient.discovery import build from oauth2client.service_account import ServiceAccountCredentials CREDENTIALS = '<サービスアカウントの認証ファイルのパス>' DOMAIN = '<サーチコンソールに設定しているドメイン>' PROJECT_ID = '<GCPのプロジェクトID>' BQ_TABLE = '<ロードするBigQueryのデータセット.テーブル>' DIMENSIONS = ['date', 'query', 'page', 'device', 'country'] ROW_LIMIT = 25000 class SearchConsoleToBigQueryClient: def __init__(self): credentials = ServiceAccountCredentials.from_json_keyfile_name( filename=CREDENTIALS, scopes=[ 'https://www.googleapis.com/auth/webmasters.readonly' ], ) self.__webmasters = build('webmasters', 'v3', credentials=credentials) credentials2 = service_account.Credentials.from_service_account_file( CREDENTIALS, scopes=["https://www.googleapis.com/auth/cloud-platform"], ) self.__bq_client = bigquery.Client(project=PROJECT_ID, credentials=credentials2) def __format_json( self, api_response: list): list_ = [] for row in api_response: list_.append( { 'date': row['keys'][0], 'query': row['keys'][1].split(' '), 'page': unquote(row['keys'][2]), 'device': row['keys'][3], 'country': row['keys'][4], 'clicks': row['clicks'], 'impressions': row['impressions'], 'ctr': float(row['ctr']), 'position': float(row['position']), } ) return list_ def get_analytics_data( self, start_date: datetime, end_date: datetime=None): if end_date is None: end_date = start_date body = { 'startDate': start_date.strftime(r'%Y-%m-%d'), 'endDate': end_date.strftime(r'%Y-%m-%d'), 'dimensions': DIMENSIONS, 'rowLimit': ROW_LIMIT, 'startRow': 0, } response = self.__webmasters.searchanalytics().query(siteUrl=DOMAIN, body=body).execute() try: rows = response['rows'] except KeyError: raise Exception('結果が取得されませんでした。') return self.__format_json(rows) def bq_load( self, data: list, partition_date: datetime=None, write_disposition: str='WRITE_TRUNCATE'): if partition_date is None: destination='{}'.format(BQ_TABLE) else: destination = '{}${}'.format(BQ_TABLE, partition_date.strftime(r'%Y%m%d')) bq_config = bigquery.LoadJobConfig() bq_config.write_disposition = write_disposition job = self.__bq_client.load_table_from_json( json_rows=data, destination=destination, num_retries=1, project=PROJECT_ID, job_config=bq_config ) job.result() def load_analytics_data( self, target_date: datetime): data = self.get_analytics_data(target_date) self.bq_load(data, target_date) if __name__ == '__main__': client = SearchConsoleToBigQueryClient() # 取得したい日付を設定 target_date = datetime.strptime('2021-09-01', r'%Y-%m-%d') # データ取得～BigQueryへのロードまでを実行 client.load_analytics_data(target_date)
以下の定数はそれぞれ自身の環境に合わせて書き換えてください。
CREDENTIALS: 作成したサービスアカウントの秘密鍵のファイルパスを設定しますDOMAIN: 'サーチコンソールに設定しているドメインを設定します。当ブログの場合は、'sc-domain:true-fly.com'PROJECT_ID: GCPのプロジェクトIDBQ_TABLE: ロードするBigQueryのデータセット.テーブル。今回の例の場合は'search_console.analytics_data'
DIMENSIONSはGoogle Search Console APIで取得する情報です。今回のBigQueryのスキーマに合わせて設定しています。
ROW_LIMITはGoogle Search Console API1回のリクエストで取得できる最大件数を設定しています。
25000件以上を取得したい場合は、APIリクエストにstartRowを指定する必要がありますが、今回のサンプルコードでは省略しています。
ターミナルでpythonを実行すると、指定した日付の情報をサーチコンソールから取得し、BigQueryの日付パーティションにロードされます。
ロードした結果は以下のようにBigQueryに格納されます。（positionのみ隠しています）
まとめ
以上、Pythonを使って、Google Search ConsoleのデータをBigQueryにロードする方法についてご紹介しました。
まずはPythonのプログラムを実行して自動で取得からロードまでを実装しました。今後は完全な自動化を求めて、GCPのプロダクトを用いてのスケジュール実行だったり、ためたデータを活用するSQLやBIもご紹介できればと思います！
