Google 提供のバッチテンプレート

Google はオープンソースの Cloud Dataflow テンプレートを提供しています。テンプレートに関する一般的な情報については、概要ページをご覧ください。Google が提供するすべてのテンプレートのリストについては、Google 提供のテンプレートの概要ページをご覧ください。

このページでは、以下のバッチテンプレートについて説明します。

Cloud Bigtable to Cloud Storage Avro
Cloud Bigtable to Cloud Storage SequenceFile
Cloud Datastore to Cloud Storage Text
Cloud Spanner to Cloud Storage Avro
Cloud Spanner to Cloud Storage Text
Cloud Storage Avro to Cloud Bigtable
Cloud Storage Avro to Cloud Spanner
Cloud Storage SequenceFile to Cloud Bigtable
Cloud Storage Text to BigQuery
Cloud Storage Text to Cloud Datastore
Cloud Storage Text to Cloud Pub/Sub (Batch)
Cloud Storage Text to Cloud Spanner
Java Database Connectivity (JDBC) to BigQuery

Cloud Bigtable to Cloud Storage Avro

Cloud Bigtable to Cloud Storage Avro テンプレートは、Cloud Bigtable テーブルからデータを読み取り、Avro 形式で Cloud Storage バケットに書き込むパイプラインです。このテンプレートは、Cloud Bigtable から Cloud Storage にデータを移動する場合に使用できます。

このパイプラインの要件:

Cloud Bigtable テーブルが存在すること。
パイプラインを実行する前に、出力 Cloud Storage バケットを作成済みであること。

テンプレートのパラメータ

パラメータ	説明
`bigtableProjectId`	データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID。
`bigtableInstanceId`	テーブルが含まれる Cloud Bigtable インスタンスの ID。
`bigtableTableId`	エクスポート対象の Cloud Bigtable テーブルの ID。
`outputDirectory`	データの書き込み先の Cloud Storage のパス。例: `gs://mybucket/somefolder`
`filenamePrefix`	Avro ファイル名の接頭辞。例: `output-`

Cloud Bigtable to Cloud Storage Avro file テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューからを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/

REST API の使用で記述されているように、このサンプルリクエストを使用します。このリクエストには承認が必要です。書き込み権限がある場合は tempLocation を指定してください。このサンプルの次の値は置き換える必要があります。

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[OUTPUT_DIRECTORY] は、書き込み先の Cloud Storage パスに置き換えます。例: gs://mybucket/somefolder
[FILENAME_PREFIX] は、Avro ファイル名の接頭辞に置き換えます。例: output-

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/ \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],outputDirectory=[OUTPUT_DIRECTORY],filenamePrefix=[FILENAME_PREFIX]

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/

REST API リクエストでこのテンプレートを実行するには、プロジェクト ID を指定して HTTP POST リクエストを送信します。このリクエストには承認が必要です。

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[OUTPUT_DIRECTORY] は、書き込み先の Cloud Storage パスに置き換えます。例: gs://mybucket/somefolder
[FILENAME_PREFIX] は、Avro ファイル名の接頭辞に置き換えます。例: output-

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "outputDirectory": "[OUTPUT_DIRECTORY]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.bigtable;

import com.google.bigtable.v2.Cell;
import com.google.bigtable.v2.Column;
import com.google.bigtable.v2.Family;
import com.google.bigtable.v2.Row;
import com.google.cloud.teleport.util.DualInputNestedValueProvider;
import com.google.cloud.teleport.util.DualInputNestedValueProvider.TranslatorInput;
import com.google.protobuf.ByteOutput;
import com.google.protobuf.ByteString;
import com.google.protobuf.UnsafeByteOperations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.SimpleFunction;

/**
 * Dataflow pipeline that exports data from a Cloud Bigtable table to Avro files in GCS. Currently,
 * filtering on Cloud Bigtable table is not supported.
 */
public class BigtableToAvro {

  /** Options for the export pipeline. */
  public interface Options extends PipelineOptions {
    @Description("The project that contains the table to export.")
    ValueProvider<String> getBigtableProjectId();

    @SuppressWarnings("unused")
    void setBigtableProjectId(ValueProvider<String> projectId);

    @Description("The Bigtable instance id that contains the table to export.")
    ValueProvider<String> getBigtableInstanceId();

    @SuppressWarnings("unused")
    void setBigtableInstanceId(ValueProvider<String> instanceId);

    @Description("The Bigtable table id to export.")
    ValueProvider<String> getBigtableTableId();

    @SuppressWarnings("unused")
    void setBigtableTableId(ValueProvider<String> tableId);

    @Description("The output location to write to (e.g. gs://mybucket/somefolder/)")
    ValueProvider<String> getOutputDirectory();

    @SuppressWarnings("unused")
    void setOutputDirectory(ValueProvider<String> outputDirectory);

    @Description("The prefix for each exported file in outputDirectory")
    ValueProvider<String> getFilenamePrefix();

    @SuppressWarnings("unused")
    void setFilenamePrefix(ValueProvider<String> filenamePrefix);
  }

  /**
   * Runs a pipeline to export data from a Cloud Bigtable table to Avro files in GCS.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    PipelineResult result = run(options);

    // Wait for pipeline to finish only if it is not constructing a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }

  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(options);

    BigtableIO.Read read =
        BigtableIO.read()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

    // Do not validate input fields if it is running as a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() != null) {
      read = read.withoutValidation();
    }

    ValueProvider<String> filePathPrefix = DualInputNestedValueProvider.of(
        options.getOutputDirectory(),
        options.getFilenamePrefix(),
        new SerializableFunction<TranslatorInput<String, String>, String>() {
          @Override
          public String apply(TranslatorInput<String, String> input) {
            return new StringBuilder(input.getX()).append(input.getY()).toString();
          }
        });

    pipeline
        .apply("Read from Bigtable", read)
        .apply("Transform to Avro", MapElements.via(new BigtableToAvroFn()))
        .apply(
            "Write to Avro in GCS",
            AvroIO.write(BigtableRow.class).to(filePathPrefix).withSuffix(".avro"));

    return pipeline.run();
  }

  /** Translates Bigtable {@link Row} to Avro {@link BigtableRow}. */
  static class BigtableToAvroFn extends SimpleFunction<Row, BigtableRow> {
    @Override
    public BigtableRow apply(Row row) {
      ByteBuffer key = ByteBuffer.wrap(toByteArray(row.getKey()));
      List<BigtableCell> cells = new ArrayList<>();
      for (Family family : row.getFamiliesList()) {
        String familyName = family.getName();
        for (Column column : family.getColumnsList()) {
          ByteBuffer qualifier = ByteBuffer.wrap(toByteArray(column.getQualifier()));
          for (Cell cell : column.getCellsList()) {
            long timestamp = cell.getTimestampMicros();
            ByteBuffer value = ByteBuffer.wrap(toByteArray(cell.getValue()));
            cells.add(new BigtableCell(familyName, qualifier, timestamp, value));
          }
        }
      }
      return new BigtableRow(key, cells);
    }
  }

  /**
   * Extracts the byte array from the given {@link ByteString} without copy.
   *
   * @param byteString A {@link ByteString} from which to extract the array.
   * @return an array of byte.
   */
  private static byte[] toByteArray(final ByteString byteString) {
    try {
      ZeroCopyByteOutput byteOutput = new ZeroCopyByteOutput();
      UnsafeByteOperations.unsafeWriteTo(byteString, byteOutput);
      return byteOutput.bytes;
    } catch (IOException e) {
      return byteString.toByteArray();
    }
  }

  private static final class ZeroCopyByteOutput extends ByteOutput {
    private byte[] bytes;

    @Override
    public void writeLazy(byte[] value, int offset, int length) {
      if (offset != 0 || length != value.length) {
        throw new UnsupportedOperationException();
      }
      bytes = value;
    }

    @Override
    public void write(byte value) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void write(byte[] value, int offset, int length) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void write(ByteBuffer value) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void writeLazy(ByteBuffer value) {
      throw new UnsupportedOperationException();
    }
  }
}

Python

Python ソースコードは利用できません。

Cloud Bigtable to Cloud Storage SequenceFile

Cloud Bigtable to Cloud Storage SequenceFile テンプレートは、Cloud Bigtable テーブルからデータを読み取り、SequenceFile 形式で Cloud Storage バケットに書き込むパイプラインです。このテンプレートは、Cloud Bigtable から Cloud Storage にデータをコピーする場合に使用できます。

このパイプラインの要件:

Cloud Bigtable テーブルが存在すること。
パイプラインを実行する前に、出力 Cloud Storage バケットを作成済みであること。

テンプレートのパラメータ

