スプレッドシートからBigQueryに差分をアップデートする4つの方法

やりたかったことはスプレッドシート情報をBigQueryに上げて差分データの更新を行うということです。

# 困っていたこと

元々下記のフローで処理を行っていました。

![図7.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/3badb31a-30cd-79bd-5243-590365f9d673.png)

しかしこれだと、新規テーブル作成後にすぐにデータを挿入(insert)してもデータが更新されないことが多々ありました。
SQLの実行を行うときは`job id` のステータスで実行完了しているかは見えるものの、
時差がかなり大きかったり、sleepを入れるとinsertできることもありましたが、
できないこともあり安定した方法を探していました。

# 結論

上記のフローを下記フローに変えることで解決しました。

![図11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/650be490-f1be-8b08-0990-0e18b02c7dd5.png)


テーブルの作成とデータのアップロードを別々で行っていたところを同時に行う処理に
変更したところ問題なく動きました。
聞けば当たり前ですが、この書き方をしらなかったので、いかにテーブル作成後に
「どうやって実行完了の把握をするか？」、「どうやって待機処理をするか？」という方法にばかり目がいっていました。

# この記事で書いている内容

今回はスプレッドシートにで更新された情報の差分データのみを
反映させるという内容を書いています。

スプレッドシートのデータはこんな下記で用意しています。

![図8.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/3da26bf5-d4c1-795e-c713-2b6b26c3f83e.png)

1行目に日本語名、2行目にBigQuery用のカラム目、3行目にデータ型、
4行目以降に実際のデータが入っているというデータ構造です。

本当はPKの設定とかも入れたかったのですが、
今回はそこは趣旨としていれていないので、A列のIDが一致すればアップデート
一致しなければ更新するという流れにしました。

BigQueryのテーブルは下記のようにアップデートされます。

![図6.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/2c3ae818-441d-3176-3087-595b80f0f8e0.png)


# 4つの実行方法

実践したのは下記の4つになります。

① SQLの実行でテーブル作成 ⇒ アップデート
② google-cloud-bigqueryでテーブル作成 ⇒ pandasからアップデート
③ pandas-gbqでテーブル作成・アップデート
④ google-cloud-bigqueryでテーブル作成・アップデート

①、②は元々考えていたフローでテーブル作成後にアップロードできないパターンです。
③と④はテーブル作成と同時にデータをアプロードしてうまくいったパターンです。

同じ悩みを抱えている方もいるかと思い、一応①、②も記載しました。

下記からコードの紹介です。

# 共通部分のコード

```python:認証
from google.colab import auth
auth.authenticate_user()
```

上記実行すると、下記の認証コードがでてきます。
![図4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/4f2fc7fd-1fbe-5f40-b2f7-96088dca369a.png)

別の記事で認証コードなしで実行する方法も紹介していますが、
異なるPJのテーブルのデータとマージさせたりする場合は、
認証コードを毎回入れた方がよいと思います。

