BigQuery の宛先を構成する

このページでは、Datastream を使用してソースデータベースからデータをストリーミングするために BigQuery の宛先を構成する方法について説明します。

宛先データセットを構成する

BigQuery の宛先のデータセットを構成する際には、次のいずれかのオプションを選択できます。

各スキーマのデータセット: データセットは、ソースのスキーマ名に基づいて、指定された BigQuery のロケーションで選択または作成されます。そのため、Datastream は、ソースのスキーマごとに自動的に BigQuery にデータセットを作成します。

このオプションを選択すると、Datastream は、ストリームを含むプロジェクトにデータセットを作成します。

たとえば、MySQL ソースがあり、このソースのデータベース内に mydb データベースと employees テーブルがある場合、Datastream は BigQuery に mydb データセットと employees テーブルを作成します。
すべてのスキーマの単一のデータセット: ストリームの BigQuery データセットを選択できます。Datastream は、すべてのデータをこのデータセットにストリーミングします。選択したデータセットに対して、Datastream はすべてのテーブルを <schema>_<table> として作成します。

たとえば、MySQL ソースがあり、このソースのデータベース内に mydb データベースと employees テーブルがある場合、Datastream は選択したデータセットに mydb_employees テーブルを作成します。

書き込み動作

Datastream は、BigQuery の宛先に書き込まれる各テーブルに datastream_metadata という名前の STRUCT 列を追加します。

ソースのテーブルに主キーがある場合、列には次のフィールドが含まれます。

UUID: このフィールドには string データ型が含まれます。
SOURCE_TIMESTAMP: このフィールドには integer データ型が含まれます。

テーブルに主キーがない場合、その列には追加のフィールド IS_DELETED が含まれます。このフィールドには boolean データ型があり、Datastream が宛先にストリーミングするデータがソースの DELETE オペレーションに関連付けられているかどうかを示します。主キーのないテーブルは追記専用です。

BigQuery にデータをストリーミングする際のイベントの最大サイズは 20 MB です。

`max_staleness` オプションで BigQuery テーブルを使用する

ほぼリアルタイムの取り込みの一環として、Datastream は、データの更新、挿入、削除などの upsert オペレーションに対する BigQuery の組み込みサポートを使用します。upsert オペレーションを使用すると、行の追加、変更、削除に応じて BigQuery の宛先を動的に更新できます。Datastream は、BigQuery の Storage Write API を使用して、これらの upsert オペレーションを宛先テーブルにストリーミングします。

データ未更新の制限を指定する

BigQuery は、構成されたデータ未更新の制限に従って、ソースの変更をバックグラウンドで継続的に、またはクエリの実行時に適用します。Datastream が BigQuery で新しいテーブルを作成する場合、テーブルの max_staleness オプションは、ストリームの現在のデータ未更新の上限値に従って設定されます。

max_staleness オプション付きで BigQuery テーブルを使用する方法については、テーブルの未更新をご覧ください。

BigQuery の費用を管理する

BigQuery の費用は、Datastream とは別に課金されます。BigQuery の費用を管理する方法については、BigQuery CDC の料金をご覧ください。

データ型のマッピング

次の表に、サポートされているソースデータベースから BigQuery の宛先へのデータ型の変換を示します。