パラメータ	説明
`bigtableProject`	データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID。
`bigtableInstanceId`	テーブルが含まれる Cloud Bigtable インスタンスの ID。
`bigtableTableId`	エクスポート対象の Cloud Bigtable テーブルの ID。
`bigtableAppProfileId`	エクスポートに使用される Cloud Bigtable アプリケーションプロファイルの ID。アプリプロファイルを指定しない場合、Cloud Bigtable ではインスタンスのデフォルトのアプリプロファイルが使用されます。
`destinationPath`	データの書き込み先の Cloud Storage のパス。例: `gs://mybucket/somefolder`
`filenamePrefix`	SequenceFile ファイル名の接頭辞。例: `output-`

Cloud Bigtable to Cloud Storage SequenceFile テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューからを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[APPLICATION_PROFILE_ID] は、エクスポートに使用する Cloud Bigtable アプリケーションプロファイルの ID に置き換えます。
[DESTINATION_PATH] は、書き込み先の Cloud Storage パスに置き換えます。例: gs://mybucket/somefolder
[FILENAME_PREFIX] は、SequenceFile ファイル名の接頭辞に置き換えます。例: output-

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/ \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],destinationPath=[DESTINATION_PATH],filenamePrefix=[FILENAME_PREFIX]

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[APPLICATION_PROFILE_ID] は、エクスポートに使用する Cloud Bigtable アプリケーションプロファイルの ID に置き換えます。
[DESTINATION_PATH] は、書き込み先の Cloud Storage パスに置き換えます。例: gs://mybucket/somefolder
[FILENAME_PREFIX] は、SequenceFile ファイル名の接頭辞に置き換えます。例: output-

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "destinationPath": "[DESTINATION_PATH]",
       "filenamePrefix": "[FILENAME_PREFIX]",
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java: SDK 2.x

このテンプレートのソースコードは、GitHub の GoogleCloudPlatform/cloud-bigtable-client リポジトリにあります。

Python

Python ソースコードは利用できません。

Cloud Datastore to Cloud Storage Text

Cloud Datastore to Cloud Storage Text テンプレートは、Cloud Datastore エンティティを読み取り、Cloud Storage にテキストファイルとして書き込むバッチパイプラインです。各エンティティを JSON 文字列として扱う関数を使用できます。このような関数を使用しない場合、出力ファイルの各行はシリアル化された JSON エンティティとなります。

このパイプラインの要件:

パイプラインを実行する前に、Cloud Datastore をプロジェクトで設定する必要があります。

テンプレートのパラメータ

パラメータ	説明
`datastoreReadGqlQuery`	取得するエンティティを指定する GQL クエリ。例: `SELECT * FROM MyKind`
`datastoreReadProjectId`	データ読み取り元の Cloud Datastore インスタンスの GCP プロジェクト ID。
`datastoreReadNamespace`	要求されたエンティティの名前空間。デフォルトの名前空間を使用するには、このパラメータを空白のままにします。
`javascriptTextTransformGcsPath`	すべての JavaScript コードを含む Cloud Storage パス。例: `gs://mybucket/mytransforms/*.js`関数を提供しない場合には、このパラメータを空白のままにします。
`javascriptTextTransformFunctionName`	呼び出される JavaScript 関数の名前。たとえば、JavaScript 関数が `function myTransform(inJson) { ...dostuff...}` の場合は、関数名は `myTransform` です。関数を提供しない場合は、このパラメータを空白のままにします。
`textWritePrefix`	データを書き込む Cloud Storage パスの接頭辞。例: `gs://mybucket/somefolder/`

Cloud Datastore to Cloud Storage Text テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Datastore から Cloud Storage へのテキストテンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。
YOUR_DATASTORE_KIND を Datastore エンティティのタイプに置き換えます。
YOUR_DATASTORE_NAMESPACE を Datastore エンティティの名前空間に置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_GCS_Text \
    --parameters \
datastoreReadGqlQuery="SELECT * FROM YOUR_DATASTORE_KIND",\
datastoreReadProjectId=YOUR_PROJECT_ID,\
datastoreReadNamespace=YOUR_DATASTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。
YOUR_DATASTORE_KIND を Datastore エンティティのタイプに置き換えます。
YOUR_DATASTORE_NAMESPACE を Datastore エンティティの名前空間に置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Datastore_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "datastoreReadGqlQuery": "SELECT * FROM YOUR_DATASTORE_KIND"
       "datastoreReadProjectId": "YOUR_PROJECT_ID",
       "datastoreReadNamespace": "YOUR_DATASTORE_NAMESPACE",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreReadOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.ReadJsonEntities;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;

/**
 * Dataflow template which copies Datastore Entities to a Text sink.
 * Text is encoded using JSON encoded entity in the v1/Entity rest format:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class DatastoreToText {

  /**
   * Custom PipelineOptions.
   */
  public interface DatastoreToTextOptions extends
      PipelineOptions,
      DatastoreReadOptions,
      JavascriptTextTransformerOptions,
      FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Entities from Datastore, passes in the JSON encoded Entities
   * to a Javascript UDF, and writes the JSON to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    DatastoreToTextOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(DatastoreToTextOptions.class);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(ReadJsonEntities.newBuilder()
            .setGqlQuery(options.getDatastoreReadGqlQuery())
            .setProjectId(options.getDatastoreReadProjectId())
            .setNamespace(options.getDatastoreReadNamespace())
            .build())
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(TextIO.write()
            .to(options.getTextWritePrefix())
            .withSuffix(".json"));

    pipeline.run();
  }
}

Python

Python ソースコードは利用できません。

Cloud Spanner to Cloud Storage Avro

Cloud Spanner to Cloud Storage テンプレートは、Cloud Spanner データベース全体を Avro 形式で Cloud Storage にエクスポートするバッチパイプラインです。Cloud Spanner データベースをエクスポートすると、選択したバケット内にフォルダが作成されます。フォルダには以下が含まれています。

spanner-export.json ファイル
エクスポートしたデータベースの角テーブルの TableName-manifest.json ファイル。
1 つ以上の TableName.avro-#####-of-##### ファイル。

たとえば、Singers と Albums の 2 つのテーブルを持つデータベースをエクスポートして、次のファイルセットを作成します。

Albums-manifest.json
Albums.avro-00000-of-00002
Albums.avro-00001-of-00002
Singers-manifest.json
Singers.avro-00000-of-00003
Singers.avro-00001-of-00003
Singers.avro-00002-of-00003
spanner-export.json

このパイプラインの要件:

Cloud Spanner データベースが存在すること。
出力先の Cloud Storage バケットが存在すること。
Cloud Dataflow ジョブの実行に必要な Cloud IAM の役割に加え、Cloud Spanner データの読み取りと Cloud Storage バケットへの書き込みに対応できる Cloud IAM の役割も必要です。

テンプレートのパラメータ

パラメータ	説明
`instanceId`	エクスポートする Cloud Spanner データベースのインスタンス ID。
`databaseId`	エクスポートする Cloud Spanner データベースのデータベース ID。
`outputDir`	Avro ファイルのエクスポート先にする Cloud Storage パス。エクスポートジョブによって、このパスの下にディレクトリが新規作成されます。ここに、エクスポートされたファイルが格納されます。

テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから the Cloud Spanner to Cloud Storage Avro template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

REST API の使用で記述されているように、このサンプルリクエストを使用します。このリクエストには承認が必要です。書き込み権限がある場合は tempLocation を指定する必要があります。このサンプルの次の値は置き換える必要があります。

[DATAFLOW_REGION] を、Cloud Dataflow ジョブを実行するリージョンで置き換えます（たとえば、us-central1）。
[YOUR_INSTANCE_ID] は、Cloud Spanner インスタンスの ID に置き換えます。
[YOUR_DATABASE_ID] は、Cloud Spanner データベースの ID に置き換えます。
[YOUR_GCS_DIRECTORY] を、Avro ファイルがエクスポートされるがインポートされました
[JOB_NAME] を任意のジョブ名に置き換えます。
- ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
- GCP Console の Cloud Spanner 部分に表示されるジョブ名は、エクスポート cloud-spanner--[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] の形式になっている必要があります。

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],outputDir=[YOUR_GCS_DIRECTORY]

API

REST API から実行する

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[DATAFLOW_REGION] を、Cloud Dataflow ジョブを実行するリージョンで置き換えます（たとえば、us-central1）。
[YOUR_INSTANCE_ID] は、Cloud Spanner インスタンスの ID に置き換えます。
[YOUR_DATABASE_ID] は、Cloud Spanner データベースの ID に置き換えます。
[YOUR_GCS_DIRECTORY] を、Avro ファイルがエクスポートされるがインポートされました
[JOB_NAME] を任意のジョブ名に置き換えます。
- ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
- GCP Console の Cloud Spanner 部分に表示されるジョブ名は、エクスポート cloud-spanner--[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] の形式になっている必要があります。

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/Cloud_Spanner_to_GCS_Avro
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "outputDir": "gs://[YOUR_GCS_DIRECTORY]"
   }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.spanner;

import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;

/** Dataflow template that exports a Cloud Spanner database to Avro files in GCS. */
public class ExportPipeline {

