Google Cloud Storage Import V2 Integration

Google Cloud Storage用のData Connectorは、GCSバケットに保存されている *.tsv および *.csv ファイルの内容をインポートできます。Export Integrationについては、Google Cloud Storage Export V2 Integrationを参照してください。

• Treasure Dataの基礎知識（TD Toolbeltを含む）
• 特定の権限を持つGoogle Cloud Platformアカウント

セキュリティポリシーで IP ホワイトリストが必要な場合は、接続を成功させるために Treasure Data の IP アドレスを許可リストに追加する必要があります。

リージョンごとに整理された静的 IP アドレスの完全なリストは、次のリンクにあります: IP Addresses for Integrations

Google Cloud Storageとの統合は、サーバー間API認証に基づいています。

JSON認証情報の生成に使用するService Accountには、Storage Object Userの権限が必要です。

1. Google Developer Consoleにアクセスします。
2. 左側のメニューでAPIs & Servicesの下のCredentialsを選択します。

3. Create credentialsを選択し、Service accountを選択します。

4. PermissionsからStorage Object User Roleを追加します。

1. IAM & Admin / Workload Identity Poolsの下でcreate poolを選択するか、既存のPoolを選択します。

2. アカウントID 523683666290 でAWSプロバイダーを追加します。

3. Configure provider attributesからAdd mappingを選択します。名前に attribute.account、値に assertion.account を持つ属性を追加し、Saveします。
4. Workload Identity Poolsから作成したプールを選択し、Connected service accountsからDownload configを選択します。
5. アカウントID 523683666290 用に作成したAWSプロバイダーを選択し、Downloadを選択して設定ファイル（Application default credential keyfile）を保存します。

1. バケットリストから宛先バケットを選択し、permissionsタブを選択します。
2. Grant accessを選択し、Principals値に principalSet://iam.googleapis.com/projects/{PROJECT_NUMBER}/locations/global/workloadIdentityPools/{POOL_ID}/attribute.account/523683666290 を追加し、ロールをStorage Object Userに設定します。

データ接続を設定する際、統合にアクセスするための認証を提供します。Treasure Dataでは、認証を設定してからソース情報を指定します。

1. Treasure コンソールを開きます。
2. Integrations Hub > Catalogに移動します。
3. Google Cloud Storage V2を検索して選択します。

4. Create Authenticationを選択します。
5. Authentication Methodを選択し、認証情報を入力します。

6. 接続の名前を入力します。
7. Continueを選択します。

認証済み接続を作成すると、自動的にAuthenticationsに移動します。

1. 作成した接続を検索します。
2. New Sourceを選択します。
3. Data Transfer フィールドでSourceの名前を入力します。
4. Nextを選択します。

1. Nextを選択します。
2. Source Tableダイアログが開きます。以下のパラメータを編集します。

この場合、Source Table設定は次のようになります：

• Bucket: your_bucket
• Path Prefix: path/to/name
• Path Regex: .csv$ （必須ではありません）
• Start after path: path/to/name1.csv（name1.csv以降のファイルをインポートしたい場合）
• Incremental: true（このジョブをスケジュール設定したい場合）

1. Nextを選択します。
2. Data Settingsページが開きます。

必要に応じてデータ設定を編集するか、このページをスキップします。

インポートを実行する前に、Generate Preview を選択してデータのプレビューを表示できます。Data preview はオプションであり、選択した場合はダイアログの次のページに安全にスキップできます。

1. Next を選択します。Data Preview ページが開きます。
2. データをプレビューする場合は、Generate Preview を選択します。
3. データを確認します。

データの配置について、データを配置したいターゲット database と table を選択し、インポートを実行する頻度を指定します。

1. Next を選択します。Storage の下で、インポートされたデータを配置する新しい database を作成するか、既存の database を選択し、新しい table を作成するか、既存の table を選択します。

2. Database を選択 > Select an existing または Create New Database を選択します。

3. オプションで、database 名を入力します。

4. Table を選択 > Select an existing または Create New Table を選択します。

