grpcurl と BigQuery Storage Write API を使ってデータを書き込む 

# BigQuery Storage Write API
gRPC でデータを書き込むことができます。

- [BigQuery Write API 特集: Write API の概要 | Google Cloud 公式ブログ](https://cloud.google.com/blog/ja/topics/developers-practitioners/bigquery-write-api-explained-overview-write-api?hl=ja)
> 2021 年に一般提供が開始された Google BigQuery Write API は、1 つの統合型 API で高性能なバッチ処理とストリーミングを実現する、BigQuery の推奨データ取り込みパスです。
> ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/101361/2a11c661-0602-ca50-354f-c272828e599c.png)

- [新しい BigQuery Storage Write API を使用して、JSON データを BigQuery にストリーミングする方法を学ぶ | Google Cloud 公式ブログ](https://cloud.google.com/blog/ja/topics/developers-practitioners/learn-how-stream-json-data-bigquery-using-new-bigquery-storage-write-api?hl=ja)
> Write API は、プロトコル バッファ形式のバイナリデータの使用を想定しています。そのため、この API は高スループットのストリーミングに非常に有効です。しかし、プロトコル バッファを扱うのは若干難しい場合もあります。

- [BigQuery Storage Write API を使用してデータを一括読み込み、ストリーミングする  |  Google Cloud](https://cloud.google.com/bigquery/docs/write-api?hl=ja)
> 効率的なプロトコル。Storage Write API は、HTTP over REST ではなく gRPC ストリーミングを使用します。このため、以前の insertAll メソッドよりも効率的です。Storage Write API は、プロトコル バッファを使用してバイナリ形式をサポートします。これは JSON よりも効率的な転送方式です。書き込みリクエストは非同期ですが、順序指定が保証されます
> ...
> 費用低減。Storage Write API は、以前の insertAll ストリーミング API よりも大幅に低コストです。また、1 か月あたり最大 2 TB を無料で取り込めます。

# API 認証情報

- [必要な権限](https://cloud.google.com/bigquery/docs/write-api?hl=ja#required_permissions)
> Storage Write API を使用するには、bigquery.tables.updateData 権限が必要です。
>bigquery.tables.updateData 権限は、以下の Identity and Access Management（IAM）事前定義ロールに含まれています。
> - bigquery.dataEditor
> - bigquery.dataOwner
> - bigquery.admin

以下のような JSON ファイルで取得できます。

```json:key.json
{
  "type": "service_account",
  "project_id": "xxx",
  "private_key_id": "xxx",
  "private_key": "-----BEGIN PRIVATE KEY-----xxx-----END PRIVATE KEY-----\n",
  "client_email": "${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com",
  "client_id": "xxx",
  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
  "token_uri": "https://oauth2.googleapis.com/token",
  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/${SERVICE_ACCOUNT_NAME}%40${PROJECT_ID}.iam.gserviceaccount.com",
  "universe_domain": "googleapis.com"
}
```

## 新規作成
以下のコマンドで作成できます。

```shell:
export SERVICE_ACCOUNT_NAME=YOUR_SERVICE_ACCOUNT_NAME
export PROJECT_ID=YOUR_PROJECT_ID

gcloud iam service-accounts create ${SERVICE_ACCOUNT_NAME} \
--display-name ${SERVICE_ACCOUNT_NAME}
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
--member serviceAccount:${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com \
--role roles/bigquery.dataEditor

gcloud projects get-iam-policy ${PROJECT_ID} \
--flatten="bindings[].members" \
--format='table(bindings.role)' \
--filter="bindings.members:${SERVICE_ACCOUNT_NAME}"

gcloud iam service-accounts keys create key.json \
--iam-account=${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com 
```

## 既存利用
すでにあるサービスアカウントから探すには以下のコマンドが使えます。

- [google cloud platform - How do I list the roles associated with a gcp service account? - Stack Overflow](https://stackoverflow.com/questions/47006062/how-do-i-list-the-roles-associated-with-a-gcp-service-account)

```shell:
export SERVICE_ACCOUNT_NAME=YOUR_SERVICE_ACCOUNT_NAME
export PROJECT_ID=YOUR_PROJECT_ID

gcloud projects get-iam-policy ${PROJECT_ID} \
--flatten="bindings[].members" \
--format="table(bindings.members)" \
--filter="bindings.role:bigquery.dataEditor"
gcloud projects get-iam-policy ${PROJECT_ID} \
--flatten="bindings[].members" \
--format="table(bindings.members)" \
--filter="bindings.role:bigquery.dataOwner"
gcloud projects get-iam-policy ${PROJECT_ID} \
--flatten="bindings[].members" \
--format="table(bindings.members)" \
--filter="bindings.role:bigquery.admin"

gcloud iam service-accounts keys create key.json \
--iam-account=${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com 
```

# データセット・テーブル・スキーマ作成
サンプルテーブルを以下のように作成します。

- [BigQueryのテーブル作成実践(データ定義言語(DDL)) | DevelopersIO](https://dev.classmethod.jp/articles/bigquery-create-table-by-ddl/)

```shell:
bq mk kyouhei

bq query --nouse_legacy_sql '
CREATE TABLE
  kyouhei.sample_table
(
  id STRING NOT NULL,
  user_uuid STRING NOT NULL,
  action_type INT64 NOT NULL,
)
'
```

# grpcurl でテータ書き込み

`grpcurl` を使ってデータを書き込みます。

- [fullstorydev/grpcurl: Like cURL, but for gRPC: Command-line tool for interacting with gRPC servers](https://github.com/fullstorydev/grpcurl)

> ```shell:
> brew install grpcurl
> ```

`row.json`　として書き込みたい JSON データを準備します。

```json:row.json
{"id":"testId","user_uuid":"testUuid","action_type":1}
```

`sample.proto` と `proto-convert` を使って `row.bin` バイナリに変換します。

- [iamazeem/proto-convert: a CLI tool to convert protobuf messages from binary to JSON and vice versa](https://github.com/iamazeem/proto-convert)

```proto:sample.proto
syntax = "proto3";

enum ActionType {
  ACTION_UNSPECIFIED = 0;
  ACTION_CREATE = 1;
  ACTION_UPDATE = 2;
  ACTION_DELETE = 3;
}

message SampleTable {
  string id = 1;
  string user_uuid = 2;
  ActionType action_type = 3;
}
```

```shell:
proto-convert -m j2b -p sample.proto -t SampleTable -i row.json -o row.bin
```

`row.bin` バイナリから `row.txt` base64 エンコードテキストに変換します。

```shell:
base64 -i row.bin -o row.txt
# cat row.txt --> CgZ0ZXN0SWQSCHRlc3RVdWlkGAE=
```

`serializedRows` に `row.txt` の内容を入れ、`test.json` を作成します。
デフォルトストリームに書き込む形をとります。

- [Package google.cloud.bigquery.storage.v1  |  BigQuery  |  Google Cloud](https://cloud.google.com/bigquery/docs/reference/storage/rpc/google.cloud.bigquery.storage.v1#google.cloud.bigquery.storage.v1.AppendRowsRequest)

```json:test.json
{
  "writeStream": "projects/${PROJECT_ID}/datasets/kyouhei/tables/sample_table/_default",
  "protoRows": {
    "writerSchema": {
      "protoDescriptor": {
        "name": "sampleTable",
        "field": [
          {
            "name": "id",
            "number": 1,
            "type": 9
          },
          {
            "name": "user_uuid",
            "number": 2,
            "type": 9
          },
          {
            "name": "action_type",
            "number": 3,
            "type": 3
          }
        ]
      }
    },
    "rows": {
      "serializedRows": [
        "CgZ0ZXN0SWQSCHRlc3RVdWlkGAE="
      ]
    }
  }
}
```

空の `serializedRows` を作って、以下のコマンドを使って挿入してもいいです。

- [コマンドライン上でJSONを上書き保存する - Qiita](https://qiita.com/pickles/items/aa5b6d5453a84511db21)

```shell:
jq '.protoRows.rows.serializedRows |= .+ ["'$(cat row.txt)'"]' test.json > tmp && mv tmp test.json
```

`storage.proto` ファイルが必要なので、ダウンロードしておきます。

- [googleapis/google/cloud/bigquery/storage/v1/storage.proto at master · googleapis/googleapis](https://github.com/googleapis/googleapis/blob/master/google/cloud/bigquery/storage/v1/storage.proto)

```shell:
git clone https://github.com/googleapis/googleapis.git
```

`test.json` と認証トークンを使って `grpcurl` でデータを書き込むことがます。

- [gRPCをcurlのようにコンソールから呼び出せるgrpcurl - 理系学生日記](https://kiririmode.hatenablog.jp/entry/20181008/1538984412#Unary-RPC-%E5%91%BC%E5%87%BA)

```shell
export GOOGLE_APPLICATION_CREDENTIALS=key.json
TOKEN=$(gcloud auth application-default print-access-token)
echo $TOKEN

grpcurl -vv \
-import-path ~/googleapis \
-proto google/cloud/bigquery/storage/v1/storage.proto \
-H "Authorization: Bearer ${TOKEN}" \
-d @ \
bigquerystorage.googleapis.com:443 \
google.cloud.bigquery.storage.v1.BigQueryWrite.AppendRows < test.json
```

```shell:result
Resolved method descriptor:
// Appends data to the given stream.
//
// If `offset` is specified, the `offset` is checked against the end of
// stream. The server returns `OUT_OF_RANGE` in `AppendRowsResponse` if an
// attempt is made to append to an offset beyond the current end of the stream
// or `ALREADY_EXISTS` if user provides an `offset` that has already been
// written to. User can retry with adjusted offset within the same RPC
// connection. If `offset` is not specified, append happens at the end of the
// stream.
//
// The response contains an optional offset at which the append
// happened.  No offset information will be returned for appends to a
// default stream.
//
// Responses are received in the same order in which requests are sent.
// There will be one response for each successful inserted request.  Responses
// may optionally embed error information if the originating AppendRequest was
// not successfully processed.
//
// The specifics of when successfully appended data is made visible to the
// table are governed by the type of stream:
//
// * For COMMITTED streams (which includes the default stream), data is
// visible immediately upon successful append.
//
// * For BUFFERED streams, data is made visible via a subsequent `FlushRows`
// rpc which advances a cursor to a newer offset in the stream.
//
// * For PENDING streams, data is not made visible until the stream itself is
// finalized (via the `FinalizeWriteStream` rpc), and the stream is explicitly
// committed via the `BatchCommitWriteStreams` rpc.
rpc AppendRows ( stream .google.cloud.bigquery.storage.v1.AppendRowsRequest ) returns ( stream .google.cloud.bigquery.storage.v1.AppendRowsResponse ) {
  option (.google.api.http) = {
    post: "/v1/{write_stream=projects/*/datasets/*/tables/*/streams/*}"
    body: "*"
  };
  option (.google.api.method_signature) = "write_stream";
}

Request metadata to send:
authorization: Bearer xxx

Response headers received:
alt-svc: h3=":443"; ma=2592000,h3-29=":443"; ma=2592000
content-disposition: attachment
content-type: application/grpc
date: Tue, 22 Aug 2023 08:10:36 GMT

Estimated response size: 74 bytes

Response contents:
{
  "appendResult": {
    
  },
  "writeStream": "projects/${PROJECT_ID}/datasets/kyouhei/tables/sample_table/_default"
}

Response trailers received:
content-disposition: attachment
Sent 1 request and received 1 response
```

以下のクエリでデータが書き込まれたことが確認できます。

```shell:
bq query --nouse_legacy_sql '
SELECT * FROM kyouhei.sample_table
'
```

```shell:
+--------+-----------+-------------+
|   id   | user_uuid | action_type |
+--------+-----------+-------------+
| testId | testUuid  |           1 |
+--------+-----------+-------------+
```

# 参考

- [Cloud Functions から BigQuery Storage Write API 使ってみた](https://zenn.dev/aldagram_tech/articles/8e09cff2bd74a1)