  /** Options for Export pipeline. */
  public interface ExportPipelineOptions extends PipelineOptions {
    @Description("Instance ID for Cloud Spanner")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @Description("Database ID for Cloud Spanner")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @Description("Output directory in GCS")
    ValueProvider<String> getOutputDir();

    void setOutputDir(ValueProvider<String> value);

    @Description("Test dataflow job identifier for Beam Direct Runner")
    @Default.String(value = "")
    ValueProvider<String> getTestJobId();

    void setTestJobId(ValueProvider<String> jobId);

    @Description("Cloud Spanner API endpoint")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @Description("If true, wait for job finish")
    @Default.Boolean(true)
    boolean getWaitUntilFinish();

    void setWaitUntilFinish(boolean value);
  }

  /**
   * Runs a pipeline to export a Cloud Spanner database to Avro files.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {

    ExportPipelineOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(ExportPipelineOptions.class);

    Pipeline p = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());
    p.begin()
        .apply(
            "Run Export",
            new ExportTransform(spannerConfig, options.getOutputDir(), options.getTestJobId()));
    PipelineResult result = p.run();
    if (options.getWaitUntilFinish() &&
        /* Only if template location is null, there is a dataflow job to wait for. Else it's
         * template generation which doesn't start a dataflow job.
         */
        options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }
}

Python

Python ソースコードは利用できません。

Cloud Spanner to Cloud Storage Text

Cloud Spanner to Cloud Storage Text テンプレートは、データを Cloud Spanner テーブルから読み込み、必要に応じて指定した JavaScript ユーザー定義関数（UDF）を介して変換し、CSV テキストファイルとして Cloud Storage に書き込むバッチパイプラインです。

このパイプラインの要件:

パイプラインを実行する前に、入力された Cloud Spanner テーブルが存在する必要があります。

テンプレートのパラメータ

パラメータ	説明
`spannerProjectId`	データを読み取る Cloud Spanner データベースの GCP プロジェクト ID。
`spannerDatabaseId`	要求されたテーブルのデータベース。
`spannerInstanceId`	要求されたテーブルのインスタンス。
`spannerTable`	エクスポートするテーブル。
`textWritePrefix`	出力テキストファイルを書き込む出力ディレクトリ。末尾に / を付加してください。例: `gs://mybucket/somefolder/`。
`javascriptTextTransformGcsPath`	（省略可）すべての JavaScript コードを含む Cloud Storage パス。例: `gs://mybucket/mytransforms/*.js` 関数を提供しない場合は、このパラメータを空白のままにします。
`javascriptTextTransformFunctionName`	（省略可）呼び出される JavaScript 関数の名前。たとえば、JavaScript 関数が `function myTransform(inJson) { ...dostuff...}` であれば、関数名は `myTransform` です。関数を提供しない場合は、このパラメータを空白のままにします。

Cloud Spanner to Cloud Storage Text テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Spanner to Cloud Storage のテキストテンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_DATABASE_ID を Cloud Spanner データベース ID に置き換えます。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。
YOUR_INSTANCE_ID を Cloud Spanner インスタンス ID に置き換えます。
YOUR_TABLE_ID を Cloud Spanner のテーブル ID に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードが含まれている .js ファイルへのクラウドストレージパスで置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters \
spannerProjectId=YOUR_PROJECT_ID,\
spannerDatabaseId=YOUR_DATABASE_ID,\
spannerInstanceId=YOUR_INSTANCE_ID,\
spannerTable=YOUR_TABLE_ID,\
textWritePrefix=gs://YOUR_BUCKET_NAME/output/,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_DATABASE_ID を Cloud Spanner データベース ID に置き換えます。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。
YOUR_INSTANCE_ID を Cloud Spanner インスタンス ID に置き換えます。
YOUR_TABLE_ID を Cloud Spanner のテーブル ID に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードが含まれている .js ファイルへのクラウドストレージパスで置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "spannerProjectId": "YOUR_PROJECT_ID",
       "spannerDatabaseId": "YOUR_DATABASE_ID",
       "spannerInstanceId": "YOUR_INSTANCE_ID",
       "spannerTable": "YOUR_TABLE_ID",
       "textWritePrefix": "gs://YOUR_BUCKET_NAME/output/",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.SpannerConverters;
import com.google.cloud.teleport.templates.common.SpannerConverters.SpannerReadOptions;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemWriteOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which copies a Spanner table to a Text sink. It exports a Spanner table using
 * <a href="https://cloud.google.com/spanner/docs/reads#read_data_in_parallel">Batch API</a>,
 * which creates multiple workers in parallel for better performance. The result is written
 * to a CSV file in Google Cloud Storage. The table schema file is saved in json format along with
 * the exported table.
 *
 * <p>Schema file sample: { "id":"INT64", "name":"STRING(MAX)" }
 *
 * <p>A sample run:
 *
 * <pre>
 * mvn compile exec:java \
 *   -Dexec.mainClass=com.google.cloud.teleport.templates.SpannerToText \
 *   -Dexec.args="--runner=DataflowRunner \
 *                --spannerProjectId=projectId \
 *                --gcpTempLocation=gs://gsTmpLocation \
 *                --spannerInstanceId=instanceId \
 *                --spannerDatabaseId=databaseId \
 *                --spannerTable=table_name \
 *                --textWritePrefix=gcsOutputPath"
 * </pre>
 */
public class SpannerToText {

  private static final Logger LOG = LoggerFactory.getLogger(SpannerToText.class);

  /**
   * Custom PipelineOptions.
   */
  public interface SpannerToTextOptions
      extends PipelineOptions,
          SpannerReadOptions,
          JavascriptTextTransformerOptions,
          FilesystemWriteOptions {}

  /**
   * Runs a pipeline which reads in Records from Spanner, passes in the CSV records to a Javascript
   * UDF, and writes the CSV to TextIO sink.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    LOG.info("Starting pipeline setup");
    PipelineOptionsFactory.register(SpannerToTextOptions.class);
    SpannerToTextOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(SpannerToTextOptions.class);

    FileSystems.setDefaultPipelineOptions(options);
    Pipeline pipeline = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withProjectId(options.getSpannerProjectId())
            .withInstanceId(options.getSpannerInstanceId())
            .withDatabaseId(options.getSpannerDatabaseId());

    PTransform<PBegin, PCollection<ReadOperation>> spannerExport =
        SpannerConverters.ExportTransformFactory.create(
            options.getSpannerTable(), spannerConfig, options.getTextWritePrefix());

    PCollection<String> csv =
        pipeline
            .apply("Create export", spannerExport)
            // We need to use SpannerIO.readAll() instead of SpannerIO.read()
            // because ValueProvider parameters such as table name required for SpannerIO.read()
            // can be read only inside DoFn but SpannerIO.read() is of type
            // PTransform<PBegin, Struct>, which prevents prepending it with DoFn that reads these
            // parameters at the pipeline execution time.
            .apply("Read all records", SpannerIO.readAll().withSpannerConfig(spannerConfig))
            .apply(
                "Struct To Csv",
                MapElements.into(TypeDescriptors.strings())
                    .via(struct -> (new SpannerConverters.StructCsvPrinter()).print(struct)));

    if (options.getJavascriptTextTransformGcsPath().isAccessible()) {
      // The UDF function takes a CSV row as an input and produces a transformed CSV row
      csv =
          csv.apply(
              "JavascriptUDF",
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }
    csv.apply(
        "Write to storage", TextIO.write().to(options.getTextWritePrefix()).withSuffix(".csv"));

    pipeline.run();
    LOG.info("Completed pipeline setup");
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage Avro to Cloud Bigtable

Cloud Storage Avro to Cloud Bigtable テンプレートは、Cloud Storage バケットの Avro ファイルからデータを読み取り、Cloud Bigtable テーブルに書き込むパイプラインです。このテンプレートは、Cloud Storage から Cloud Bigtable にデータをコピーする場合に使用できます。

このパイプラインの要件:

Cloud Bigtable テーブルが存在し、Avro ファイルにエクスポートしたものと同じ列ファミリがこのテーブルにあること。
入力 Avro ファイルが、パイプラインを実行する前に Cloud Storage バケット内に存在すること。
Cloud Bigtable が入力 Avro ファイルに特定のスキーマを想定していること。

テンプレートのパラメータ

パラメータ	説明
`bigtableProjectId`	データを書き込む Cloud Bigtable インスタンスの GCP プロジェクトの ID。
`bigtableInstanceId`	テーブルが含まれる Cloud Bigtable インスタンスの ID。
`bigtableTableId`	インポート対象の Cloud Bigtable テーブルの ID。
`inputFilePattern`	データが存在する Cloud Storage パスのパターン。例: `gs://mybucket/somefolder/prefix*`

Cloud Storage Avro file to Cloud Bigtable テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Spanner to Cloud Storage のテキストテンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[INPUT_FILE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters bigtableProjectId=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],inputFilePattern=[INPUT_FILE_PATTERN]

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[INPUT_FILE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProjectId": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "inputFilePattern": "[INPUT_FILE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.bigtable;

import com.google.bigtable.v2.Mutation;
import com.google.bigtable.v2.Mutation.SetCell;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import java.nio.ByteBuffer;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;

/**
 * Dataflow pipeline that imports data from Avro files in GCS to a Cloud Bigtable table. The Cloud
 * Bigtable table must be created before running the pipeline and must have a compatible table
 * schema. For example, if {@link BigtableCell} from the Avro files has a 'family' of "f1", the
 * Bigtable table should have a column family of "f1".
 */
final class AvroToBigtable {