ソースデータベース	ソースのデータ型	BigQuery のデータ型
MySQL	`BIGINT(size)`	`INT64`
MySQL	`BIGINT (unsigned)`	`NUMERIC`
MySQL	`BINARY(size)`	`STRING (hex encoded)`
MySQL	`BIT(size)`	`INT64`
MySQL	`BLOB(size)`	`STRING (hex encoded)`
MySQL	`BOOL`	`INT64`
MySQL	`CHAR(size)`	`STRING`
MySQL	`DATE`	`DATE`
MySQL	`DATETIME(fsp)`	`DATETIME`
MySQL	`DECIMAL(precision, scale)`	精度値が 38 以下で、スケール値が 9 以下の場合、`NUMERIC` です。それ以外の場合 `BIGNUMERIC`
MySQL	`DOUBLE(size, d)`	`FLOAT64`
MySQL	`ENUM(val1, val2, val3, ...)`	`STRING`
MySQL	`FLOAT(precision)`	`FLOAT64`
MySQL	`FLOAT(size, d)`	`FLOAT64`
MySQL	`INTEGER(size)`	`INT64`
MySQL	`JSON`	`JSON` `BOOLEAN`、`NUMBER`、`OBJECT`、`STRING`、`ARRAY`、`NULL` データ型がサポートされています。
MySQL	`LONGBLOB`	`STRING (hex encoded)`
MySQL	`LONGTEXT`	`STRING (hex encoded)`
MySQL	`MEDIUMBLOB`	`STRING (hex encoded)`
MySQL	`MEDIUMINT(size)`	`INT64`
MySQL	`MEDIUMTEXT`	`STRING`
MySQL	`SET(val1, val2, val3, ...)`	`STRING`
MySQL	`SMALLINT(size)`	`INT64`
MySQL	`TEXT(size)`	`STRING`
MySQL	`TIME(fsp)`	`INTERVAL`
MySQL	`TIMESTAMP(fsp)`	`TIMESTAMP`
MySQL	`TINYBLOB`	`STRING (hex encoded)`
MySQL	`TINYINT(size)`	`INT64`
MySQL	`TINYTEXT`	`STRING`
MySQL	`VARBINARY(size)`	`STRING (hex encoded)`
MySQL	`VARCHAR`	`STRING`
MySQL	`YEAR`	`INT64`
Oracle	`ANYDATA`	`UNSUPPORTED`
Oracle	`BFILE`	`STRING`
Oracle	`BINARY DOUBLE`	`FLOAT64`
Oracle	`BINARY FLOAT`	`FLOAT64`
Oracle	`BLOB`	`BYTES`
Oracle	`CHAR`	`STRING`
Oracle	`CLOB`	`STRING`
Oracle	`DATE`	`DATETIME` ゼロ日付はサポートされておらず、`NULL` の値に置き換えられます。
Oracle	`DOUBLE PRECISION`	`FLOAT64`
Oracle	`FLOAT(p)`	`FLOAT64`
Oracle	`INTERVAL DAY TO SECOND`	`UNSUPPORTED`
Oracle	`INTERVAL YEAR TO MONTH`	`UNSUPPORTED`
Oracle	`LONG`/`LONG RAW`	`STRING`
Oracle	`NCHAR`	`STRING`
Oracle	`NCLOB`	`STRING`
Oracle	`NUMBER`	`STRING`
Oracle	`NUMBER(precision=*)`	`STRING`
Oracle	`NUMBER(precision, scale<=0)`	p<=18 の場合、`INT64`。18<p=78 の場合、パラメータ化された小数型にマッピング。p>=77 の場合、`STRING` にマッピング
Oracle	`NUMBER(precision, scale>0)`	0<p=77 の場合、パラメータ化された小数型にマッピング。p>=77 の場合、`STRING` にマッピング
Oracle	`NVARCHAR2`	`STRING`
Oracle	`RAW`	`STRING`
Oracle	`ROWID`	`STRING`
Oracle	`SDO_GEOMETRY`	`UNSUPPORTED`
Oracle	`SMALLINT`	`INT64`
Oracle	`TIMESTAMP`	`TIMESTAMP` ゼロの日付はサポートされておらず、`NULL` の値に置き換えられます。
Oracle	`TIMESTAMP WITH TIME ZONE`	`TIMESTAMP` ゼロの日付はサポートされておらず、`NULL` の値に置き換えられます。
Oracle	`UDT (user-defined type)`	`UNSUPPORTED`
Oracle	`UROWID`	`STRING`
Oracle	`VARCHAR`	`STRING`
Oracle	`VARCHAR2`	`STRING`
Oracle	`XMLTYPE`	`UNSUPPORTED`
PostgreSQL	`ARRAY`	`JSON` `DATE`、`TIMESTAMP`、`TIMESTAMP WITH TIME ZONE` データ型の配列はサポートされていません。
PostgreSQL	`BIGINT`	`INT64`
PostgreSQL	`BIT`	`BYTES`
PostgreSQL	`BIT_VARYING`	`BYTES`
PostgreSQL	`BOOLEAN`	`BOOLEAN`
PostgreSQL	`BOX`	`UNSUPPORTED`
PostgreSQL	`BYTEA`	`BYTES`
PostgreSQL	`CHARACTER`	`STRING`
PostgreSQL	`CHARACTER_VARYING`	`STRING`
PostgreSQL	`CIDR`	`STRING`
PostgreSQL	`CIRCLE`	`UNSUPPORTED`
PostgreSQL	`DATE`	`DATE`
PostgreSQL	`DOUBLE_PRECISION`	`FLOAT64`
PostgreSQL	`ENUM`	`STRING`
PostgreSQL	`INET`	`STRING`
PostgreSQL	`INTEGER`	`INT64`
PostgreSQL	`INTERVAL`	`INTERVAL`
PostgreSQL	`JSON`	`JSON`
PostgreSQL	`JSONB`	`JSON`
PostgreSQL	`LINE`	`UNSUPPORTED`
PostgreSQL	`LSEG`	`UNSUPPORTED`
PostgreSQL	`MACADDR`	`STRING`
PostgreSQL	`MONEY`	`FLOAT64`
PostgreSQL	`NUMERIC`	精度 = `-1` の場合は、`STRING`（BigQuery `NUMERIC` 型には固定精度が必要です）。それ以外の場合は `BIGNUMERIC`/`NUMERIC`。詳細については、PostgreSQL ドキュメントの任意精度の数値をご覧ください。
PostgreSQL	`OID`	`INT64`
PostgreSQL	`PATH`	`UNSUPPORTED`
PostgreSQL	`POINT`	`UNSUPPORTED`
PostgreSQL	`POLYGON`	`UNSUPPORTED`
PostgreSQL	`REAL`	`FLOAT64`
PostgreSQL	`SMALLINT`	`INT64`
PostgreSQL	`SMALLSERIAL`	`INT64`
PostgreSQL	`SERIAL`	`INT64`
PostgreSQL	`TEXT`	`STRING`
PostgreSQL	`TIME`	`TIME`
PostgreSQL	`TIMESTAMP`	`TIMESTAMP`
PostgreSQL	`TIMESTAMP_WITH_TIMEZONE`	`TIMESTAMP`
PostgreSQL	`TIME_WITH_TIMEZONE`	`TIME`
PostgreSQL	`TSQUERY`	`STRING`
PostgreSQL	`TSVECTOR`	`STRING`
PostgreSQL	`TXID_SNAPSHOT`	`STRING`
PostgreSQL	`UUID`	`STRING`
PostgreSQL	`XML`	`STRING`
SQL Server	`BIGINT`	`INT64`
SQL Server	`BINARY`	`BYTES`
SQL Server	`BIT`	`BOOL`
SQL Server	`CHAR`	`STRING`
SQL Server	`DATE`	`DATE`
SQL Server	`DATETIME2`	`DATETIME`
SQL Server	`DATETIME`	`DATETIME`
SQL Server	`DATETIMEOFFSET`	`TIMESTAMP`
SQL Server	`DECIMAL`	`BIGNUMERIC`
SQL Server	`FLOAT`	`FLOAT64`
SQL Server	`IMAGE`	`BYTES`
SQL Server	`INT`	`INT64`
SQL Server	`MONEY`	`BIGNUMERIC`
SQL Server	`NCHAR`	`STRING`
SQL Server	`NTEXT`	`STRING`
SQL Server	`NUMERIC`	`BIGNUMERIC`
SQL Server	`NVARCHAR`	`STRING`
SQL Server	`NVARCHAR(MAX)`	`STRING`
SQL Server	`REAL`	`FLOAT64`
SQL Server	`SMALLDATETIME`	`DATETIME`
SQL Server	`SMALLINT`	`INT64`
SQL Server	`SMALLMONEY`	`NUMERIC`
SQL Server	`TEXT`	`STRING`
SQL Server	`TIME`	`TIME`
SQL Server	`TIMESTAMP`/`ROWVERSION`	`BYTES`
SQL Server	`TINYINT`	`INT64`
SQL Server	`UNIQUEIDENTIFIER`	`STRING`
SQL Server	`VARBINARY`	`BYTES`
SQL Server	`VARBINARY(MAX)`	`BYTES`
SQL Server	`VARCHAR`	`STRING`
SQL Server	`VARCHAR(MAX)`	`STRING`
SQL Server	`XML`	`STRING`