5. オプションで、table 名を入力します。

6. データをインポートする方法を選択します。

   • Append (デフォルト) - データインポートの結果は table に追加されます。 table が存在しない場合は作成されます。
   • Always Replace - 既存の table の全体の内容をクエリの結果出力で置き換えます。table が存在しない場合は、新しい table が作成されます。
   • Replace on New Data - 新しいデータがある場合のみ、既存の table の全体の内容をクエリの結果出力で置き換えます。

7. Timestamp-based Partition Key 列を選択します。 デフォルトキーとは異なるパーティションキーシードを設定したい場合は、long または timestamp 列をパーティショニング時刻として指定できます。デフォルトの時刻列として、add_time フィルターで upload_time を使用します。

8. データストレージの Timezone を選択します。

9. Schedule の下で、このクエリを実行するタイミングと頻度を選択できます。

1. Off を選択します。
2. Scheduling Timezone を選択します。
3. Create & Run Now を選択します。

1. On を選択します。
2. Schedule を選択します。UI では、@hourly、@daily、@monthly、またはカスタム cron の 4 つのオプションが提供されます。
3. Delay Transfer を選択して、実行時間の遅延を追加することもできます。
4. Scheduling Timezone を選択します。
5. Create & Run Now を選択します。

転送が実行された後、Data Workbench > Databases で転送の結果を確認できます。

GCSからデータをインポートするサンプルワークフローについては、Treasure Boxesを参照してください。

コネクタを設定する前に、最新のTD Toolbeltをインストールしてください。

Data Connectorは「boolean」、「long」、「double」、「string」、「timestamp」タイプの解析をサポートしています。

seed.yml をJSON keyfileで準備します。バケット名とターゲットファイル名（または複数ファイルの場合はプレフィックス）も指定する必要があります。

Service Account認証の場合

in:
  type: gcs_v2
  bucket: sample_bucket
  path_prefix: path/to/sample_file  # GCSバケット上の*.csvまたは*.tsvファイルへのパス
  path_match_pattern: "*.csv.gz$"
  auth_method: json_key
  json_keyfile:
    content: |
      {
        "private_key_id": "1234567890",
        "private_key": "-----BEGIN PRIVATE KEY-----\nABCDEF",
        "client_id": "...",
        "client_email": "...",
        "type": "service_account"
      }
out:
  mode: append

Workload Identity Federation認証の場合

in:
  type: gcs_v2
  bucket: sample_bucket
  path_prefix: path/to/sample_file  # GCSバケット上の*.csvまたは*.tsvファイルへのパス
  path_match_pattern: "*.csv.gz$"
  auth_method: wif
  adc_keyfile:
    content: |
      {
        ..........
      }
out:
  mode: append

Google Cloud Storage用のData Connectorは、指定されたプレフィックスにマッチするすべてのファイルをインポートします（例：path_prefix: path/to/sample_ –> path/to/sample_201501.csv.gz、path/to/sample_201502.csv.gz、…、path/to/sample_201505.csv.gz）

connector:guessを使用します。このコマンドは自動的にターゲットファイルを読み取り、ファイル形式をインテリジェントに推測します。

td connector:guess seed.yml -o load.yml

load.ymlを開くと、ファイル形式、エンコーディング、カラム名、型を含む推測されたファイル形式定義が表示されます。

in:
  type: gcs_v2
  bucket: sample_bucket
  path_prefix: path/to/sample_file
  auth_method: json_key
  json_keyfile:
    content: |
      {
        "private_key_id": "1234567890",
        "private_key": "-----BEGIN PRIVATE KEY-----\nABCDEF",
        "client_id": "...",
        "client_email": "...",
        "type": "service_account"
      }
  decoders:
  - {type: gzip}
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: ''
    skip_header_lines: 1
    columns:
    - name: id
      type: long
    - name: company
      type: string
    - name: customer
      type: string
    - name: created_at
      type: timestamp
      format: '%Y-%m-%d %H:%M:%S'
out:
  mode: append

