このドキュメントでは、以前の tabledata.insertAll
メソッドを使用して BigQuery にデータをストリーミングする方法について説明します。
新しいプロジェクトでは、tabledata.insertAll
メソッドの代わりに BigQuery Storage Write API を使用することをおすすめします。Storage Write API は、1 回限りの配信セマンティクスなど、低価格でより堅牢な機能を備えています。既存のプロジェクトを tabledata.insertAll
メソッドから Storage Write API に移行する場合は、デフォルト ストリームを選択することをおすすめします。tabledata.insertAll
メソッドは引き続き完全にサポートされます。
始める前に
宛先テーブルを含んだデータセットへの書き込みアクセス権があることを確認します。テーブルにデータを書き込む前に、まずテーブルが存在している必要があります。ただし、テンプレート テーブルを使用する場合は異なります。テンプレート テーブルについて詳しくは、テンプレート テーブルを使用した自動的なテーブル作成をご覧ください。
データ ストリーミングの割り当てポリシーをチェックします。
-
Make sure that billing is enabled for your Google Cloud project.
このドキュメントの各タスクを実行するために必要な権限をユーザーに与える Identity and Access Management(IAM)のロールを付与します。
ストリーミングは、無料枠では利用できません。課金を有効にせずにストリーミングを使用しようとすると、次のエラーが表示されます。BigQuery: Streaming insert is not allowed in the free tier.
必要な権限
BigQuery にデータをストリーミングするには、次の IAM 権限が必要です。
bigquery.tables.updateData
(テーブルにデータを挿入)bigquery.tables.get
(テーブルのメタデータを取得)bigquery.datasets.get
(データセットのメタデータを取得)bigquery.tables.create
(テンプレート テーブルを使用してテーブルを自動的に作成する場合は必須)
次の IAM 事前定義ロールには、BigQuery にデータをストリーミングするために必要な権限が含まれています。
roles/bigquery.dataEditor
roles/bigquery.dataOwner
roles/bigquery.admin
BigQuery での IAM のロールと権限について詳しくは、事前定義ロールと権限をご覧ください。
BigQuery にデータをストリーミングする
C#
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある C# の設定手順を完了してください。詳細については、BigQuery C# API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Go
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Go の設定手順を完了してください。詳細については、BigQuery Go API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Java
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Java の設定手順を完了してください。詳細については、BigQuery Java API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Node.js
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Node.js の設定手順を完了してください。詳細については、BigQuery Node.js API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
PHP
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある PHP の設定手順を完了してください。詳細については、BigQuery PHP API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Python
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Python の設定手順を完了してください。詳細については、BigQuery Python API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Ruby
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Ruby の設定手順を完了してください。詳細については、BigQuery Ruby API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
行の挿入時に insertID
フィールドに値を入力する必要はありません。次の例は、ストリーミング時に各行の insertID
の送信を防ぐ方法を示しています。
Java
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Java の設定手順を完了してください。詳細については、BigQuery Java API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
Python
このサンプルを試す前に、クライアント ライブラリを使用した BigQuery クイックスタートにある Python の設定手順を完了してください。詳細については、BigQuery Python API のリファレンス ドキュメントをご覧ください。
BigQuery に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、クライアント ライブラリの認証情報を設定するをご覧ください。
日時データを送信する
日時フィールドの場合、tabledata.insertAll
メソッドで次のようにデータをフォーマットします。
タイプ | 形式 |
---|---|
DATE |
"YYYY-MM-DD" 形式の文字列 |
DATETIME |
"YYYY-MM-DD [HH:MM:SS]" 形式の文字列 |
TIME |
"HH:MM:SS" 形式の文字列 |
TIMESTAMP |
1970-01-01(Unix エポック)からの経過秒数、または "YYYY-MM-DD HH:MM[:SS]" 形式の文字列 |
範囲データを送信する
RANGE<T>
型のフィールドの場合は、tabledata.insertAll
メソッド内のデータの形式を、start
と end
の 2 つのフィールドを持つ JSON オブジェクトにします。start
フィールドと end
フィールドに値が指定されていない場合、または値が NULL の場合は、境界が存在しないことを表します。これらのフィールドは、タイプ T
のサポートされている JSON 形式(T
は DATE
、DATETIME
、TIMESTAMP
のいずれか)と同じである必要があります。
次の例では、f_range_date
フィールドはテーブルの RANGE<DATE>
列を表します。tabledata.insertAll
API を使用して、この列に 1 行挿入します。
{
"f_range_date": {
"start": "1970-01-02",
"end": null
}
}
ストリーム データの可用性
GoogleSQL クエリを使用したリアルタイム分析にデータは、BigQuery が tabledata.insertAll
リクエストの確認応答に成功した直後に使用できるようになります。
最近取り込み時間パーティション分割テーブルにストリーミングした行は、_PARTITIONTIME
疑似列の値が一時的に NULL になります。このような行に対して、BigQuery はバックグラウンドで PARTITIONTIME
列の最終的な NULL 以外の値を割り当てます。これは通常、数分以内に行われます。まれに最大 90 分かかることがあります。
最近ストリーミングされた行の一部は、通常、数分間はテーブルのコピーに使用できない可能性があります。まれに最大 90 分かかることがあります。データがコピーに使用できるかどうかを確認するには、tables.get
に対するレスポンスに streamingBuffer
というセクションがあるかどうかを確認します。streamingBuffer
セクションがなければ、データはコピーに利用できます。streamingBuffer.oldestEntryTime
フィールドを使用して、ストリーミング バッファ内のレコードの経過時間を特定することもできます。
ベスト エフォート型の重複排除
挿入された行に対して insertId
を指定した場合、BigQuery はこの ID を使用して、ベスト エフォート型の重複排除を最大 1 分間サポートします。つまり、その期間内に同じ insertId
の同じ行を同じテーブルに複数回ストリーミングしようとすると、BigQuery はその行の複数のオカレンスを重複排除して、それらのオカレンスの一つだけを保持する可能性があります。
このシステムでは、同じ insertId
が指定された行も同一のものであると想定されます。2 つの行が同じ insertId
を持つ場合、どちらの行を BigQuery が保持するかは非決定的になります。
重複排除は一般に、システムと BigQuery 間のネットワーク エラーや BigQuery の内部エラーといった特定のエラー状態でストリーミング挿入の状態を判断する方法がない分散システムでの再試行シナリオ向けです。挿入を再試行する場合は、BigQuery がデータの重複排除を試行できるよう、同じ行セットに同じ insertId
を使用するようにしてください。詳細については、ストリーミング挿入に関するトラブルシューティングをご覧ください。
BigQuery の重複排除はベスト エフォート型であり、データの重複がないことを保証するメカニズムとしての使用には適していません。さらに、データの高い信頼性と可用性を保証するために、BigQuery はベスト エフォート型の重複排除の品質を低下させる可能性があります。
データの重複排除に関して厳密な要件がある場合は、Google Cloud Datastore がトランザクションをサポートする代替サービスとなります。
ベスト エフォート型の重複排除の無効化
ベスト エフォート型の重複排除を無効にするには、挿入された各行の insertId
フィールドに値を設定しないようにします。これは、データの挿入を行う場合に推奨される方法です。
Apache Beam と Dataflow
Apache Beam の Java 用 BigQuery I/O コネクタを使用しているときにベスト エフォート型の重複排除を無効にするには、ignoreInsertIds()
メソッドを使用します。
手動の重複排除
ストリーミングの実行後に重複行が残らないようにするには、次の手動プロセスを使用します。
- テーブル スキーマ内の列として
insertId
を追加し、各行のデータにinsertId
値を含めます。 - ストリーミングが停止した後に、次のクエリを実行して重複をチェックします。
結果が 1 より大きい場合は、重複が存在します。#standardSQL SELECT MAX(count) FROM( SELECT ID_COLUMN, count(*) as count FROM `TABLE_NAME` GROUP BY ID_COLUMN)
- 重複を排除するには、次のクエリを実行します。宛先テーブルを指定してサイズの大きい結果を許容し、結果のフラット化を無効にします。
#standardSQL SELECT * EXCEPT(row_number) FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY ID_COLUMN) row_number FROM `TABLE_NAME`) WHERE row_number = 1
重複排除クエリに関する注記:
- 重複排除クエリの戦略として、新規テーブルをターゲットにした方が安全です。なお、書き込み処理
WRITE_TRUNCATE
を使用して、ソーステーブルをターゲットにすることもできます。 - 重複排除クエリは、テーブル スキーマの末尾に値を
1
に設定したrow_number
列を追加します。このクエリでは、GoogleSQL のSELECT * EXCEPT
ステートメントを使用して、このrow_number
列を宛先テーブルから除外します。#standardSQL
接頭辞を使用すると、このクエリで GoogleSQL が有効になります。また、特定の列名を指定して、この列を省略することもできます。 - ライブデータのクエリで重複を排除するには、重複排除クエリを使用してテーブルのビューを作成するという方法もあります。なお、ビューに対するクエリコストはビュー内の選択列に基づいて計算されるため、スキャンされるバイトサイズが大きくなる可能性があります。
時間パーティション分割テーブルへのストリーミング
時間パーティション分割テーブルにデータをストリーミングする場合は、各パーティションにストリーミング バッファが存在します。writeDisposition
プロパティを WRITE_TRUNCATE
に設定すると、パーティションを上書きする読み込み、クエリ、またはコピーのジョブを実行したときにストリーミング バッファが保持されます。ストリーミング バッファを削除するには、そのパーティションに対して tables.get
を呼び出し、ストリーミング バッファが空であることを確認します。
取り込み時間パーティショニング
取り込み時間パーティション分割テーブルにストリーミングすると、BigQuery は現在の UTC 時間から宛先パーティションを推測します。
新しく到着したデータは、ストリーミング バッファ内にある間、一時的に __UNPARTITIONED__
パーティションに配置されます。パーティション分割されていないデータが十分蓄積されると、BigQuery はデータを正しいパーティションに分割します。ただし、データが __UNPARTITIONED__
パーティションから移動するために要する時間に関する SLA はありません。いずれかの疑似列(優先するデータ型に応じて _PARTITIONTIME
または _PARTITIONDATE
)を使用して、__UNPARTITIONED__
パーティションからの NULL
値を除外することにより、ストリーミング バッファ内のデータをクエリの対象外にできます。
日次パーティション分割テーブルにデータをストリーミングする場合は、insertAll
リクエストの一部としてパーティション デコレータを指定することで、日付の推定をオーバーライドできます。tableId
パラメータにデコレータを含めます。たとえば、次のようにパーティション デコレータを使用して、table1
テーブルの 2021-03-01 に対応するパーティションにストリーミングできます。
table1$20210301
パーティション デコレータを使用してストリーミングを行う際は、現在の UTC 時間に基づき、過去 31 日以内のパーティションと現在の日付から 16 日後までのパーティションにストリーミングできます。この範囲に含まれない日付のパーティションに書き込むには、パーティション分割テーブルデータの追加と上書きで説明するように、読み込みジョブまたはクエリジョブを使用します。
パーティション デコレータを使用したストリーミングは、日単位のパーティション分割テーブルでのみサポートされています。時間単位、月単位、年単位のパーティション分割テーブルではサポートされていません。
テストには、bq コマンドライン ツールの bq insert
CLI コマンドを使用できます。たとえば、次のコマンドを実行すると、2017 年 1 月 1 日($20170101
)のパーティション全体のデータを、mydataset.mytable
という名前のパーティション分割テーブルに 1 行がストリーミングされます。
echo '{"a":1, "b":2}' | bq insert 'mydataset.mytable$20170101'
時間単位列パーティショニング
DATE
列、DATETIME
列、TIMESTAMP
列で分割されたテーブルに、過去 5 年間、向こう 1 年間のデータをストリーミングできます。この範囲外のデータは拒否されます。
データがストリーミングされると、最初に __UNPARTITIONED__
パーティションに配置されます。パーティション分割されていないデータが十分蓄積されると、BigQuery はデータのパーティション再設定を自動的に行い、適切なパーティションに配置します。
ただし、データが __UNPARTITIONED__
パーティションから移動するために要する時間に関する SLA はありません。
- 注: 日単位のパーティションは、時間、月、年単位のパーティションとは異なる方法で処理されます。期間(過去 7 日間から将来の 3 日間まで)外のデータのみがパーティション分割されていないパーティションに抽出され、パーティションの再分割を待ちます。一方、時間単位のパーティション分割テーブルの場合、データは常にパーティション分割されていないパーティションに抽出され、後でパーティションの再分割が行われます。
テンプレート テーブルを使用してテーブルを自動的に作成する
テンプレート テーブルは、論理テーブルを多数の小さなテーブルに分割して、より小さなデータの集合(たとえば、ユーザー ID ごと)を作成するメカニズムを提供します。テンプレート テーブルには、以下に説明するいくつかの制限があります。代わりにパーティション分割テーブルとクラスタ化テーブルを使用してこの動作を実現することをおすすめします。
BigQuery API を介してテンプレート テーブルを使用するには、insertAll
リクエストに templateSuffix
パラメータを追加します。bq コマンドライン ツールの場合は、insert
コマンドに template_suffix
フラグを追加します。BigQuery は、templateSuffix
パラメータまたは template_suffix
フラグを検出した場合、ターゲット テーブルをベース テンプレートとして扱います。さらに、ターゲット テーブルと同じスキーマを共有する新しいテーブルを、指定されたサフィックスを含む名前で作成します。
<targeted_table_name> + <templateSuffix>
テンプレート テーブルを使用すると、各テーブルを個別に作成し、各テーブルのスキーマを指定するためのオーバーヘッドを回避できます。テンプレートを 1 つ作成し、複数のサフィックスを指定するだけで、新規テーブルを BigQuery に自動作成させることができます。BigQuery は、各テーブルを同じプロジェクトとデータセット内に配置します。
テンプレート テーブルを通じて作成されたテーブルは、通常数秒以内に利用可能になります。ただし、それより長い時間を要する場合もまれにあります。
テンプレート テーブル スキーマを変更する
テンプレート テーブル スキーマを変更した場合は、それ以降に生成されるすべてのテーブルで、更新後のスキーマが使用されます。以前に生成されたテーブルには影響はありません(既存のテーブルにストリーミング バッファが残っている場合を除く)。
ストリーミング バッファが残っている既存のテーブルについては、テンプレート テーブル スキーマに対する変更が後方互換性のあるものであれば、現にストリーミングが行われているそれらの生成済テーブルのスキーマも更新されます。ただし、テンプレート テーブル スキーマに対する変更が後方互換性のないものである場合は、古いスキーマを使用するバッファデータがすべて失われます。また、互換性がなくなった古いスキーマを使用する既存の生成済テーブルに、新しいデータをストリーミングすることはできません。
テンプレート テーブル スキーマを変更した後は、その変更が伝播されるまで、新しいデータの挿入や、生成されたテーブルに対するクエリを行わないでください。新しいフィールドを挿入するリクエストは、数分以内に処理されます。新規フィールドに対するクエリ実行には、最大 90 分の待機時間を要する場合があります。
生成されたテーブルのスキーマを変更する場合は、テンプレート テーブルを経由したストリーミングが停止し、生成済みテーブルのストリーミング統計セクションが tables.get()
レスポンスからなくなる(テーブルでバッファされているデータがなくなる)まで、スキーマを変更しないようにしてください。
パーティション分割テーブルとクラスタ化テーブルには上述の制限がないため、おすすめのメカニズムです。
テンプレート テーブルの詳細
- テンプレート サフィックス値
templateSuffix
(または--template_suffix
)値には、英字(a~z、A~Z)、数字(0~9)、アンダースコア(_)のみを含める必要があります。テーブル名とテーブル サフィックスの最大連結文字数は 1,024 文字です。- 割り当て
テンプレート テーブルには、ストリーミング割り当ての制限が適用されます。
tables.insert
API と同様に、プロジェクトでは、テンプレート テーブルを使用して、1 秒あたり最大 10 個のテーブルを作成できます。この割り当ては作成されるテーブルにのみ適用され、変更されるテーブルには適用されません。アプリケーションで 1 秒あたり 10 個を超えるテーブルを作成する必要がある場合は、クラスタ化テーブルの使用をおすすめします。たとえば、カーディナリティが高いテーブル ID を単一のクラスタリング テーブルのキー列に配置できます。
- 有効期間
生成されたテーブルの有効期間はデータセットから継承されます。通常のストリーミング データと同様、生成されたテーブルをすぐにコピーすることはできません。
- 重複排除
重複排除は、宛先テーブルへの同質な参照間でのみ実行されます。 たとえば、テンプレート テーブルと通常の
insertAll
コマンドの両方を使用して生成済テーブルへのストリーミングを同時に実行した場合、テンプレート テーブルと通常のinsertAll
コマンドによって挿入された各行の間で重複排除は実行されません。- ビュー
ビューをテンプレート テーブルとして使用することはできません。また、テンプレート テーブルを元にビューを自動生成させることもできません。
ストリーミング挿入のトラブルシューティング
以降のセクションでは、以前のストリーミング API を使用して BigQuery にデータをストリーミングする際に発生するエラーのトラブルシューティングについて説明します。ストリーミング挿入の割り当てエラーを解決する方法について詳しくは、ストリーミング挿入の割り当てエラーをご覧ください。
失敗 HTTP レスポンス コード
ネットワーク エラーなどの失敗 HTTP レスポンス コードが返された場合、ストリーミング挿入が成功したかどうかを確認する方法はありません。リクエストを再送信しようとすると、テーブル内に重複行が発生する可能性があります。テーブルを重複から保護するには、リクエストの送信時に insertId
プロパティを設定します。insertId
プロパティは BigQuery によって重複排除に使用されます。
権限エラー、無効なテーブル名エラー、または割り当て超過エラーが返された場合、行は挿入されず、リクエスト全体が失敗します。
成功 HTTP レスポンス コード
成功 HTTP レスポンス コードが返された場合でも、BigQuery による行の挿入が部分的にしか成功しなかった可能性があるため、レスポンスの insertErrors
プロパティをチェックして、行の挿入が成功したかどうかを確認する必要があります。次のシナリオのいずれかが発生することがあります。
- すべての行が正常に挿入されました。
insertErrors
プロパティが空のリストになっている場合は、すべての行が正常に挿入されています。 - 一部の行が正常に挿入されました。いずれかの行にスキーマの不一致がある場合を除き、
insertErrors
プロパティで示された行が挿入されておらず、その他すべての行は正常に挿入されています。errors
プロパティには、行の挿入が失敗した理由に関する詳細情報が含まれます。index
プロパティは、エラーに該当するリクエストの 0 ベースの行インデックスを示します。 - 挿入された行がない。BigQuery がリクエスト内で個別行のスキーマ不一致に遭遇した場合は、いずれの行も挿入されず、スキーマ不一致がなかった行も含めて、各行の
insertErrors
エントリが返されます。スキーマ不一致がなかった行は、エラーのreason
プロパティがstopped
に設定され、そのまま再送信できます。失敗した行には、スキーマ不一致に関する詳細情報が含まれます。BigQuery データ型でサポートされているプロトコル バッファの型については、データ型の変換をご覧ください。
ストリーミング挿入のメタデータ エラー
BigQuery のストリーミング API はインサート率が高くなるように設計されているため、ストリーミング システムを操作する際、基盤となるテーブルのメタデータの変更は結果整合が保たれます。ほとんどの場合、メタデータの変更は数分以内にプロパゲートされますが、その間は API レスポンスで、対象テーブルの不整合な状態が反映されることがあります。
次のようなシナリオがあります。
- スキーマの変更。スキーマの変更がストリーミング システムにすぐに反映されない場合があるため、ストリーミング挿入を最近受け取ったテーブルのスキーマを変更すると、スキーマの不一致エラーが発生することがあります。
- テーブルの作成 / 削除。存在しないテーブルへのストリーミングによって、
notFound
レスポンスのバリエーションが返されます。レスポンスで作成されたテーブルは、後続のストリーミング挿入ですぐには認識されない可能性があります。同様に、テーブルを削除または、再作成すると、古いテーブルにストリーミング挿入が実質的に配信される期間が生じることがあります。このストリーミング挿入は新しいテーブルには存在しない可能性があります。 - テーブルの切り捨て - (WRITE_TRUNCATE の writeDisposition を使用するクエリジョブによって)テーブルのデータを切り捨てる場合も同様に、整合性期間中の後続の挿入が破棄されることがあります。
欠損 / 利用不可能なデータ
ストリーミング挿入は一時的に書き込み最適化ストレージに存在します。このストレージは、マネージド ストレージとは異なる可用性の特性を持ちます。BigQuery の一部のオペレーション(テーブル コピー ジョブや tabledata.list
などの API メソッドなど)では、書き込み用に最適化されたストレージは操作されません。最新のストリーミング データが宛先テーブルまたは出力に存在しません。