PostgreSQL の配列を BigQuery 配列データ型としてクエリする

PostgreSQL 配列を BigQuery ARRAY データ型としてクエリする場合は、BigQuery の JSON_VALUE_ARRAY 関数を使用することで JSON の値を BigQuery の列に変換できます。:

  SELECT ARRAY(SELECT CAST(element AS TYPE) FROM UNNEST(JSON_VALUE_ARRAY(BQ_COLUMN_NAME,'$')) AS element) AS array_col

次のように置き換えます。

TYPE: PostgreSQL ソース配列の要素型と一致する BigQuery の型。たとえば、ソースタイプが BIGINT 値の配列の場合は、TYPE を INT64 に置き換えます。

データ型をマッピングする方法については、データ型のマッピングをご覧ください。
BQ_COLUMN_NAME: BigQuery テーブル内の関連する列の名前。

値の変換方法には、次の 2 つの例外があります。

ソース列内の BIT、BIT_VARYING、または BYTEA の値の配列に対して、次のクエリを実行します。

SELECT ARRAY(SELECT FROM_BASE64(element) FROM UNNEST(JSON_VALUE_ARRAY(BQ_COLUMN_NAME,'$')) AS element) AS array_of_bytes

ソース列内の JSON 値または JSONB 値の配列には、JSON_QUERY_ARRAY 関数を使用します。