  /** Options for the import pipeline. */
  public interface Options extends PipelineOptions {
    @Description("The project that contains the table to import into.")
    ValueProvider<String> getBigtableProjectId();

    @SuppressWarnings("unused")
    void setBigtableProjectId(ValueProvider<String> projectId);

    @Description("The Bigtable instance id that contains the table to import into.")
    ValueProvider<String> getBigtableInstanceId();

    @SuppressWarnings("unused")
    void setBigtableInstanceId(ValueProvider<String> instanceId);

    @Description("The Bigtable table id to import into.")
    ValueProvider<String> getBigtableTableId();

    @SuppressWarnings("unused")
    void setBigtableTableId(ValueProvider<String> tableId);

    @Description(
        "The input file patterm to read from. (e.g. gs://mybucket/somefolder/table1*.avro)")
    ValueProvider<String> getInputFilePattern();

    @SuppressWarnings("unused")
    void setInputFilePattern(ValueProvider<String> inputFilePattern);
  }

  /**
   * Runs a pipeline to import Avro files in GCS to a Cloud Bigtable table.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    PipelineResult result = run(options);

    // Wait for pipeline to finish only if it is not constructing a template.
    if (options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }

  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(options);

    BigtableIO.Write write =
        BigtableIO.write()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

    pipeline
        .apply("Read from Avro", AvroIO.read(BigtableRow.class).from(options.getInputFilePattern()))
        .apply("Transform to Bigtable", MapElements.via(new AvroToBigtableFn()))
        .apply("Write to Bigtable", write);

    return pipeline.run();
  }

  /**
   * Translates {@link BigtableRow} to {@link Mutation}s along with a row key. The mutations are
   * {@link SetCell}s that set the value for specified cells with family name, column qualifier and
   * timestamp.
   */
  static class AvroToBigtableFn
      extends SimpleFunction<BigtableRow, KV<ByteString, Iterable<Mutation>>> {
    @Override
    public KV<ByteString, Iterable<Mutation>> apply(BigtableRow row) {
      ByteString key = toByteString(row.getKey());
      // BulkMutation doesn't split rows. Currently, if a single row contains more than 100,000
      // mutations, the service will fail the request.
      ImmutableList.Builder<Mutation> mutations = ImmutableList.builder();
      for (BigtableCell cell : row.getCells()) {
        SetCell setCell =
            SetCell.newBuilder()
                .setFamilyName(cell.getFamily().toString())
                .setColumnQualifier(toByteString(cell.getQualifier()))
                .setTimestampMicros(cell.getTimestamp())
                .setValue(toByteString(cell.getValue()))
                .build();
        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
      }
      return KV.of(key, mutations.build());
    }
  }