[参考：Google Colaboratoryで認証コードなしで実行する方法](https://qiita.com/plumfield56/items/3d9e234366bcaea794ac)

スプレッドシート操作、BigQuery操作は別記事で書いているので詳細は省きます。
[Pythonでのスプレッドシート操作方法(gspreadの使い方)](https://qiita.com/plumfield56/items/dab6230512f3381fdcad)
[PythonでBigQueryを操作する方法](https://qiita.com/plumfield56/items/664d1a09edecb28880ca)

```python:スプレッドシートの使用準備
!pip install gspread
from google.colab import auth
from oauth2client.client import GoogleCredentials
import gspread

# 認証処理
auth.authenticate_user()
gc = gspread.authorize(GoogleCredentials.get_application_default())

ss_id = `スプレッドシートID`
sht_name = 'シート名'

workbook = gc.open_by_key(ss_id)
worksheet = workbook.worksheet(sht_name)
values = worksheet.get_all_values()

values
# [['ID', '名前', '年齢', '好きなもの'],
#  ['id', 'name', 'age', 'favorit'],
#  ['int64', 'string', 'int64', 'string'],
#  ['1', 'Bob', '25', 'apple'],
#  ['2', 'Tom', '32', 'orange'],
#  ['3', 'Jay', '28', 'grape']]
```

```python:BigQueryの使用準備
project_id = 'GCPのプロジェクトID'
dateset = 'データセット名'
table_name = 'テーブル名'

from google.cloud import bigquery
client = bigquery.Client(project=project_id)
```

ここまでは①~④で共有で使用する部分になります。

# ① SQLの実行でテーブル作成・アップデート

SQLの実行という表現が正しいのかわからないですが、
クエリエディタで操作するのと同じSQLで実行できる方法です。

メリットは`書きやすいこと`と`job id`でステータスがわかることかなと思っています。

```python:tempテーブル作成用のSQL作成
table_id = f'{project_id}.{dateset}.{table_name}'
query = f'create or replace table `{table_id}_temp`(\n'

for column, dtype in zip(values[1],values[2]):
  query += f'{column} {dtype},\n'
query = query[:-2] + ')'
query

# create or replace table `{project_id}.{dateset}.{table_name}_temp`(
# id int64,
# name string,
# age int64,
# favorit string)
```

```python:SQLの実行
# 実行
result = client.query(query)

# ジョブの状態確認
client.get_job(result.job_id, location='asia-northeast1').state

# 'DONE'
```

ここまでのテーブルの作り方が①～③で違いますが、
以下の差分をアップデートするためのSQLは共通で使用します。

```python:update用のSQL作成
query = f'''merge into `{project_id}.{dateset}.{table_name}` as T
using `{project_id}.{dateset}.{table_name}_temp` as S
on T.{values[1][0]} = S.{values[1][0]}
when matched then
update set
'''

insert_query = ''
for val in values[1][1:]:
  query += f'{val} = S.{val},\n'
  insert_query += f'{val}, '
insert_query = insert_query[:-2]

query = query[:-2] + '\n'

query += 'when not matched then\n'
query += f'insert({insert_query})\n'
query += f'values({insert_query})'
query

# merge into `{project_id}.{dateset}.{table_name}` as T
# using `{project_id}.{dateset}.{table_name}_temp` as S
# on T.id = S.id
# when matched then
# update set
# name = S.name,
# age = S.age,
# favorit = S.favorit
# when not matched then
# insert(id, name, age, favorit)
# values(id, name, age, favorit)
```

```python:update用のSQL実行
client.query(query)
```

すべて実行を終えて、BigQuery上でクエリ結果をみると
下記画像のように成功になっています。

![図9.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/7916c329-0d2e-bf72-22a8-80275cf292d6.png)

ただしテーブルを見てみると、下記のようにからな状態になっています。
jobのステータスは`DONE`なんだから反映されていてくれよ…。

![図2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/428239/18e70d81-01a8-cac3-87a1-5b14f4de249a.png)

サーバーの問題か5分ぐらいたつとデータが反映されたりするのですが、
すぐに連続して実行したいので使い物にならなかったです。


# ② google-cloud-bigqueryでテーブル作成・アップデート

ここに載っている内容でスプレッドシート情報をアップロードしたケースです。
https://googleapis.dev/python/bigquery/latest/usage/tables.html

```python:テーブル削除とスキーマの作成
table_id = f'{project_id}.{dateset}.{table_name}'
client = bigquery.Client(project = project_id)

# テーブルがあったら削除
client.delete_table(f'{table_id}_temp', not_found_ok=True)

schema = []
for column, dtype in zip(values[1],values[2]):
  schema.append(bigquery.SchemaField(column, dtype, mode="NULLABLE"))
schema

# [SchemaField('id', 'INT64', 'NULLABLE', None, ()),
#  SchemaField('name', 'STRING', 'NULLABLE', None, ()),
#  SchemaField('age', 'INT64', 'NULLABLE', None, ()),
#  SchemaField('favorit', 'STRING', 'NULLABLE', None, ())]
```

```python:tempテーブルの作成
table = bigquery.Table(f'{table_id}_temp', schema=schema)
client.create_table(table)
```

下記は①と同じコードなのでまとめています。

```python:アップデート用のSQLを作成して実行
df = pd.DataFrame(values[3:], columns=values[1])
# 空白削除、空白処理の処理
df = df.replace({'': None})
df = df.dropna(subset=[values[1][0]])

# tempのテーブルにアップデート
client.insert_rows(table, df.values.tolist())

# アップデート用SQLの作成
query = f'''merge into `{project_id}.{dateset}.{table_name}` as T
using `{project_id}.{dateset}.{table_name}_temp` as S
on T.{values[1][0]} = S.{values[1][0]}
when matched then
update set
'''

insert_query = ''
for val in values[1][1:]:
  query += f'{val} = S.{val},\n'
  insert_query += f'{val}, '
insert_query = insert_query[:-2]

query = query[:-2] + '\n'

query += 'when not matched then\n'
query += f'insert({insert_query})\n'
query += f'values({insert_query})'

# SQLを実行してデータのアップデート
client.query(query)
```

これでも①と同様に`person_temp` のテーブルが作成される確率が低かったです。
DPとの連携時はこのコードを良く使っていたのですが、
その際はsleepを20秒ぐらい入れて動くようにしていました。


# ③ pandas-gbqでテーブル作成・アップデート

他のブログで邪教だと書いてあったものもあったのであまり期待していなかったですが、
やってみてよかったです。



```python:DataFrameの作成
# 使用するためにインストールが必要
!pip install pandas-gbq

df = pd.DataFrame(values[3:], columns=values[1])
# 空白削除、空白処理の処理
df = df.dropna(subset=[values[1][0]])
df = df.replace({'': None})
```

このスキーマの作成は`name`と`type`の辞書型を作成します。


```python:schemaの作成
schema = []
for column, dtype in zip(values[1],values[2]):
  schema.append({'name': column, 'type': dtype})
schema

# [{'name': 'id', 'type': 'int64'},
#  {'name': 'name', 'type': 'string'},
#  {'name': 'age', 'type': 'int64'},
#  {'name': 'favorit', 'type': 'string'}]
```

```python:一時テーブルへのアップデート
df.to_gbq(f'{dateset}.{table_name}_temp', project_id, if_exists='replace', table_schema=schema)
```


```python:アップデート用のSQLを作成して実行
query = f'''merge into `{project_id}.{dateset}.{table_name}` as T
using `{project_id}.{dateset}.{table_name}_temp` as S
on T.{values[1][0]} = S.{values[1][0]}
when matched then
update set
'''

insert_query = ''
for val in values[1]:
  query += f'{val} = S.{val},\n'
  insert_query += f'{val}, '
insert_query = insert_query[:-2]

query = query[:-2] + '\n'

query += 'when not matched then\n'
query += f'insert({insert_query})\n'
query += f'values({insert_query})'

client.query(query)
```


こちらで問題なくアップデートができました。
ただし、下記のサイトを見ると`pandas-gbq`から`google-cloud-bigquery`へ移行してくださいと書かれています。
https://cloud.google.com/bigquery/docs/pandas-gbq-migration?hl=ja

なので、次に`google-cloud-bigquery`を使った方法を記載しました。


④ google-cloud-bigqueryでテーブル作成・アップデート

```python:pyarrowをインストール
# pyarrowをインストール
!pip install pyarrow

table_id = f'{project_id}.{dateset}.{table_name}'

df = pd.DataFrame(values[3:], columns=values[1])
# 空白削除、空白処理の処理
df = df.replace({'': None})
df = df.dropna(subset=[values[1][0]])
```



```python:schemaの作成及びDataFrameのデータ型変更
schema = []
for column, dtype in zip(values[1],values[2]):
  schema.append(bigquery.SchemaField(column,dtype))
  if dtype != 'string':
    df[column] = df[column].astype(dtype)
schema
```

google-cloud-bigqueryはpandas-gbqとは違って
テーブルとDataFrameのデータ型も一致していないとエラーになります。
公式の方が設定が細かいですね…。

df.dtpyesでデータ型が変わっているかの確認が可能です。

```python:dtypes
df.dtypes

# id          int64
# name       object
# age         int64
# favorit    object
# dtype: object
```

続いて、この後実行する`load_table_from_dataframe()`で行う設定を行います。
`config`は日本語で`設定`という意味です。

下記で設定可能な属性(Attributes)一覧が載っています。
https://googleapis.dev/python/bigquery/latest/generated/google.cloud.bigquery.job.LoadJobConfig.html

今回はそのうち`schema `と`write_disposition`で既存テーブルがあったときに上書きをする設定にします。

```python:jonの設定
job_config = bigquery.LoadJobConfig(
  schema = schema,
  write_disposition = 'WRITE_TRUNCATE'
)
```

```python:dfのデータをアップロード
job = client.load_table_from_dataframe(
  df, f'{table_id}_temp', job_config=job_config
)

# Wait for the load job to complete.
job.result()
```

下記は①と同じコードなのでまとめています。

```python:アップデート用のSQLを作成して実行
query = f'''merge into `{project_id}.{dateset}.{table_name}` as T
using `{project_id}.{dateset}.{table_name}_temp` as S
on T.{values[1][0]} = S.{values[1][0]}
when matched then
update set
'''

insert_query = ''
for val in values[1]:
  query += f'{val} = S.{val},\n'
  insert_query += f'{val}, '
insert_query = insert_query[:-2]

query = query[:-2] + '\n'

query += 'when not matched then\n'
query += f'insert({insert_query})\n'
query += f'values({insert_query})'

client.query(query)
```


③の方がデータ型の指定が楽な分良い気がしますが、
④の方が動きが早いのかな？

あと気になっているのはデータ量が多くなった際に、
google-cloud-bigqueryでのinsert上限が1万レコードだったので
それ以上を行いたい場合の処理内容を各必要はあるのかもしれないです。

その場合は、1回目のみテーブルの差し替えで
2回目以降はisertされる処理にすれば問題なさそうです。