SELECT ARRAY(SELECT element FROM UNNEST(JSON_QUERY_ARRAY(BQ_COLUMN_NAME,'$')) AS element) AS array_of_jsons

既知の制限事項

宛先として BigQuery を使用する場合の既知の制限事項は次のとおりです。

データは、Datastream ストリームと同じ Google Cloud プロジェクトにある BigQuery データセットにのみ複製できます。
ソースデータベースのテーブルに主キー（例: ユーザー ID）があり、主キーが異なるようにテーブルが変更される場合（この例では、主キーがユーザー ID と日付になる）、Datastream はこの変更を宛先にストリーミングできません。

Datastream で、変更した主キーを使用してテーブルを宛先にストリーミングするには、次の手順を行います。
1. ソースデータベーステーブルの主キーを変更します。
2. ストリームの合計レイテンシ指標を確認し、少なくとも現在のレイテンシと同じ時間待機して、処理中のイベントが宛先に書き込まれるようにします。これにより、元の主キーを持つすべてのイベントが正常にストリーミングされます。
3. ストリーミングを一時停止します。
4. 宛先でテーブルを削除します。
5. ストリーミングを再開します。
6. テーブルに対してバックフィルをトリガーします。
BigQuery の宛先に書き込まれるイベントが遅れて到着することがあります。ただし、テーブルをバックフィルすると、同期されていない行を修正できます。
BigQuery の主キーは次のデータ型でなければなりません。
- DATE
- BOOL
- GEOGRAPHY
- INT64
- NUMERIC
- BIGNUMERIC
- STRING
- TIMESTAMP
- DATETIME
サポートされていないデータ型の主キーを含むテーブルは、Datastream によって複製されません。
BigQuery は、.、$、/、@、+ 文字を含むテーブル名をサポートしていません。Datastream は、宛先テーブルの作成時にこのような文字をアンダースコアに置き換えます。

たとえば、ソースデータベースの table.name は BigQuery では table_name になります。

BigQuery でのテーブル名の詳細については、テーブルの命名をご覧ください。
BigQuery は、4 つを超えるクラスタリング列はサポートしていません。4 つを超える主キー列を持つテーブルを複製する場合、Datastream はクラスタリング列として 4 つの主キー列を使用します。
Datastream は、PostgreSQL 無限大型などの範囲外の日時リテラルを次の値にマッピングします。
- 正の DATE を 9999-12-31 の値に
- 負の DATE を 0001-01-01 の値に
- 正の TIMESTAMP を 9999-12-31 23:59:59.999000 UTC の値に
- 負の TIMESTAMP を 0001-01-01 00:00:00 UTC の値に
BigQuery は、FLOAT データ型または REAL データ型の主キーを持つストリーミングテーブルをサポートしていません。このようなテーブルは複製されません。

BigQuery のデータ型と期間の詳細については、データ型をご覧ください。

分析のために Datastream と Dataflow を実装する

Cloud Storage の宛先を構成する