  /** Copies the content in {@code byteBuffer} into a {@link ByteString}. */
  private static ByteString toByteString(ByteBuffer byteBuffer) {
    return ByteString.copyFrom(byteBuffer.array());
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage Avro to Cloud Spanner

Cloud Storage Avro files to Cloud Spanner テンプレートは、Cloud Storage から Avro ファイルを読み取り、そのファイルを Cloud Spanner データベースにインポートするバッチパイプラインです。

このパイプラインの要件:

ターゲットの Cloud Spanner データベースが存在し、空である必要があります。
Cloud Storage バケットの読み取り権限と、対象の Cloud Spanner データベースに対する書き込み権限が必要です。
入力された Cloud Storage パスが存在する必要があります。また、インポートするファイルの JSON 記述を含む spanner-export.json ファイルがそこに格納されている必要があります。

テンプレートのパラメータ

パラメータ	説明
`instanceId`	Cloud Spanner データベースのインスタンス ID。
`databaseId`	Cloud Spanner データベースのデータベース ID。
`inputDir`	Avro ファイルのインポート元にする Cloud Storage パス。

テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Avro to Cloud Spanner テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/ GCS_Avro_to_Cloud_Spanner

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[INPUT_FILE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/ GCS_Avro_to_Cloud_Spanner ' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],inputDir=[YOUR_GCS_DIRECTORY]'

API

REST API から実行する

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/ GCS_Avro_to_Cloud_Spanner

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[INPUT_FILE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "inputDir": "gs://[YOUR_GCS_DIRECTORY]"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.spanner;

import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;

/** Avro to Cloud Spanner Import pipeline. */
public class ImportPipeline {

  /** Options for {@link ImportPipeline}. */
  public interface Options extends PipelineOptions {

    @Description("Instance ID to write to Spanner")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @Description("Database ID to write to Spanner")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @Description("Input directory")
    ValueProvider<String> getInputDir();

    void setInputDir(ValueProvider<String> value);

    @Description("Spanner host")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @Description("If true wait until indexes are finished. Useful for testing.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForIndexes();

    void setWaitForIndexes(ValueProvider<Boolean> value);

    @Description("If true, wait for job finish")
    @Default.Boolean(true)
    boolean getWaitUntilFinish();

    void setWaitUntilFinish(boolean value);
  }

  public static void main(String[] args) {

    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    Pipeline p = Pipeline.create(options);

    SpannerConfig spannerConfig =
        SpannerConfig.create()
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());

    p.apply(new ImportTransform(spannerConfig, options.getInputDir(), options.getWaitForIndexes()));

    PipelineResult result = p.run();
    if (options.getWaitUntilFinish() &&
        /* Only if template location is null, there is a dataflow job to wait for. Else it's
         * template generation which doesn't start a dataflow job.
         */
        options.as(DataflowPipelineOptions.class).getTemplateLocation() == null) {
      result.waitUntilFinish();
    }
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage SequenceFile to Cloud Bigtable

Cloud Storage SequenceFile to Cloud Bigtable テンプレートは、Cloud Storage バケットの SequenceFiles からデータを読み取り、Cloud Bigtable テーブルに書き込むパイプラインです。このテンプレートは、Cloud Storage から Cloud Bigtable にデータをコピーする場合に使用できます。

このパイプラインの要件:

Cloud Bigtable テーブルが存在すること。
入力 SequenceFiles が、パイプラインを実行する前に Cloud Storage バケット内に存在していること。
入力 SequenceFiles が Cloud Bigtable または HBase からエクスポートされていること。

テンプレートのパラメータ

パラメータ	説明
`bigtableProject`	データを書き込む Cloud Bigtable インスタンスの GCP プロジェクトの ID。
`bigtableInstanceId`	テーブルが含まれる Cloud Bigtable インスタンスの ID。
`bigtableTableId`	インポート対象の Cloud Bigtable テーブルの ID。
`bigtableAppProfileId`	インポートに使用される Cloud Bigtable アプリケーションプロファイルの ID。アプリプロファイルを指定しない場合、Cloud Bigtable ではインスタンスのデフォルトのアプリプロファイルが使用されます。
`sourcePattern`	データが存在する Cloud Storage パスのパターン。例: `gs://mybucket/somefolder/prefix*`

Cloud Storage SequenceFile to Cloud Bigtable テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Avro to Cloud Spanner テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/ GCS_Avro_to_Cloud_Spanner

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[APPLICATION_PROFILE_ID] は、エクスポートに使用する Cloud Bigtable アプリケーションプロファイルの ID に置き換えます。
[SOURCE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location gs://dataflow-templates/latest/ GCS_Avro_to_Cloud_Spanner  \
    --parameters bigtableProject=[PROJECT_ID],bigtableInstanceId=[INSTANCE_ID],bigtableTableId=[TABLE_ID],bigtableAppProfileId=[APPLICATION_PROFILE_ID],sourcePattern=[SOURCE_PATTERN]

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/ GCS_Avro_to_Cloud_Spanner

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
[PROJECT_ID] は、データを読み取る Cloud Bigtable インスタンスの GCP プロジェクトの ID に置き換えます。
[INSTANCE_ID] は、テーブルが含まれる Cloud Bigtable インスタンスの ID に置き換えます。
[TABLE_ID] は、エクスポート対象の Cloud Bigtable テーブルの ID に置き換えます。
[APPLICATION_PROFILE_ID] は、エクスポートに使用する Cloud Bigtable アプリケーションプロファイルの ID に置き換えます。
[SOURCE_PATTERN] は、データが存在する Cloud Storage パスのパターンに置き換えます。例: gs://mybucket/somefolder/prefix*

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/templates:launch?gcsPath=gs://dataflow-templates/latest/ GCS_Avro_to_Cloud_Spanner 
{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "bigtableProject": "[PROJECT_ID]",
       "bigtableInstanceId": "[INSTANCE_ID]",
       "bigtableTableId": "[TABLE_ID]",
       "bigtableAppProfileId": "[APPLICATION_PROFILE_ID]",
       "sourcePattern": "[SOURCE_PATTERN]",
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java: SDK 2.x

このテンプレートのソースコードは、GitHub の GoogleCloudPlatform/cloud-bigtable-client リポジトリにあります。

Python

Python ソースコードは利用できません。

Cloud Storage Text to BigQuery

Cloud Storage Text to BigQuery テンプレートは、Cloud Storage に格納されているテキストファイルを読み取り、ユーザーが提供する JavaScript ユーザー定義関数（UDF）を使用してファイルを変換し、その結果を BigQuery に出力するバッチパイプラインです。

重要: 既存の BigQuery テーブルを再利用すると、テーブルが上書きされます。

このパイプラインの要件:

BigQuery スキーマを記述する JSON ファイルを作成します。

BigQuery Schema というタイトルになっているトップレベルの JSON 配列があり、その内容が {"name": "COLUMN_NAME", "type": "DATA_TYPE"} のパターンに従っていることを確認します。例:

{
  "BigQuery Schema": [
    {
      "name": "location",
      "type": "STRING"
    },
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "STRING"
    },
    {
      "name": "color",
      "type": "STRING"
    },
    {
      "name": "coffee",
      "type": "STRING"
    }
  ]
}

JavaScript（.js）ファイルを作成し、このファイル内に、テキスト行の変換ロジックを提供する UDF 関数を含めます。この関数は、JSON 文字列を返さなければならないことに注意してください。
たとえば、次の関数は、CSV ファイルの各行を分割し、値を変換してから JSON 文字列を返します。
```
function transform(line) {
var values = line.split(',');

var obj = new Object();
obj.location = values[0];
obj.name = values[1];
obj.age = values[2];
obj.color = values[3];
obj.coffee = values[4];
var jsonString = JSON.stringify(obj);

return jsonString;
}
```

テンプレートのパラメータ

パラメータ	説明
`javascriptTextTransformFunctionName`	`.js` ファイルから呼び出す関数の名前。
`JSONPath`	Cloud Storage に格納された BigQuery スキーマを定義する JSON ファイルへの `gs://` パス。例: `gs://path/to/my/schema.json`
`javascriptTextTransformGcsPath`	UDF を定義する JavaScript ファイルへの `gs://` パス。例: `gs://path/to/my/javascript_function.js`
`inputFilePattern`	Cloud Storage 中の処理するテキストの `gs://` パス。例: `gs://path/to/my/text/data.txt`
`outputTable`	処理されたデータを格納するために作成する BigQuery テーブル名。既存の BigQuery テーブルを再利用すると、テーブルが上書きされます。例: `my-project-name:my-dataset.my-table`
`bigQueryLoadingTemporaryDirectory`	BigQuery 読み込みプロセスで使用する一時ディレクトリ。例: `gs://my-bucket/my-files/temp_dir`

Cloud Storage Text to BigQuery テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Text to BigQuery テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_JAVASCRIPT_FUNCTION を UDF の名前に置き換えます。
PATH_TO_BIGQUERY_SCHEMA_JSON を、スキーマ定義を含む JSON ファイルの Cloud Storage パスに置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。
PATH_TO_YOUR_TEXT_DATA を、テキストデータセットの Cloud Storage パスに置き換えます。
BIGQUERY_TABLE を BigQuery テーブル名に置き換えます。
PATH_TO_TEMP_DIR_ON_GCS を、一時ディレクトリへの Cloud Storage パスに置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_BigQuery \
    --parameters \
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
inputFilePattern=PATH_TO_YOUR_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_JAVASCRIPT_FUNCTION を UDF の名前に置き換えます。
PATH_TO_BIGQUERY_SCHEMA_JSON を、スキーマ定義を含む JSON ファイルの Cloud Storage パスに置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。
PATH_TO_YOUR_TEXT_DATA を、テキストデータセットの Cloud Storage パスに置き換えます。
BIGQUERY_TABLE を BigQuery テーブル名に置き換えます。
PATH_TO_TEMP_DIR_ON_GCS を、一時ディレクトリへの Cloud Storage パスに置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_YOUR_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.teleport.templates.common.BigQueryConverters;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Templated pipeline to read text from TextIO, apply a javascript UDF to it, and write it to GCS.
 */
public class TextIOToBigQuery {

  /** Options supported by {@link TextIOToBigQuery}. */
  public interface Options extends DataflowPipelineOptions, JavascriptTextTransformerOptions {
    @Description("The GCS location of the text you'd like to process")
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @Description("JSON file with BigQuery Schema description")
    ValueProvider<String> getJSONPath();

    void setJSONPath(ValueProvider<String> value);

    @Description("Output topic to write to")
    ValueProvider<String> getOutputTable();

    void setOutputTable(ValueProvider<String> value);

    @Description("GCS path to javascript fn for transforming output")
    ValueProvider<String> getJavascriptTextTransformGcsPath();

    void setJavascriptTextTransformGcsPath(ValueProvider<String> jsTransformPath);

    @Validation.Required
    @Description("UDF Javascript Function Name")
    ValueProvider<String> getJavascriptTextTransformFunctionName();

    void setJavascriptTextTransformFunctionName(
        ValueProvider<String> javascriptTextTransformFunctionName);

    @Validation.Required
    @Description("Temporary directory for BigQuery loading process")
    ValueProvider<String> getBigQueryLoadingTemporaryDirectory();

    void setBigQueryLoadingTemporaryDirectory(ValueProvider<String> directory);
  }

  private static final Logger LOG = LoggerFactory.getLogger(TextIOToBigQuery.class);

  private static final String BIGQUERY_SCHEMA = "BigQuery Schema";
  private static final String NAME = "name";
  private static final String TYPE = "type";
  private static final String MODE = "mode";

  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply("Read from source", TextIO.read().from(options.getInputFilePattern()))
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(BigQueryConverters.jsonToTableRow())
        .apply(
            "Insert into Bigquery",
            BigQueryIO.writeTableRows()
                .withSchema(
                    NestedValueProvider.of(
                        options.getJSONPath(),
                        new SerializableFunction<String, TableSchema>() {

                          @Override
                          public TableSchema apply(String jsonPath) {

                            TableSchema tableSchema = new TableSchema();
                            List<TableFieldSchema> fields = new ArrayList<>();
                            SchemaParser schemaParser = new SchemaParser();
                            JSONObject jsonSchema;

                            try {

                              jsonSchema = schemaParser.parseSchema(jsonPath);

                              JSONArray bqSchemaJsonArray =
                                  jsonSchema.getJSONArray(BIGQUERY_SCHEMA);

                              for (int i = 0; i < bqSchemaJsonArray.length(); i++) {
                                JSONObject inputField = bqSchemaJsonArray.getJSONObject(i);
                                TableFieldSchema field =
                                    new TableFieldSchema()
                                        .setName(inputField.getString(NAME))
                                        .setType(inputField.getString(TYPE));

                                if (inputField.has(MODE)) {
                                  field.setMode(inputField.getString(MODE));
                                }

                                fields.add(field);
                              }
                              tableSchema.setFields(fields);

                            } catch (Exception e) {
                              throw new RuntimeException(e);
                            }
                            return tableSchema;
                          }
                        }))
                .to(options.getOutputTable())
                .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory()));

    pipeline.run();
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage Text to Cloud Datastore

Cloud Storage Text to Cloud Datastore テンプレートは、Cloud Storage に格納されているテキストファイルを読み取って JSON でエンコードし、それらのエンティティを Cloud Datastore に書き込むバッチパイプラインです。入力テキストファイルの各行は、https://cloud.google.com/datastore/docs/reference/rest/v1/Entity で指定されている JSON 形式である必要があります。

このパイプラインの要件:

データストアが宛先プロジェクトで有効にされている必要があります。

テンプレートのパラメータ

パラメータ	説明
`textReadPattern`	テキストデータファイルの場所を指定する、Cloud Storage ファイルのパスパターン。例: `gs://mybucket/somepath/*.json`
`javascriptTextTransformGcsPath`	すべての JavaScript コードを格納する Cloud Storage パス。例: `gs://mybucket/mytransforms/*.js`関数を提供しない場合には、このパラメータを空白のままにします。
`javascriptTextTransformFunctionName`	呼び出される JavaScript 関数の名前。たとえば、JavaScript 関数が `function myTransform(inJson) { ...dostuff...}` の場合は、関数名は `myTransform` です。関数を提供しない場合は、このパラメータを空白のままにします。
`datastoreWriteProjectId`	Cloud Datastore エンティティを書き込む GCP プロジェクトのプロジェクト ID。
`errorWritePath`	処理中に発生したエラーを書き込むために使用するエラーログ出力ファイル。例: `gs://bucket-name/errors.txt`

Cloud Storage Text to Datastore テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Text to Datastore テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
PATH_TO_INPUT_TEXT_FILES を、Cloud Storage 上の入力ファイルのパターンに置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。
ERROR_FILE_WRITE_PATH を、Cloud Storage 上のエラー・ファイルへの必要なパスに置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Datastore \
    --parameters \
textReadPattern=PATH_TO_INPUT_TEXT_FILES,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=YOUR_JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=YOUR_PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
PATH_TO_INPUT_TEXT_FILES を、Cloud Storage 上の入力ファイルのパターンに置き換えます。
YOUR_JAVASCRIPT_FUNCTION を JavaScript 関数名に置き換えます。
PATH_TO_JAVASCRIPT_UDF_FILE を、JavaScript コードを含む .js ファイルへの Cloud Storage パスに置き換えます。
ERROR_FILE_WRITE_PATH を、Cloud Storage 上のエラー・ファイルへの必要なパスに置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Datastore
{
   "jobName": "JOB_NAME",
   "parameters": {
       "textReadPattern": "PATH_TO_INPUT_TEXT_FILES",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "YOUR_JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "YOUR_PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.cloud.teleport.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.templates.common.TextConverters.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.TupleTag;

/**
 * Dataflow template which reads from a Text Source and writes JSON encoded Entities into Datastore.
 * The Json is expected to be in the format of:
 * https://cloud.google.com/datastore/docs/reference/rest/v1/Entity
 */
public class TextToDatastore {

  /**
   * TextToDatastore Pipeline Options.
   */
  public interface TextToDatastoreOptions extends
      PipelineOptions,
      FilesystemReadOptions,
      JavascriptTextTransformerOptions,
      DatastoreWriteOptions,
      ErrorWriteOptions {}

  /**
   * Runs a pipeline which reads from a Text Source, passes the Text to a Javascript UDF,
   * writes the JSON encoded Entities to a TextIO sink.
   *
   * <p>If your Text Source does not contain JSON encoded Entities, then you'll need to supply a
   * Javascript UDF which transforms your data to be JSON encoded Entities.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    TextToDatastoreOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(TextToDatastoreOptions.class);

    TupleTag<String> errorTag = new TupleTag<String>("errors"){};

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(TextIO.read()
            .from(options.getTextReadPattern()))
        .apply(TransformTextViaJavascript.newBuilder()
            .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
            .setFunctionName(options.getJavascriptTextTransformFunctionName())
            .build())
        .apply(WriteJsonEntities.newBuilder()
            .setProjectId(options.getDatastoreWriteProjectId())
            .setErrorTag(errorTag)
            .build())
        .apply(LogErrors.newBuilder()
            .setErrorWritePath(options.getErrorWritePath())
            .setErrorTag(errorTag)
            .build());

    pipeline.run();
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage Text to Cloud Pub/Sub (Batch)

このテンプレートは、Cloud Storage に保存されたテキストファイルからレコードを読み取り、Cloud Pub/Sub トピックに公開するバッチパイプラインを作成します。このテンプレートを使用すると、JSON レコードを含む改行区切りのファイルや CSV ファイルのレコードを Cloud Pub/Sub トピックに公開し、リアルタイムで処理できます。また、Cloud Pub/Sub でデータを再生することもできます。

このテンプレートでは、個々のレコードにタイムスタンプを設定しません。このため、実行中はイベント時間と公開時間が同じになります。パイプラインの処理が正確なイベント時間に依存している場合は、このパイプラインを使用しないでください。

このパイプラインの要件:

読み込むファイルは、改行区切りの JSON または CSV 形式でなければなりません。ソースファイル内に複数行にわたるレコードがあると、ファイル内の各行がメッセージとして Cloud Pub/Sub に公開されるため、ダウンストリームで問題が発生する可能性があります。
Cloud Pub / Sub トピックは、パイプラインを実行する前に存在している必要があります。

テンプレートのパラメータ

パラメータ	説明
`inputFilePattern`	読み込み元の入力ファイルのパターン。例: `gs://bucket-name/files/*.json`
`outputTopic`	書き込み先の Cloud Pub/Sub 入力トピック。名前は `projects/<project-id>/topics/<topic-name>` の形式にします。

Cloud Storage Text to Cloud Pub/Sub (Batch) テンプレートの実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Text to Cloud Pub/Sub (Batch) テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_TOPIC_NAME を Cloud Pub/Sub のトピック名に置き換えます。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub \
    --parameters \
inputFilePattern=gs://YOUR_BUCKET_NAME/files/*.json,\
outputTopic=projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
YOUR_TOPIC_NAME を Cloud Pub/Sub のトピック名に置き換えます。
YOUR-BUCKET-NAME を Cloud Storage バケットの名前に置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://YOUR_BUCKET_NAME/files/*.json",
       "outputTopic": "projects/YOUR_PROJECT_ID/topics/YOUR_TOPIC_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * The {@code TextToPubsub} pipeline publishes records to
 * Cloud Pub/Sub from a set of files. The pipeline reads each
 * file row-by-row and publishes each record as a string message.
 * At the moment, publishing messages with attributes is unsupported.
 *
 * <p>Example Usage:
 *
 * <pre>
 * {@code mvn compile exec:java \
    -Dexec.mainClass=com.google.cloud.teleport.templates.TextToPubsub \
    -Dexec.args=" \
    --project=${PROJECT_ID} \
    --stagingLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/staging \
    --tempLocation=gs://${PROJECT_ID}/dataflow/pipelines/${PIPELINE_FOLDER}/temp \
    --runner=DataflowRunner \
    --inputFilePattern=gs://path/to/demo_file.csv \
    --outputTopic=projects/${PROJECT_ID}/topics/${TOPIC_NAME}"
 * }
 * </pre>
 *
 */
public class TextToPubsub {

  /**
   * The custom options supported by the pipeline. Inherits
   * standard configuration options.
   */
  public interface Options extends PipelineOptions {
    @Description("The file pattern to read records from (e.g. gs://bucket/file-*.csv)")
    @Required
    ValueProvider<String> getInputFilePattern();
    void setInputFilePattern(ValueProvider<String> value);

    @Description("The name of the topic which data should be published to. "
        + "The name should be in the format of projects/<project-id>/topics/<topic-name>.")
    @Required
    ValueProvider<String> getOutputTopic();
    void setOutputTopic(ValueProvider<String> value);
  }

  /**
   * Main entry-point for the pipeline. Reads in the
   * command-line arguments, parses them, and executes
   * the pipeline.
   *
   * @param args  Arguments passed in from the command-line.
   */
  public static void main(String[] args) {

    // Parse the user options passed from the command-line
    Options options = PipelineOptionsFactory
        .fromArgs(args)
        .withValidation()
        .as(Options.class);

    run(options);
  }

  /**
   * Executes the pipeline with the provided execution
   * parameters.
   *
   * @param options The execution parameters.
   */
  public static PipelineResult run(Options options) {
    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps:
     *  1) Read from the text source.
     *  2) Write each text record to Pub/Sub
     */
    pipeline
        .apply("Read Text Data", TextIO.read().from(options.getInputFilePattern()))
        .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

Python

Python ソースコードは利用できません。

Cloud Storage Text to Cloud Spanner

Cloud Storage Text to Cloud Spanner テンプレートは、Cloud Storage から CSV テキストファイルを読み取り、Cloud Spanner データベースにインポートするバッチパイプラインです。

このパイプラインの要件:

ターゲットの Cloud Spanner データベースとテーブルが存在している必要があります。
Cloud Storage バケットの読み取り権限と、対象の Cloud Spanner データベースに対する書き込み権限が必要です。
CSV ファイルを含む入力 Cloud Storage パスが存在している必要があります。
CSV ファイルの JSON 記述を含むインポートマニフェストファイルを作成し、そのマニフェストを Cloud Storage に保存する必要があります。
ターゲットの Cloud Spanner データベースにすでにスキーマがある場合、マニフェストファイルで指定された列は、ターゲットデータベースのスキーマ内の対応する列と同じデータ型である必要があります。

ASCII または UTF-8 でエンコードされたマニフェストファイルは、次の形式に一致する必要があります。

マニフェストの形式と例

次のメッセージタイプに対応するマニフェストファイルの形式は、プロトコルバッファで参照できます。

message ImportManifest {
  // The per-table import manifest.
  message TableManifest {
    // Required. The name of the destination table.
    string table_name = 1;
    // Required. The CSV files to import. This value can be either a filepath or a glob pattern.
    repeated string file_patterns = 2;
    // The schema for a table column.
    message Column {
      // Required for each Column that you specify. The name of the column in the
      // destination table.
      string column_name = 1;
      // Required for each Column that you specify. The type of the column. Supports the
      // following data types: BOOL, INT64, FLOAT64, STRING, DATE, and TIMESTAMP.
      string type_name = 2;
    }
    // Optional. The schema for the table columns.
    repeated Column columns = 3;
  }
  // Required. The TableManifest of the tables to be imported.
  repeated TableManifest tables = 1;
}

次の例は、Albums と Singers というテーブルをインポートするためのマニフェストファイルを示しています。Albums テーブルは、ジョブがデータベースから取得する列スキーマを使用し、Singers テーブルはマニフェストファイルが指定するスキーマを使用します。

{
  "tables": [
    {
      "table_name": "Albums",
      "file_patterns": [
        "gs://bucket1/Albums_1.csv",
        "gs://bucket1/Albums_2.csv"
      ]
    },
    {
      "table_name": "Singers",
      "file_patterns": [
        "gs://bucket1/Singers*.csv"
      ],
      "columns": [
        {"column_name": "SingerId", "type_name": "INT64"},
        {"column_name": "FirstName", "type_name": "STRING"},
        {"column_name": "LastName", "type_name": "STRING"}
      ]
    }
  ]
}

インポートするテキストファイルは、ASCII または UTF-8 エンコードの CSV 形式である必要があります。UTF-8 エンコードファイルではバイトオーダーマーク（BOM）を使用しないことをおすすめします。
データは次のタイプのいずれかに一致する必要があります:
- INT64
- FLOAT64
- BOOL
- STRING
- DATE
- TIMESTAMP

注: インポートマニフェストファイルで対象のテーブルの列名とデータ型を指定しない場合、CSV ファイルの列は対象のデータベースの列と同じ順序である必要があります。次のクエリを実行すると、テーブル内の列の順序を確認できます。

SELECT * FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME =
      [YOUR_TABLE_NAME] ORDER BY ORDINAL_POSITION

テンプレートのパラメータ

パラメータ	説明
`instanceId`	Cloud Spanner データベースのインスタンス ID。
`databaseId`	Cloud Spanner データベースのデータベース ID。
`importManifest`	Cloud Storage のインポートマニフェストファイルへのパス。
`columnDelimiter`	ソースファイルが使用する列区切り文字。デフォルト値は `,` です。
`fieldQualifier`	`columnDelimiter` を含む、ソースファイル内の任意の値を囲む文字。デフォルト値は `"` です。
`trailingDelimiter`	ソースファイルの行の末尾に区切り文字があるかどうかを指定します（つまり、`columnDelimiter` 文字が各行の最後の列の値の後に表示されるかどうか）。デフォルト値は `true` です。
`escape`	ソースファイルが使用するエスケープ文字。デフォルトでは、このパラメータは設定されておらず、テンプレートではエスケープ文字は使用されません。
`nullString`	`NULL` 値を表す文字列。デフォルトでは、このパラメータは設定されておらず、テンプレートでは null 文字列は使用されません。
`dateFormat`	日付列を解析するために使用される形式。デフォルトでは、パイプラインは日付列を `yyyy-M-d[' 00:00:00']` として解析します。たとえば 2019-01-31 または 2019-1-1 00:00:00 とします。日付形式が異なる場合は、`java.time.format.DateTimeFormatter` のパターンを使って指定します。
`timestampFormat`	timestamp 列を解析するために使用される形式。タイムスタンプが長整数の場合、Unix エポックとして解析されます。それ以外の場合は、`java.time.format.DateTimeFormatter.ISO_INSTANT` フォーマットを使用して、文字列として解析されます。その他の場合は、独自のパターン文字列を指定します。たとえば、`MMM dd yyyy HH:mm:ss.SSSVV` タイムスタンプの形式は `"Jan 21 1998 01:02:03.456+08:00".` です。

日付形式やタイムスタンプ形式をカスタマイズする必要がある場合は、有効な java.time.format.DateTimeFormatter パターンであることを確認してください。date 列と timestamp 列のカスタマイズされた形式のその他の例を次の表に示します。

種類	入力値	フォーマット	備考
`DATE`	2011-3-31		デフォルトでは、テンプレートはこの形式を解析できます。`dateFormat` パラメータを指定する必要はありません。
`DATE`	2011-3-31 00:00:00		デフォルトでは、テンプレートはこの形式を解析できます。形式を指定する必要はありません。必要に応じて `yyyy-M-d' 00:00:00'` が使用できます。
`DATE`	2018 年 4 月 1 日	dd MMM, yy
`DATE`	西暦 2019 年 4 月 3 日水曜日	EEEE, LLLL d, yyyy G
`TIMESTAMP`	2019-01-02T11:22:33Z 2019-01-02T11:22:33.123Z 2019-01-02T11:22:33.12356789Z		デフォルトの形式 `ISO_INSTANT` はこのタイプのタイムスタンプを解析できます。`timestampFormat` パラメータを指定する必要はありません。
`TIMESTAMP`	1568402363		デフォルトでは、テンプレートはこのタイプのタイムスタンプを解析し、Unix エポック時間として扱います。
`TIMESTAMP`	2008 年 6 月 3 日（火曜日）11:05:30 グリニッジ標準時	EEE, d MMM yyyy HH:mm:ss VV
`TIMESTAMP`	2018 年 12 月 31 日 110530.123 太平洋標準時	yyyy/MM/dd HHmmss.SSSz
`TIMESTAMP`	2019-01-02T11:22:33Z または 2019-01-02T11:22:33.123Z	yyyy-MM-dd'T'HH:mm:ss [.SSS]VV	入力列が 2019-01-02T11:22:33Z と 2019-01-02T11:22:33.123Z の場合、この形式のタイムスタンプはデフォルトの形式で解析できます。独自のフォーマットパラメータを指定する必要はありません。ただし、`yyyy-MM-dd'T'HH:mm:ss[.SSS]VV` を使用すると、両方のケースに対応します。接尾辞「Z」は文字リテラルではなくタイムゾーン ID として解析する必要があるため、`yyyy-MM-dd'T'HH:mm:ss[.SSS]'Z'`は使用できないことに注意してください。内部的には、タイムスタンプ列は `java.time.Instant` に変換されます。そのため、UTC で指定するか、タイムゾーン情報を設定する必要があります。2019-01-02 11:22:33 のようなローカル日時は、有効な `java.time.Instant` として解析されません。

テンプレートの実行

Console

Google Cloud Platform Console での実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから Cloud Storage Text to Cloud Spanner テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

gcloud コマンドラインツールでの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

[DATAFLOW_REGION] を Cloud Dataflow ジョブを実行するリージョンで置き換えます（たとえば、us-central1）。
[YOUR_INSTANCE_ID] は、Cloud Spanner インスタンスの ID に置き換えます。
[YOUR_DATABASE_ID] は、Cloud Spanner データベースの ID に置き換えます。
[GCS_PATH_TO_IMPORT_MANIFEST] を、インポートマニフェストファイルへの Cloud Storage パスに置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。

gcloud dataflow jobs run [JOB_NAME] \
    --gcs-location='gs://dataflow-templates/[VERSION]/GCS_Text_to_Cloud_Spanner' \
    --region=[DATAFLOW_REGION] \
    --parameters='instanceId=[YOUR_INSTANCE_ID],databaseId=[YOUR_DATABASE_ID],importManifest=[GCS_PATH_TO_IMPORT_MANIFEST]'

API

REST API を使用して実行する

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

[YOUR_PROJECT_ID] を実際のプロジェクト ID に置き換えます。
[DATAFLOW_REGION] を Cloud Dataflow ジョブを実行するリージョンで置き換えます（たとえば、us-central1）。
[YOUR_INSTANCE_ID] は、Cloud Spanner インスタンスの ID に置き換えます。
[YOUR_DATABASE_ID] は、Cloud Spanner データベースの ID に置き換えます。
[GCS_PATH_TO_IMPORT_MANIFEST] を、インポートマニフェストファイルへの Cloud Storage パスに置き換えます。
[JOB_NAME] を任意のジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。

POST https://dataflow.googleapis.com/v1b3/projects/[YOUR_PROJECT_ID]/locations/[DATAFLOW_REGION]/templates:launch?gcsPath=gs://dataflow-templates/[VERSION]/GCS_Text_to_Cloud_Spanner

{
   "jobName": "[JOB_NAME]",
   "parameters": {
       "instanceId": "[YOUR_INSTANCE_ID]",
       "databaseId": "[YOUR_DATABASE_ID]",
       "importManifest": "[GCS_PATH_TO_IMPORT_MANIFEST]"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

テンプレートのソースコード

Java: SDK 2.x

このテンプレートのソースコードは、GitHub 上の GoogleCloudPlatform/DataflowTemplates リポジトリにあります。

Python

Python ソースコードは利用できません。

Java Database Connectivity (JDBC) to BigQuery

JDBC to BigQuery テンプレートは、リレーショナルデータベーステーブルから既存の BigQuery テーブルにデータをコピーするバッチパイプラインです。このパイプラインは、JDBC を使用してリレーショナルデータベースに接続します。このテンプレートを使用すると、使用可能な JDBC ドライバがある任意のリレーショナルデータベースから BigQuery にデータをコピーできます。保護をさらに強化するために、Cloud KMS 鍵で暗号化された Base64 でエンコードされたユーザー名、パスワード、接続文字列パラメータを渡すこともできます。詳しくは Cloud KMS API 暗号化エンドポイントで、ユーザー名、パスワード、接続文字列パラメータの暗号化の詳細をご覧ください。

このパイプラインの要件:

リレーショナルデータベース用の JDBC ドライバが使用可能である必要があります。
パイプラインの実行前に BigQuery テーブルが存在する必要があります。
BigQuery テーブルに互換性のあるスキーマが必要です。
リレーショナルデータベースは、Cloud Dataflow が実行されているサブネットからアクセス可能である必要があります。

テンプレートのパラメータ

パラメータ	説明
`driverJars`	ドライバ jar のカンマ区切りリスト。例: `gs://<my-bucket>/driver_jar1.jar,gs://<my-bucket>/driver_jar2.jar`
`driverClassName`	JDBC ドライバのクラス名。例: `com.mysql.jdbc.Driver`
`connectionURL`	JDBC 接続 URL 文字列。例: `jdbc:mysql://some-host:3306/sampledb`Cloud KMS 鍵で暗号化された Base64 エンコード文字列として渡すことができます。
`query`	ソースで実行されるクエリでデータを抽出します。例: `select * from sampledb.sample_table`
`outputTable`	BigQuery 出力テーブルの場所。`<my-project>:<my-dataset>.<my-table>` の形式で指定します。
`bigQueryLoadingTemporaryDirectory`	BigQuery 読み込みプロセスで使用する一時ディレクトリ。例: `gs://<my-bucket>/my-files/temp_dir`
`connectionProperties`	（省略可）JDBC 接続に使用するプロパティ文字列。例: `unicode=true&characterEncoding=UTF-8`
`username`	（省略可）JDBC 接続に使用するユーザー名。Cloud KMS 鍵で暗号化された Base64 エンコード文字列として渡すことができます。
`password`	（省略可）JDBC 接続に使用するパスワード。Cloud KMS 鍵で暗号化された Base64 エンコード文字列として渡すことができます。
`KMSEncryptionKey`	（省略可）ユーザー名、パスワード、接続文字列を復号するための Cloud KMS 暗号鍵。Cloud KMS 鍵が渡された場合、ユーザー名、パスワード、接続文字列はすべて暗号化されて渡されます。

BigQuery テンプレートへの JDBC の実行

Console

Google Cloud Platform Console からの実行

GCP Console の Cloud Dataflow ページに移動します。

Cloud Dataflow ページに移動

[テンプレートからジョブを作成] をクリックします。

[Cloud Dataflow テンプレート] プルダウンメニューから JDBC to BigQuery テンプレートを選択します。
[ジョブ名] フィールドにジョブ名を入力します。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

GCLOUD

gcloud コマンドラインツールからの実行

注: テンプレートを実行する gcloud コマンドラインツールを使用するには、Cloud SDK のバージョン 138.0.0 以降が必要です。

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
DRIVER_PATHS を、カンマで区切った JDBC ドライバの Cloud Storage パスに置き換えます。
DRIVER_CLASS_NAME をドライブクラス名に置き換えます。
JDBC_CONNECTION_URL を JDBC 接続の URL に置き換えます。
SOURCE_SQL_QUERY を、ソースデータベースで実行する SQL クエリに置き換えます。
YOUR_DATASET を BigQuery データセットに置き換え、YOUR_TABLE_NAME を BigQuery テーブル名に置き換えます。
PATH_TO_TEMP_DIR_ON_GCS を、一時ディレクトリへの Cloud Storage パスに置き換えます。
CONNECTION_PROPERTIES を必要に応じて JDBC 接続プロパティに置き換えます。
CONNECTION_USERNAME を JDBC 接続ユーザー名に置き換えます。
CONNECTION_PASSWORD を JDBC 接続パスワードに置き換えます。
KMS_ENCRYPTION_KEY を Cloud KMS 暗号鍵に置き換えます。

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Jdbc_to_BigQuery \
    --parameters \
driverJars=DRIVER_PATHS,\
driverClassName=DRIVER_CLASS_NAME,\
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

API

REST API からの実行

このテンプレートを実行する場合、テンプレートへの Cloud Storage パスが必要です。

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

このサンプルの次の値は置き換える必要があります。

YOUR-PROJECT-ID をプロジェクト ID に置き換えます。
JOB_NAME を、選択したジョブ名に置き換えます。ジョブ名は正規表現 [a-z]([-a-z0-9]{0,38}[a-z0-9])? と一致する必要があります。
DRIVER_PATHS を、カンマで区切った JDBC ドライバの Cloud Storage パスに置き換えます。
DRIVER_CLASS_NAME をドライブクラス名に置き換えます。
JDBC_CONNECTION_URL を JDBC 接続の URL に置き換えます。
SOURCE_SQL_QUERY を、ソースデータベースで実行する SQL クエリに置き換えます。
YOUR_DATASET を BigQuery データセットに置き換え、YOUR_TABLE_NAME を BigQuery テーブル名に置き換えます。
PATH_TO_TEMP_DIR_ON_GCS を、一時ディレクトリへの Cloud Storage パスに置き換えます。
CONNECTION_PROPERTIES を必要に応じて JDBC 接続プロパティに置き換えます。
CONNECTION_USERNAME を JDBC 接続ユーザー名に置き換えます。
CONNECTION_PASSWORD を JDBC 接続パスワードに置き換えます。
KMS_ENCRYPTION_KEY を Cloud KMS 暗号鍵に置き換えます。

POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Jdbc_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverJars": "DRIVER_PATHS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "JDBC_CONNECTION_URL",
       "query": "SOURCE_SQL_QUERY",
       "outputTable": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD",
       "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
   },
   "environment": { "zone": "us-central1-f" }
}

テンプレートのソースコード

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2018 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package com.google.cloud.teleport.templates;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.io.DynamicJdbcIO;
import com.google.cloud.teleport.templates.common.JdbcConverters;
import com.google.cloud.teleport.util.KMSEncryptedNestedValueProvider;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 */
public class JdbcToBigQuery {

  private static ValueProvider<String> maybeDecrypt(
      ValueProvider<String> unencryptedValue, ValueProvider<String> kmsKey) {
    return new KMSEncryptedNestedValueProvider(unencryptedValue, kmsKey);
  }

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link
   * JdbcToBigQuery#run(JdbcConverters.JdbcToBigQueryOptions)} method to start the pipeline and
   * invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {

    // Parse the user options passed from the command-line
    JdbcConverters.JdbcToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(JdbcConverters.JdbcToBigQueryOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(JdbcConverters.JdbcToBigQueryOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    pipeline
        /*
         * Step 1: Read records via JDBC and convert to TableRow
         *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
         */
        .apply(
            "Read from JdbcIO",
            DynamicJdbcIO.<TableRow>read()
                .withDataSourceConfiguration(
                    DynamicJdbcIO.DynamicDataSourceConfiguration.create(
                            options.getDriverClassName(),
                            maybeDecrypt(options.getConnectionURL(), options.getKMSEncryptionKey()))
                        .withUsername(
                            maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()))
                        .withPassword(
                            maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()))
                        .withDriverJars(options.getDriverJars())
                        .withConnectionProperties(options.getConnectionProperties()))
                .withQuery(options.getQuery())
                .withCoder(TableRowJsonCoder.of())
                .withRowMapper(JdbcConverters.getResultSetToTableRow()))
        /*
         * Step 2: Append TableRow to an existing BigQuery table
         */
        .apply(
            "Write to BigQuery",
            BigQueryIO.writeTableRows()
                .withoutValidation()
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory())
                .to(options.getOutputTable()));

    // Execute the pipeline and return the result.
    return pipeline.run();
  }
}

Python

Python ソースコードは利用できません。

Google 提供のバッチ テンプレート

Cloud Bigtable to Cloud Storage Avro

テンプレートのパラメータ

Cloud Bigtable to Cloud Storage Avro file テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Bigtable to Cloud Storage SequenceFile

テンプレートのパラメータ

Cloud Bigtable to Cloud Storage SequenceFile テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java: SDK 2.x

Python

Cloud Datastore to Cloud Storage Text

テンプレートのパラメータ

Cloud Datastore to Cloud Storage Text テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Spanner to Cloud Storage Avro

テンプレートのパラメータ

テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Spanner to Cloud Storage Text

テンプレートのパラメータ

Cloud Spanner to Cloud Storage Text テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Storage Avro to Cloud Bigtable

テンプレートのパラメータ

Cloud Storage Avro file to Cloud Bigtable テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Storage Avro to Cloud Spanner

テンプレートのパラメータ

テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java

Python

Cloud Storage SequenceFile to Cloud Bigtable

テンプレートのパラメータ

Cloud Storage SequenceFile to Cloud Bigtable テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Java: SDK 2.x

Python

Cloud Storage Text to BigQuery

テンプレートのパラメータ

Cloud Storage Text to BigQuery テンプレートの実行

Console

GCLOUD

API

テンプレートのソースコード

Google 提供のバッチテンプレート