previewコマンドを使用して、システムがファイルをどのように解析するかをプレビューできます。

td connector:preview load.yml

guessコマンドは、ソースデータファイルに3行以上、2列以上が必要です。これは、ソースデータのサンプル行を使用してカラム定義を推測するためです。

カラム名やカラムタイプが予期せず検出された場合は、load.ymlを直接変更して再度プレビューしてください。

previewコマンドは、指定されたバケットから1つのファイルをダウンロードし、そのファイルの結果を表示します。これにより、previewコマンドとissueコマンドの結果に違いが生じる可能性があります。

ロードジョブを送信します。データのサイズによっては、数時間かかる場合があります。ユーザーは、データが保存されるデータベースとテーブルを指定する必要があります。

Treasure Dataのストレージは時間で分割されているため、--time-columnオプションを指定することを推奨します。オプションが指定されない場合、Data Connectorは最初のlongまたはtimestampカラムをパーティショニング時間として選択します。--time-columnで指定するカラムのタイプは、longまたはtimestampタイプである必要があります。

データにtimeカラムがない場合は、add_timeフィルターオプションを使用して追加できます。詳細はadd_time filter pluginを参照してください。

td connector:issue load.yml --database td_sample_db --table td_sample_table   --time-column created_at

上記のコマンドは、database(td_sample_db)とtable(td_sample_table)が既に作成されていることを前提としています。データベースまたはテーブルがTDに存在しない場合、このコマンドは成功しないため、データベースとテーブルを手動で作成するか、td connector:issueコマンドで--auto-create-tableオプションを使用してデータベースとテーブルを自動作成してください：

td connector:issue load.yml --database td_sample_db --table td_sample_table --time-column created_at --auto-create-table

Data Connectorはサーバー側でレコードをソートしません。時間ベースのパーティショニングを効果的に使用するには、事前にファイル内のレコードをソートしてください。

timeというフィールドがある場合は、--time-columnオプションを指定する必要はありません。

td connector:issue load.yml --database td_sample_db --table td_sample_table

インクリメンタルなGoogle Cloud Storageファイルインポートのために、定期的なData Connector実行をスケジュール設定できます。TDはスケジューラを慎重に管理し、高可用性を確保します。この機能を使用することで、ローカルデータセンターでcronデーモンを使用する必要がなくなります。

スケジュールされたインポートの場合、Google Cloud Storage用のData Connectorは、指定されたプレフィックスにマッチするすべてのファイル（例：path_prefix: path/to/sample_ –> path/to/sample_201501.csv.gz、path/to/sample_201502.csv.gz、…、path/to/sample_201505.csv.gz）を最初にインポートし、次回の実行のために最後のパス（path/to/sample_201505.csv.gz）を記憶します。

2回目以降の実行では、アルファベット順（辞書順）で最後のパスの後に来るファイルのみをインポートします（path/to/sample_201506.csv.gz、…）。

td connector:createコマンドを使用して新しいスケジュールを作成できます。スケジュールの名前、cronスタイルのスケジュール、データが保存されるデータベースとテーブル、およびData Connector設定ファイルが必要です。

td connector:create daily_import "10 0 * * *" td_sample_db td_sample_table load.yml

Treasure Dataのストレージは時間で分割されているため、--time-columnオプションを指定することを推奨します。

td connector:create daily_import "10 0 * * *" td_sample_db td_sample_table load.yml --time-column created_at

cronパラメータは、@hourly、@daily、@monthlyの3つの特別なオプションも受け付けます。デフォルトでは、スケジュールはUTCタイムゾーンで設定されます。-tまたは--timezoneオプションを使用して、タイムゾーンでスケジュールを設定できます。--timezoneオプションは、'Asia/Tokyo'、'America/Los_Angeles'などの拡張タイムゾーン形式のみをサポートします。PST、CSTなどのタイムゾーンの略語は*サポートされておらず*、予期しないスケジュールにつながる可能性があります。

td connector:listコマンドを実行することで、現在スケジュールされているエントリのリストを確認できます。

$ td connector:list