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

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

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

BigQuery to Cloud Storage TFRecords
BigQuery to Cloud Storage Parquet (Storage API)
Cloud Bigtable to Cloud Storage Avro
Cloud Bigtable to Cloud Storage Parquet
Cloud Bigtable to Cloud Storage SequenceFile
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 Parquet to Cloud Bigtable
Cloud Storage SequenceFile to Cloud Bigtable
Cloud Storage Text to BigQuery
Cloud Storage Text to Datastore
Cloud Storage Text to Pub/Sub (Batch)
Cloud Storage Text to Cloud Spanner
Java Database Connectivity（JDBC）to BigQuery
Apache Cassandra to Cloud Bigtable
Apache Hive to BigQuery
File Format Conversion

BigQuery to Cloud Storage TFRecords

BigQuery to Cloud Storage TFRecords テンプレートは、BigQuery クエリからデータを読み取り、Cloud Storage バケットに TFRecord 形式で書き込むパイプラインです。トレーニング、テスト、検証の分割パーセンテージを指定できます。デフォルトでは、トレーニングセットの分割パーセンテージは 1 または 100%、テストセットと検証セットは 0 または 0% です。データセットの分割を設定する場合は、トレーニング、テスト、検証の合計が 1 または 100% になるようにする必要があります（たとえば、0.6 + 0.2 + 0.2）。Dataflow では、各出力データセットに最適なシャード数が自動的に設定されます。

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

BigQuery のデータセットとテーブルが存在すること。
パイプラインの実行前に出力先の Cloud Storage バケットが存在すること。トレーニング、テスト、検証のサブディレクトリは、事前に作成する必要はありません。自動的に生成されます。

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

パラメータ	説明
`readQuery`	ソースからデータを抽出する BigQuery SQL クエリ。例: `select * from dataset1.sample_table`
`outputDirectory`	トレーニング、テスト、検証の TFRecord ファイルを書き込む最上位の Cloud Storage パスの接頭辞。例: `gs://mybucket/output`トレーニング、テスト、検証の TFRecord ファイルのサブディレクトリは、`outputDirectory` から自動的に生成されます。例: `gs://mybucket/output/train`
`trainingPercentage`	（省略可）トレーニングの TFRecord ファイルに割り当てられるクエリデータの割合。デフォルト値は 1 または 100% です。
`testingPercentage`	（省略可）テストの TFRecord ファイルに割り当てられるクエリデータの割合。デフォルト値は 0 または 0% です。
`validationPercentage`	（省略可）検証の TFRecord ファイルに割り当てられるクエリデータの割合。デフォルト値は 0 または 0% です。
`outputSuffix`	（省略可）トレーニング、テスト、検証で書き込まれる TFRecord ファイルの接頭辞。デフォルト値は `.tfrecord` です。

BigQuery to Cloud Storage TFRecord ファイルテンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the BigQuery to TFRecords template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cloud_BigQuery_to_GCS_TensorFlow_Records

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cloud_BigQuery_to_GCS_TensorFlow_Records \
    --parameters readQuery=READ_QUERY,outputDirectory=OUTPUT_DIRECTORY,trainingPercentage=TRAINING_PERCENTAGE,testingPercentage=TESTING_PERCENTAGE,validationPercentage=VALIDATION_PERCENTAGE,outputSuffix=OUTPUT_FILENAME_SUFFIX

次の値を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
READ_QUERY: 実行される BigQuery クエリ
OUTPUT_DIRECTORY: 出力データセットの Cloud Storage パス接頭辞
TRAINING_PERCENTAGE: トレーニングデータセットの分割割合の小数値
TESTING_PERCENTAGE: テストデータセットの分割割合の小数値
VALIDATION_PERCENTAGE: 検証データセットの分割割合の小数値
OUTPUT_FILENAME_SUFFIX: 出力される TensorFlow レコードのファイルサフィックス

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Cloud_BigQuery_to_GCS_TensorFlow_Records

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_BigQuery_to_GCS_TensorFlow_Records
{
   "jobName": "JOB_NAME",
   "parameters": {
       "readQuery":"READ_QUERY",
       "outputDirectory":"OUTPUT_DIRECTORY",
       "trainingPercentage":"TRAINING_PERCENTAGE",
       "testingPercentage":"TESTING_PERCENTAGE",
       "validationPercentage":"VALIDATION_PERCENTAGE",
       "outputSuffix":"OUTPUT_FILENAME_SUFFIX"
   },
   "environment": { "zone": "us-central1-f" }
}

次の値を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
READ_QUERY: 実行される BigQuery クエリ
OUTPUT_DIRECTORY: 出力データセットの Cloud Storage パス接頭辞
TRAINING_PERCENTAGE: トレーニングデータセットの分割割合の小数値
TESTING_PERCENTAGE: テストデータセットの分割割合の小数値
VALIDATION_PERCENTAGE: 検証データセットの分割割合の小数値
OUTPUT_FILENAME_SUFFIX: 出力される TensorFlow レコードのファイルサフィックス

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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.cloud.teleport.templates.common.BigQueryConverters.BigQueryReadOptions;
import com.google.protobuf.ByteString;
import java.util.Iterator;
import java.util.Random;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.TFRecordIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.Partition;
import org.apache.beam.sdk.transforms.Reshuffle;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.tensorflow.example.Example;
import org.tensorflow.example.Feature;
import org.tensorflow.example.Features;

/**
 * Dataflow template which reads BigQuery data and writes it to GCS as a set of TFRecords. The
 * source is a SQL query.
 */
public class BigQueryToTFRecord {

  /**
   * The {@link BigQueryToTFRecord#buildFeatureFromIterator(Class, Object, Feature.Builder)} method
   * handles {@link GenericData.Array} that are passed into the {@link
   * BigQueryToTFRecord#buildFeature} method creating a TensorFlow feature from the record.
   */
  private static final String TRAIN = "train/";
  private static final String TEST = "test/";
  private static final String VAL = "val/";

  private static void buildFeatureFromIterator(
      Class<?> fieldType, Object field, Feature.Builder feature) {
    ByteString byteString;
    GenericData.Array f = (GenericData.Array) field;
    if (fieldType == Long.class) {
      Iterator<Long> longIterator = f.iterator();
      while (longIterator.hasNext()) {
        Long longValue = longIterator.next();
        feature.getInt64ListBuilder().addValue(longValue);
      }
    } else if (fieldType == double.class) {
      Iterator<Double> doubleIterator = f.iterator();
      while (doubleIterator.hasNext()) {
        double doubleValue = doubleIterator.next();
        feature.getFloatListBuilder().addValue((float) doubleValue);
      }
    } else if (fieldType == String.class) {
      Iterator<Utf8> stringIterator = f.iterator();
      while (stringIterator.hasNext()) {
        String stringValue = stringIterator.next().toString();
        byteString = ByteString.copyFromUtf8(stringValue);
        feature.getBytesListBuilder().addValue(byteString);
      }
    } else if (fieldType == boolean.class) {
      Iterator<Boolean> booleanIterator = f.iterator();
      while (booleanIterator.hasNext()) {
        Boolean boolValue = booleanIterator.next();
        int boolAsInt = boolValue ? 1 : 0;
        feature.getInt64ListBuilder().addValue(boolAsInt);
      }
    }

  }

  /**
   * The {@link BigQueryToTFRecord#buildFeature} method takes in an individual field and type
   * corresponding to a column value from a SchemaAndRecord Object returned from a BigQueryIO.read()
   * step. The method builds a TensorFlow Feature based on the type of the object- ie: STRING, TIME,
   * INTEGER etc..
   */
  private static Feature buildFeature(Object field, String type) {
    Feature.Builder feature = Feature.newBuilder();
    ByteString byteString;

    switch (type) {
      case "STRING":
      case "TIME":
      case "DATE":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(String.class, field, feature);
        } else {
          byteString = ByteString.copyFromUtf8(field.toString());
          feature.getBytesListBuilder().addValue(byteString);
        }
        break;
      case "BYTES":
        byteString = ByteString.copyFrom((byte[]) field);
        feature.getBytesListBuilder().addValue(byteString);
        break;
      case "INTEGER":
      case "INT64":
      case "TIMESTAMP":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(Long.class, field, feature);
        } else {
          feature.getInt64ListBuilder().addValue((long) field);
        }
        break;
      case "FLOAT":
      case "FLOAT64":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(double.class, field, feature);
        } else {
          feature.getFloatListBuilder().addValue((float) (double) field);
        }
        break;
      case "BOOLEAN":
      case "BOOL":
        if (field instanceof GenericData.Array) {
          buildFeatureFromIterator(boolean.class, field, feature);
        } else {
          int boolAsInt = (boolean) field ? 1 : 0;
          feature.getInt64ListBuilder().addValue(boolAsInt);
        }
        break;
      default:
        throw new RuntimeException("Unsupported type: " + type);
    }
    return feature.build();
  }

  /**
   * The {@link BigQueryToTFRecord#record2Example(SchemaAndRecord)} method uses takes in a
   * SchemaAndRecord Object returned from a BigQueryIO.read() step and builds a TensorFlow Example
   * from the record.
   */
  @VisibleForTesting
  protected static byte[] record2Example(SchemaAndRecord schemaAndRecord) {
    Example.Builder example = Example.newBuilder();
    Features.Builder features = example.getFeaturesBuilder();
    GenericRecord record = schemaAndRecord.getRecord();
    for (TableFieldSchema field : schemaAndRecord.getTableSchema().getFields()) {
      Feature feature = buildFeature(record.get(field.getName()), field.getType());
      features.putFeature(field.getName(), feature);
    }
    return example.build().toByteArray();
  }

  /**
   * The {@link BigQueryToTFRecord#concatURI} method uses takes in a Cloud Storage URI and a
   * subdirectory name and safely concatenates them. The resulting String is used as a sink for
   * TFRecords.
   */
  private static String concatURI(String dir, String folder) {
    if (dir.endsWith("/")) {
      return dir + folder;
    } else {
      return dir + "/" + folder;
    }
  }

  /**
   * The {@link BigQueryToTFRecord#applyTrainTestValSplit} method transforms the PCollection by
   * randomly partitioning it into PCollections for each dataset.
   */
  static PCollectionList<byte[]> applyTrainTestValSplit(PCollection<byte[]> input,
      ValueProvider<Float> trainingPercentage,
      ValueProvider<Float> testingPercentage,
      ValueProvider<Float> validationPercentage,
      Random rand) {
    return input
        .apply(Partition.of(
            3,
            (Partition.PartitionFn<byte[]>) (number, numPartitions) -> {
              Float train = trainingPercentage.get();
              Float test = testingPercentage.get();
              Float validation = validationPercentage.get();
              Double d = rand.nextDouble();
              if (train + test + validation != 1) {
                throw new RuntimeException(String.format("Train %.2f, Test %.2f, Validation"
                    + " %.2f percentages must add up to 100 percent", train, test, validation));
              }
              if (d < train) {
                return 0;
              } else if (d >= train && d < train + test) {
                return 1;
              } else {
                return 2;
              }
            }));
  }

  /**
   * Run the pipeline.
   */
  public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {
    Random rand = new Random(100); // set random seed
    Pipeline pipeline = Pipeline.create(options);

    PCollection<byte[]> bigQueryToExamples =
        pipeline
            .apply(
                "RecordToExample",
                BigQueryIO.read(BigQueryToTFRecord::record2Example)
                    .fromQuery(options.getReadQuery())
                    .withCoder(ByteArrayCoder.of())
                    .withTemplateCompatibility()
                    .withoutValidation()
                    .usingStandardSql()
                    .withMethod(BigQueryIO.TypedRead.Method.DIRECT_READ)
                // Enable BigQuery Storage API
            ).apply("ReshuffleResults", Reshuffle.viaRandomKey());

    PCollectionList<byte[]> partitionedExamples = applyTrainTestValSplit(
        bigQueryToExamples,
        options.getTrainingPercentage(),
        options.getTestingPercentage(),
        options.getValidationPercentage(),
        rand);

    partitionedExamples.get(0).apply(
        "WriteTFTrainingRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, TRAIN)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    partitionedExamples.get(1).apply(
        "WriteTFTestingRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, TEST)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    partitionedExamples.get(2).apply(
        "WriteTFValidationRecord",
        FileIO.<byte[]>write()
            .via(TFRecordIO.sink())
            .to(ValueProvider.NestedValueProvider.of(
                options.getOutputDirectory(),
                dir -> concatURI(dir, VAL)))
            .withNumShards(0)
            .withSuffix(options.getOutputSuffix()));

    return pipeline.run();
  }

  /**
   * Define command line arguments.
   */
  public interface Options
      extends BigQueryReadOptions {

    @Description("The GCS directory to store output TFRecord files.")
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> outputDirectory);

    @Description("The output suffix for TFRecord Files")
    @Default.String(".tfrecord")
    ValueProvider<String> getOutputSuffix();

    void setOutputSuffix(ValueProvider<String> outputSuffix);

    @Description("The training percentage split for TFRecord Files")
    @Default.Float(1)
    ValueProvider<Float> getTrainingPercentage();

    void setTrainingPercentage(ValueProvider<Float> trainingPercentage);

    @Description("The testing percentage split for TFRecord Files")
    @Default.Float(0)
    ValueProvider<Float> getTestingPercentage();

    void setTestingPercentage(ValueProvider<Float> testingPercentage);

    @Description("The validation percentage split for TFRecord Files")
    @Default.Float(0)
    ValueProvider<Float> getValidationPercentage();

    void setValidationPercentage(ValueProvider<Float> validationPercentage);
  }
}

Python

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

BigQuery の Parquet へのエクスポート（Storage API 経由）

BigQuery export to Parquet テンプレートは、BigQuery テーブルからデータを読み取り、Parquet 形式で Cloud Storage バケットに書き込むバッチパイプラインです。このテンプレートは、BigQuery Storage API を使用してデータをエクスポートします。

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

パイプラインを実行する前に、入力 BigQuery テーブルが存在すること。
パイプラインを実行する前に、出力先の Cloud Storage バケットが存在すること。

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

パラメータ	説明
`tableRef`	BigQuery 入力テーブルの場所。例: `<my-project>:<my-dataset>.<my-table>`
`bucket`	Parquet ファイルを書き込む Cloud Storage フォルダ。例:`gs://mybucket/exports`
`numShards`	（省略可）出力ファイルのシャード数。デフォルト値は 1 です。
`fields`	（省略可）入力 BigQuery テーブルから選択するフィールドのカンマ区切りのリスト。

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the BigQuery export to Parquet template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/flex/BigQuery_to_Parquet

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --template-file-gcs-location=gs://dataflow-templates/latest/flex/BigQuery_to_Parquet \
    --parameters \
tableRef=BIGQUERY_TABLE,\
bucket=OUTPUT_DIRECTORY,\
numShards=NUM_SHARDS,\
fields=FIELDS

次の値を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
BIGQUERY_TABLE: BigQuery テーブル名
OUTPUT_DIRECTORY: 出力ファイルを格納する Cloud Storage フォルダ
NUM_SHARDS: 目的の出力ファイルシャードの数
FIELDS: 入力 BigQuery テーブルから選択するフィールドのカンマ区切りリスト
LOCATION: 実行リージョン（例: us-central1）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/flex/BigQuery_to_Parquet

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "tableRef": "BIGQUERY_TABLE",
          "bucket": "OUTPUT_DIRECTORY",
          "numShards": "NUM_SHARDS",
          "fields": "FIELDS"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/latest/flex/BigQuery_to_Parquet",
   }
}

次の値を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
BIGQUERY_TABLE: BigQuery テーブル名
OUTPUT_DIRECTORY: 出力ファイルを格納する Cloud Storage フォルダ
NUM_SHARDS: 目的の出力ファイルシャードの数
FIELDS: 入力 BigQuery テーブルから選択するフィールドのカンマ区切りリスト
LOCATION: 実行リージョン（例: us-central1）

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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.v2.templates;

import com.google.api.gax.rpc.InvalidArgumentException;
import com.google.api.services.bigquery.model.TableReference;
import com.google.cloud.bigquery.storage.v1beta1.BigQueryStorageClient;
import com.google.cloud.bigquery.storage.v1beta1.ReadOptions.TableReadOptions;
import com.google.cloud.bigquery.storage.v1beta1.Storage.CreateReadSessionRequest;
import com.google.cloud.bigquery.storage.v1beta1.Storage.ReadSession;
import com.google.cloud.bigquery.storage.v1beta1.TableReferenceProto;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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.Validation.Required;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link BigQueryToParquet} pipeline exports data from a BigQuery table to Parquet file(s) in a
 * Google Cloud Storage bucket.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>BigQuery Table exists.
 *   <li>Google Cloud Storage bucket exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT=my-project
 * BUCKET_NAME=my-bucket
 * TABLE={$PROJECT}:my-dataset.my-table
 *
 * # Set containerization vars
 * IMAGE_NAME=my-image-name
 * TARGET_GCR_IMAGE=gcr.io/${PROJECT}/${IMAGE_NAME}
 * BASE_CONTAINER_IMAGE=my-base-container-image
 * BASE_CONTAINER_IMAGE_VERSION=my-base-container-image-version
 * APP_ROOT=/path/to/app-root
 * COMMAND_SPEC=/path/to/command-spec
 *
 * # Build and upload image
 * mvn clean package \
 * -Dimage=${TARGET_GCR_IMAGE} \
 * -Dbase-container-image=${BASE_CONTAINER_IMAGE} \
 * -Dbase-container-image.version=${BASE_CONTAINER_IMAGE_VERSION} \
 * -Dapp-root=${APP_ROOT} \
 * -Dcommand-spec=${COMMAND_SPEC}
 *
 * # Create an image spec in GCS that contains the path to the image
 * {
 *    "docker_template_spec": {
 *       "docker_image": $TARGET_GCR_IMAGE
 *     }
 *  }
 *
 * # Execute template:
 * API_ROOT_URL="https://dataflow.googleapis.com"
 * TEMPLATES_LAUNCH_API="${API_ROOT_URL}/v1b3/projects/${PROJECT}/templates:launch"
 * JOB_NAME="bigquery-to-parquet-`date +%Y%m%d-%H%M%S-%N`"
 *
 * time curl -X POST -H "Content-Type: application/json"     \
 *     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 *     "${TEMPLATES_LAUNCH_API}"`
 *     `"?validateOnly=false"`
 *     `"&dynamicTemplate.gcsPath=${BUCKET_NAME}/path/to/image-spec"`
 *     `"&dynamicTemplate.stagingLocation=${BUCKET_NAME}/staging" \
 *     -d '
 *      {
 *       "jobName":"'$JOB_NAME'",
 *       "parameters": {
 *           "tableRef":"'$TABLE'",
 *           "bucket":"'$BUCKET_NAME/results'",
 *           "numShards":"5",
 *           "fields":"field1,field2"
 *        }
 *       }
 *      '
 * </pre>
 */
public class BigQueryToParquet {

  /* Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(BigQueryToParquet.class);

  /** File suffix for file to be written. */
  private static final String FILE_SUFFIX = ".parquet";

  /** Factory to create BigQueryStorageClients. */
  static class BigQueryStorageClientFactory {

    /**
     * Creates BigQueryStorage client for use in extracting table schema.
     * @return BigQueryStorageClient
     */
    static BigQueryStorageClient create() {
      try {
        return BigQueryStorageClient.create();
      } catch (IOException e) {
        LOG.error("Error connecting to BigQueryStorage API: " + e.getMessage());
        throw new RuntimeException(e);
      }
    }
  }

  /** Factory to create ReadSessions. */
  static class ReadSessionFactory {

    /**
     * Creates ReadSession for schema extraction.
     *
     * @param client BigQueryStorage client used to create ReadSession.
     * @param tableString String that represents table to export from.
     * @param tableReadOptions TableReadOptions that specify any fields in the table to filter on.
     * @return session ReadSession object that contains the schema for the export.
     */
    static ReadSession create(
        BigQueryStorageClient client, String tableString, TableReadOptions tableReadOptions) {
      TableReference tableReference = BigQueryHelpers.parseTableSpec(tableString);
      String parentProjectId = "projects/" + tableReference.getProjectId();

      TableReferenceProto.TableReference storageTableRef =
          TableReferenceProto.TableReference.newBuilder()
              .setProjectId(tableReference.getProjectId())
              .setDatasetId(tableReference.getDatasetId())
              .setTableId(tableReference.getTableId())
              .build();

      CreateReadSessionRequest.Builder builder =
          CreateReadSessionRequest.newBuilder()
              .setParent(parentProjectId)
              .setReadOptions(tableReadOptions)
              .setTableReference(storageTableRef);
      try {
        return client.createReadSession(builder.build());
      } catch (InvalidArgumentException iae) {
        LOG.error("Error creating ReadSession: " + iae.getMessage());
        throw new RuntimeException(iae);
      }
    }
  }

  /**
   * The {@link BigQueryToParquetOptions} class provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface BigQueryToParquetOptions extends PipelineOptions {
    @Description("BigQuery table to export from in the form <project>:<dataset>.<table>")
    @Required
    String getTableRef();

    void setTableRef(String tableRef);

    @Description("GCS bucket to export BigQuery table data to (e.g. gs://mybucket/folder/).")
    @Required
    String getBucket();

    void setBucket(String bucket);

    @Description("Optional: Number of shards for output file.")
    @Default.Integer(0)
    Integer getNumShards();

    void setNumShards(Integer numShards);

    @Description("Optional: Comma separated list of fields to select from the table.")
    String getFields();

    void setFields(String fields);
  }

  /**
   * The {@link BigQueryToParquet#getTableSchema(ReadSession)} method gets Avro schema for table
   * using from the {@link ReadSession} object.
   *
   * @param session ReadSession that contains schema for table, filtered by fields if any.
   * @return avroSchema Avro schema for table. If fields are provided then schema will only contain
   *     those fields.
   */
  private static Schema getTableSchema(ReadSession session) {
    Schema avroSchema;

    avroSchema = new Schema.Parser().parse(session.getAvroSchema().getSchema());
    LOG.info("Schema for export is: " + avroSchema.toString());

    return avroSchema;
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    BigQueryToParquetOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(BigQueryToParquetOptions.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(BigQueryToParquetOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);

    TableReadOptions.Builder builder = TableReadOptions.newBuilder();

    /* Add fields to filter export on, if any. */
    if (options.getFields() != null) {
      builder.addAllSelectedFields(Arrays.asList(options.getFields().split(",\\s*")));
    }

    TableReadOptions tableReadOptions = builder.build();
    BigQueryStorageClient client = BigQueryStorageClientFactory.create();
    ReadSession session =
        ReadSessionFactory.create(client, options.getTableRef(), tableReadOptions);

    // Extract schema from ReadSession
    Schema schema = getTableSchema(session);
    client.close();

    TypedRead<GenericRecord> readFromBQ =
        BigQueryIO.read(SchemaAndRecord::getRecord)
            .from(options.getTableRef())
            .withTemplateCompatibility()
            .withMethod(Method.DIRECT_READ)
            .withCoder(AvroCoder.of(schema));

    if (options.getFields() != null) {
      List<String> selectedFields = Splitter.on(",").splitToList(options.getFields());
      readFromBQ =
          selectedFields.isEmpty() ? readFromBQ : readFromBQ.withSelectedFields(selectedFields);
    }

    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Write records to Google Cloud Storage in Parquet format.
     */
    pipeline
        /*
         * Step 1: Read records via BigQueryIO using supplied schema as a PCollection of
         *         {@link GenericRecord}.
         */
        .apply(
            "ReadFromBigQuery", readFromBQ)
        /*
         * Step 2: Write records to Google Cloud Storage as one or more Parquet files
         *         via {@link ParquetIO}.
         */
        .apply(
            "WriteToParquet",
            FileIO.<GenericRecord>write()
                .via(ParquetIO.sink(schema))
                .to(options.getBucket())
                .withNumShards(options.getNumShards())
                .withSuffix(FILE_SUFFIX));

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

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 インスタンスの Google Cloud プロジェクトの 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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Cloud Bigtable to Avro Files on Cloud Storage template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_Avro

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

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
OUTPUT_DIRECTORY: データの書き込み先の Cloud Storage パス（例: gs://mybucket/somefolder）
FILENAME_PREFIX: Avro ファイル名の接頭辞（例: output-）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_Avro

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
OUTPUT_DIRECTORY: データの書き込み先の Cloud Storage パス（例: gs://mybucket/somefolder）
FILENAME_PREFIX: Avro ファイル名の接頭辞（例: output-）

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

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(PipelineUtils.tweakPipelineOptions(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.
   */
  protected 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();
    }
  }
}

Cloud Bigtable to Cloud Storage Parquet

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

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

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

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

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

Cloud Bigtable to Cloud Storage Parquet ファイルテンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Cloud Bigtable to Parquet template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_Parquet

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
OUTPUT_DIRECTORY: データの書き込み先の Cloud Storage パス（例: gs://mybucket/somefolder）
FILENAME_PREFIX: Parquet ファイル名の接頭辞（例: output-）
NUM_SHARDS: 出力する Parquet ファイルの数（例: 1）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_Parquet

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
OUTPUT_DIRECTORY: データの書き込み先の Cloud Storage パス（例: gs://mybucket/somefolder）
FILENAME_PREFIX: Parquet ファイル名の接頭辞（例: output-）
NUM_SHARDS: 出力する Parquet ファイルの数（例: 1）

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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 static com.google.cloud.teleport.bigtable.BigtableToAvro.toByteArray;

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 java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
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.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.PCollection;

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

  /** 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. Default file prefix is set to"
            + " \"output\".")
    @Default.String("output")
    ValueProvider<String> getFilenamePrefix();

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

    @Description(
        "The maximum number of output shards produced when writing. Default number is runner"
            + " defined. ")
    @Default.Integer(0)
    ValueProvider<Integer> getNumShards();

    @SuppressWarnings("unused")
    void setNumShards(ValueProvider<Integer> numShards);
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line 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();
    }
  }

  /**
   * Runs a pipeline to export data from a Cloud Bigtable table to Parquet file(s) in GCS.
   *
   * @param options arguments to the pipeline
   */
  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(PipelineUtils.tweakPipelineOptions(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();
    }

    /**
     * Steps:
     * 1) Read records from Bigtable.
     * 2) Convert a Bigtable Row to a GenericRecord.
     * 3) Write GenericRecord(s) to GCS in parquet format.
     */
    pipeline
        .apply("Read from Bigtable", read)
        .apply("Transform to Parquet", MapElements.via(new BigtableToParquetFn()))
        .setCoder(AvroCoder.of(GenericRecord.class, BigtableRow.getClassSchema()))
        .apply(
            "Write to Parquet in GCS",
            FileIO.<GenericRecord>write()
                .via(ParquetIO.sink(BigtableRow.getClassSchema()))
                .to(options.getOutputDirectory())
                .withPrefix(options.getFilenamePrefix())
                .withSuffix(".parquet")
                .withNumShards(options.getNumShards()));

    return pipeline.run();
  }

  /**
   * Translates a {@link PCollection} of Bigtable {@link Row} to a {@link PCollection} of {@link
   * GenericRecord}.
   */
  static class BigtableToParquetFn extends SimpleFunction<Row, GenericRecord> {
    @Override
    public GenericRecord 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 GenericRecordBuilder(BigtableRow.getClassSchema())
          .set("key", key)
          .set("cells", cells)
          .build();
    }
  }
}

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 インスタンスの Google Cloud プロジェクトの 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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Cloud Bigtable to SequenceFile Files on Cloud Storage template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_SequenceFile

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID]: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの 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-）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Cloud_Bigtable_to_GCS_SequenceFile

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cloud_Bigtable_to_GCS_SequenceFile
{
   "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" }
}

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID]: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの 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-）

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

Java: SDK 2.x

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

Datastore to Cloud Storage Text

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

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

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

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

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

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Datastore to Text Files on Cloud Storage template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Datastore_to_GCS_Text \
    --parameters \
datastoreReadGqlQuery="SELECT * FROM DATASTORE_KIND",\
datastoreReadProjectId=PROJECT_ID,\
datastoreReadNamespace=DATASTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://BUCKET_NAME/output/

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
BUCKET_NAME: Cloud Storage バケットの名前
DATASTORE_KIND: Datastore エンティティのタイプ
DATASTORE_NAMESPACE: Datastore エンティティの名前空間
JAVASCRIPT_FUNCTION: JavaScript 関数名
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Datastore_to_GCS_Text

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

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

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
BUCKET_NAME: Cloud Storage バケットの名前
DATASTORE_KIND: Datastore エンティティのタイプ
DATASTORE_NAMESPACE: Datastore エンティティの名前空間
JAVASCRIPT_FUNCTION: JavaScript 関数名
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス

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

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();
  }
}

Cloud Spanner to Cloud Storage Avro

Cloud Spanner to Avro Files on 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 バケットが存在すること。
Dataflow ジョブの実行に必要な IAM ロールに加えて、Cloud Spanner のデータの読み取りと Cloud Storage バケットへの書き込みに適切な IAM ロールも必要です。

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

パラメータ	説明
`instanceId`	エクスポートする Cloud Spanner データベースのインスタンス ID。
`databaseId`	エクスポートする Cloud Spanner データベースのデータベース ID。
`outputDir`	Avro ファイルのエクスポート先にする Cloud Storage パス。エクスポートジョブによって、このパスの下にディレクトリが新規作成されます。ここに、エクスポートされたファイルが格納されます。
`snapshotTime`	（省略可）読み取る Cloud Spanner データベースのバージョンに対応するタイムスタンプ。タイムスタンプは RFC 3339 UTC「Zulu」形式で指定する必要があります。例: `1990-12-31T23:59:60Z`タイムスタンプは過去の日付でなければならず、タイムスタンプステイルネスの最大値が適用されます。
`spannerProjectId`	（省略可）データを読み取る Cloud Spanner データベースの Google Cloud プロジェクト ID。

Cloud Spanner to Avro Files on Cloud Storage テンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Cloud Spanner to Avro Files on Cloud Storage template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
- Cloud Console の Cloud Spanner 部分に表示するため、ジョブ名は、cloud-spanner-export-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] の形式になっている必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

gcloud dataflow jobs run JOB_NAME \
    --gcs-location='gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro' \
    --region=DATAFLOW_REGION \
    --parameters='instanceId=INSTANCE_ID,databaseId=DATABASE_ID,outputDir=GCS_DIRECTORY'

以下を置き換えます。

JOB_NAME: 任意のジョブ名。
DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
GCS_STAGING_LOCATION: 一時ファイルを書き込むパス（例: gs://mybucket/temp）
INSTANCE_ID: Cloud Spanner インスタンス ID
DATABASE_ID: Cloud Spanner データベース ID
GCS_DIRECTORY: Avro ファイルがにエクスポートされる Cloud Storage パス

API

REST API から実行する

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

gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/DATAFLOW_REGION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_Spanner_to_GCS_Avro
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "outputDir": "gs://GCS_DIRECTORY"
   }
}

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

PROJECT_ID: プロジェクト ID
DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
INSTANCE_ID: Cloud Spanner インスタンス ID
DATABASE_ID: Cloud Spanner データベース ID
GCS_DIRECTORY: Avro ファイルがにエクスポートされる Cloud Storage パス
JOB_NAME: 任意のジョブ名
- Cloud Console の Cloud Spanner 部分に表示されるジョブ名は、cloud-spanner-export-INSTANCE_ID-DATABASE_ID というフォーマットに一致させる必要があります。

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

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);

    @Description("If set, specifies the time when the snapshot must be taken."
      + " String is in the RFC 3339 format. "
      + " Example - 1990-12-31T23:59:60Z"
      + " Timestamp must be in the past and Maximum timestamp staleness applies."
      + " https://cloud.google.com/spanner/docs/timestamp-bounds#maximum_timestamp_staleness")
    @Default.String(value = "")
    ValueProvider<String> getSnapshotTime();
    void setSnapshotTime(ValueProvider<String> value);

    @Description("GCP Project Id of where the Spanner table lives.")
    ValueProvider<String> getSpannerProjectId();
    void setSpannerProjectId(ValueProvider<String> value);

    @Description("If true, Timestamps are exported with timestamp-micros logical type."
      + "By default,Timestamps are exported as ISO8601 strings at nanosecond precision.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportTimestampAsLogicalType();
    void setShouldExportTimestampAsLogicalType(ValueProvider<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()
            .withProjectId(options.getSpannerProjectId())
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());
    p.begin()
        .apply(
            "Run Export",
            new ExportTransform(spannerConfig, options.getOutputDir(),
                options.getTestJobId(), options.getSnapshotTime(),
                options.getShouldExportTimestampAsLogicalType()));
    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();
    }
  }
}

Cloud Spanner to Cloud Storage Text

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

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

パイプラインを実行する前に、入力 Spanner テーブルが存在すること。

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

パラメータ	説明
`spannerProjectId`	データを読み取る Cloud Spanner データベースの Google Cloud プロジェクト ID。
`spannerDatabaseId`	リクエストされたテーブルのデータベース ID。
`spannerInstanceId`	リクエストされたテーブルのインスタンス ID。
`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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Cloud Spanner to Text Files on Cloud Storage template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Spanner_to_GCS_Text \
    --parameters \
spannerProjectId=PROJECT_ID,\
spannerDatabaseId=DATABASE_ID,\
spannerInstanceId=INSTANCE_ID,\
spannerTable=TABLE_ID,\
textWritePrefix=gs://BUCKET_NAME/output/,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
DATABASE_ID: Spanner データベース ID
BUCKET_NAME: Cloud Storage バケットの名前
INSTANCE_ID: Spanner インスタンス ID
TABLE_ID: Spanner テーブル ID
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
JAVASCRIPT_FUNCTION: JavaScript 関数名

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Spanner_to_GCS_Text

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Spanner_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "spannerProjectId": "PROJECT_ID",
       "spannerDatabaseId": "DATABASE_ID",
       "spannerInstanceId": "INSTANCE_ID",
       "spannerTable": "TABLE_ID",
       "textWritePrefix": "gs://BUCKET_NAME/output/",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION"
   },
   "environment": { "zone": "us-central1-f" }
}

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
DATABASE_ID: Spanner データベース ID
BUCKET_NAME: Cloud Storage バケットの名前
INSTANCE_ID: Spanner インスタンス ID
TABLE_ID: Spanner テーブル ID
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
JAVASCRIPT_FUNCTION: JavaScript 関数名

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

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.LocalSpannerIO;
import org.apache.beam.sdk.io.gcp.spanner.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
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()
            .withHost(options.getSpannerHost())
            .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", LocalSpannerIO.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");
  }
}

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 インスタンスの Google Cloud プロジェクトの ID。
`bigtableInstanceId`	テーブルが含まれる Cloud Bigtable インスタンスの ID。
`bigtableTableId`	インポート対象の Cloud Bigtable テーブルの ID。
`inputFilePattern`	データが存在する Cloud Storage パスのパターン（例: `gs://mybucket/somefolder/prefix*`）。

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Avro Files on Cloud Storage to Cloud Bigtable template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Bigtable

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
INPUT_FILE_PATTERN: データが存在する Cloud Storage パスのパターン（例: gs://mybucket/somefolder/prefix*）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Bigtable

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
INPUT_FILE_PATTERN: データが存在する Cloud Storage パスのパターン（例: gs://mybucket/somefolder/prefix*）

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

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.base.MoreObjects;
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.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 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 {
  private static final Logger LOG = LoggerFactory.getLogger(AvroToBigtable.class);

  /** Maximum number of mutations allowed per row by Cloud bigtable. */
  private static final int MAX_MUTATIONS_PER_ROW = 100000;

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

  /** 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("If true, a large row is split into multiple MutateRows requests.")
    ValueProvider<Boolean> getSplitLargeRows();

    @Description(
        "Set the option to split a large row into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);

    @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(PipelineUtils.tweakPipelineOptions(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",
            ParDo.of(
                AvroToBigtableFn.createWithSplitLargeRows(
                    options.getSplitLargeRows(), MAX_MUTATIONS_PER_ROW)))
        .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 DoFn<BigtableRow, KV<ByteString, Iterable<Mutation>>> {
    private final ValueProvider<Boolean> splitLargeRowsFlag;
    private Boolean splitLargeRows;
    private final int maxMutationsPerRow;

    public static AvroToBigtableFn create() {
      return new AvroToBigtableFn(StaticValueProvider.of(false), MAX_MUTATIONS_PER_ROW);
    }

    public static AvroToBigtableFn createWithSplitLargeRows(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      return new AvroToBigtableFn(splitLargeRowsFlag, maxMutationsPerRequest);
    }

    private AvroToBigtableFn(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      this.splitLargeRowsFlag = splitLargeRowsFlag;
      this.maxMutationsPerRow = maxMutationsPerRequest;
    }

    @Setup
    public void setup() {
      if (splitLargeRowsFlag != null) {
        splitLargeRows = splitLargeRowsFlag.get();
      }
      splitLargeRows = MoreObjects.firstNonNull(splitLargeRows, DEFAULT_SPLIT_LARGE_ROWS);
      LOG.info("splitLargeRows set to: " + splitLargeRows);
    }

    @ProcessElement
    public void processElement(
        @Element BigtableRow row, OutputReceiver<KV<ByteString, Iterable<Mutation>>> out) {
      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();
      int cellsProcessed = 0;
      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());
        cellsProcessed++;

        if (this.splitLargeRows && cellsProcessed % maxMutationsPerRow == 0) {
          // Send a MutateRow request when we have accumulated max mutations per row.
          out.output(KV.of(key, mutations.build()));
          mutations = ImmutableList.builder();
        }
      }

      // Flush any remaining mutations.
      ImmutableList remainingMutations = mutations.build();
      if (!remainingMutations.isEmpty()) {
        out.output(KV.of(key, remainingMutations));
      }
    }
  }

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

Cloud Storage Avro to Cloud Spanner

Cloud Storage Avro files to Cloud Spanner テンプレートは、Cloud Storage に保存されている Cloud Spanner からエクスポートされた 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 パス。

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Avro Files on Cloud Storage to Cloud Spanner template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
- Cloud Console の Cloud Spanner 部分に表示するため、ジョブ名は、cloud-spanner-import-[YOUR_INSTANCE_ID]-[YOUR_DATABASE_ID] の形式になっている必要があります。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

gcloud dataflow jobs run JOB_NAME \
    --gcs-location='gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner' \
    --region=DATAFLOW_REGION \
    --staging-location=GCS_STAGING_LOCATION \
    --parameters='instanceId=INSTANCE_ID,databaseId=DATABASE_ID,inputDir=GCS_DIRECTORY'

（API のみ）PROJECT_ID: プロジェクト ID
DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
JOB_NAME: 任意のジョブ名
INSTANCE_ID: データベースを含む Spanner インスタンスの ID
DATABASE_ID: インポート先の Spanner データベースの ID
（gcloud のみ）GCS_STAGING_LOCATION: 一時ファイルを書き込むためのパス（例: gs://mybucket/temp）
GCS_DIRECTORY: Avro ファイルのインポート元となる Cloud Storage パス（例: gs://mybucket/somefolder）

API

REST API から実行する

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

gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/DATAFLOW_REGION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/GCS_Avro_to_Cloud_Spanner
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "inputDir": "gs://GCS_DIRECTORY"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

（API のみ）PROJECT_ID: プロジェクト ID
DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
JOB_NAME: 任意のジョブ名
INSTANCE_ID: データベースを含む Spanner インスタンスの ID
DATABASE_ID: インポート先の Spanner データベースの ID
（gcloud のみ）GCS_STAGING_LOCATION: 一時ファイルを書き込むためのパス（例: gs://mybucket/temp）
GCS_DIRECTORY: Avro ファイルのインポート元となる Cloud Storage パス（例: gs://mybucket/somefolder）

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

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("By default the import pipeline is not blocked on index creation, and it "
      + "may complete with indexes still being created in the background. In testing, it may "
      + "be useful to set this option to false so that the pipeline waits until indexes are "
      + "finished.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForIndexes();

    void setWaitForIndexes(ValueProvider<Boolean> value);

    @Description("By default the import pipeline is not blocked on foreign key creation, and it "
      + "may complete with foreign keys still being created in the background. In testing, it may "
      + "be useful to set this option to false so that the pipeline waits until foreign keys are "
      + "finished.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getWaitForForeignKeys();

    void setWaitForForeignKeys(ValueProvider<Boolean> value);

    @Description("Indexes and Foreign keys are created after dataload. If there are more than "
      + "40 DDL statements to be executed after dataload, it is preferable to create the "
      + "indexes before datalod. This is the flag to turn the feature off.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getEarlyIndexCreateFlag();

    void setEarlyIndexCreateFlag(ValueProvider<Boolean> value);

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

    @Description("GCP Project Id of where the Spanner table lives.")
    ValueProvider<String> getSpannerProjectId();

    void setSpannerProjectId(ValueProvider<String> value);

    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()
            .withProjectId(options.getSpannerProjectId())
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId());

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

    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();
    }
  }
}

Cloud Storage Parquet to Cloud Bigtable

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

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

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

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

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

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Parquet files on Cloud Storage to Cloud Bigtable template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Parquet_to_Cloud_Bigtable

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
INPUT_FILE_PATTERN: データが存在する Cloud Storage パスのパターン（例: gs://mybucket/somefolder/prefix*）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_Parquet_to_Cloud_Bigtable

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの ID
INSTANCE_ID: テーブルが含まれる Cloud Bigtable インスタンスの ID
TABLE_ID: エクスポート対象の Cloud Bigtable テーブルの ID
INPUT_FILE_PATTERN: データが存在する Cloud Storage パスのパターン（例: gs://mybucket/somefolder/prefix*）

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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 static com.google.cloud.teleport.bigtable.AvroToBigtable.toByteString;

import com.google.bigtable.v2.Mutation;
import com.google.protobuf.ByteString;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.runners.direct.DirectRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
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.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link ParquetToBigtable} pipeline imports data from Parquet 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 Parquet files has a 'family' of "f1", the
 * Bigtable table should have a column family of "f1".
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Bigtable instance.
 *   <li>Bigtable table with compatible table schema.
 *   <li>Google Cloud Storage input bucket and parquet file(s) exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 *
 * # Set the pipeline vars
 * PROJECT_ID=PROJECT ID HERE
 * PIPELINE_FOLDER=gs://${PROJECT_ID}/dataflow/pipelines/parquet-to-bigtable
 * BIGTABLE_INSTANCE_ID=BIGTABLE INSTANCE ID HERE
 * BIGTABLE_TABLE_ID=BIGTABLE TABLE ID HERE
 *
 * # Set the runner
 * RUNNER=DataflowRunner
 *
 * # Build the template
 * mvn compile exec:java \
 * -Dexec.mainClass=com.google.cloud.teleport.bigtable.ParquetToBigtable \
 * -Dexec.cleanupDaemonThreads=false \
 * -Dexec.args=" \
 * --project=${PROJECT_ID} \
 * --stagingLocation=${PIPELINE_FOLDER}/staging \
 * --tempLocation=${PIPELINE_FOLDER}/temp \
 * --templateLocation=${PIPELINE_FOLDER}/template \
 * --runner=${RUNNER}"
 *
 * # Execute the template
 * JOB_NAME=parquet-to-bigtable-$USER-`date +"%Y%m%d-%H%M%S%z"`
 *
 * gcloud dataflow jobs run ${JOB_NAME} \
 * --gcs-location=${PIPELINE_FOLDER}/template \
 * --zone=us-east1-d \
 * --parameters \
 * "bigtableProjectId=${PROJECT_ID},\
 * bigtableInstanceId=${BIGTABLE_INSTANCE_ID},\
 * bigtableTableId=${BIGTABLE_TABLE_ID},\
 * inputFilePattern=${PIPELINE_FOLDER}/path/to/file/filename-*.parquet"
 * </pre>
 */
public class ParquetToBigtable {
  private static final Logger LOG = LoggerFactory.getLogger(AvroToBigtable.class);

  /** Maximum number of mutations allowed per row by Cloud bigtable. */
  private static final int MAX_MUTATIONS_PER_ROW = 100000;

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

  /**
   * 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*.parquet)")
    ValueProvider<String> getInputFilePattern();

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

    @Description(
        "If true, a large row is split into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

  /**
   * Runs a pipeline to import Parquet 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);
  }

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

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

    /**
     * Steps: 1) Read records from Parquet File. 2) Convert a GenericRecord to a
     * KV<ByteString,Iterable<Mutation>>. 3) Write KV to Bigtable's table.
     */
    pipeline
        .apply(
            "Read from Parquet",
            ParquetIO.read(BigtableRow.getClassSchema()).from(options.getInputFilePattern()))
        .apply(
            "Transform to Bigtable",
            ParDo.of(
                ParquetToBigtableFn.createWithSplitLargeRows(
                    options.getSplitLargeRows(), MAX_MUTATIONS_PER_ROW)))
        .apply("Write to Bigtable", write);

    return pipeline.run();
  }

  static class ParquetToBigtableFn
          extends DoFn<GenericRecord, KV<ByteString, Iterable<Mutation>>> {

    private final ValueProvider<Boolean> splitLargeRowsFlag;
    private Boolean splitLargeRows;
    private final int maxMutationsPerRow;

    public static ParquetToBigtableFn create() {
      return new ParquetToBigtableFn(StaticValueProvider.of(false), MAX_MUTATIONS_PER_ROW);
    }

    public static ParquetToBigtableFn createWithSplitLargeRows(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      return new ParquetToBigtableFn(splitLargeRowsFlag, maxMutationsPerRequest);
    }

    @Setup
    public void setup() {
      if (splitLargeRowsFlag != null) {
        splitLargeRows = splitLargeRowsFlag.get();
      }
      splitLargeRows = MoreObjects.firstNonNull(splitLargeRows, DEFAULT_SPLIT_LARGE_ROWS);
      LOG.info("splitLargeRows set to: " + splitLargeRows);
    }

    private ParquetToBigtableFn(
        ValueProvider<Boolean> splitLargeRowsFlag, int maxMutationsPerRequest) {
      this.splitLargeRowsFlag = splitLargeRowsFlag;
      this.maxMutationsPerRow = maxMutationsPerRequest;
    }

    @ProcessElement
    public void processElement(ProcessContext ctx) {
      Class runner = ctx.getPipelineOptions().getRunner();
      ByteString key = toByteString((ByteBuffer) ctx.element().get(0));

      // 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();
      List<Object> cells = (List) ctx.element().get(1);
      int cellsProcessed = 0;
      for (Object element : cells) {
        Mutation.SetCell setCell = null;
        if (runner.isAssignableFrom(DirectRunner.class)) {
          setCell =
                  Mutation.SetCell.newBuilder()
                          .setFamilyName(((GenericData.Record) element).get(0).toString())
                          .setColumnQualifier(toByteString((ByteBuffer) ((GenericData.Record) element).get(1)))
                          .setTimestampMicros((Long) ((GenericData.Record) element).get(2))
                          .setValue(toByteString((ByteBuffer) ((GenericData.Record) element).get(3)))
                          .build();
        } else {
          BigtableCell bigtableCell = (BigtableCell) element;
          setCell =
                  Mutation.SetCell.newBuilder()
                          .setFamilyName(bigtableCell.getFamily().toString())
                          .setColumnQualifier(toByteString(bigtableCell.getQualifier()))
                          .setTimestampMicros(bigtableCell.getTimestamp())
                          .setValue(toByteString(bigtableCell.getValue()))
                          .build();
        }
        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
        cellsProcessed++;

        if (this.splitLargeRows && cellsProcessed % maxMutationsPerRow == 0) {
          // Send a MutateRow request when we have accumulated max mutations per row.
          ctx.output(KV.of(key, mutations.build()));
          mutations = ImmutableList.builder();
        }
      }

      // Flush any remaining mutations.
      ImmutableList remainingMutations = mutations.build();
      if (!remainingMutations.isEmpty()) {
        ctx.output(KV.of(key, remainingMutations));
      }
    }
  }
}

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 インスタンスの Google Cloud プロジェクトの 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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Parquet files on Cloud Storage to Cloud Bigtable template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_SequenceFile_to_Cloud_Bigtable

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

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの 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*）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_SequenceFile_to_Cloud_Bigtable

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/GCS_SequenceFile_to_Cloud_Bigtable
{
   "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" }
}

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PROJECT_ID: データを読み取る Cloud Bigtable インスタンスの Google Cloud プロジェクトの 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*）

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

Java: SDK 2.x

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

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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Text Files on Cloud Storage to BigQuery template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

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

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
JAVASCRIPT_FUNCTION: UDF の名前
PATH_TO_BIGQUERY_SCHEMA_JSON: スキーマ定義を含む JSON ファイルへの Cloud Storage パス
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
PATH_TO_TEXT_DATA: テキストデータセットへの Cloud Storage パス
BIGQUERY_TABLE: BigQuery テーブル名
PATH_TO_TEMP_DIR_ON_GCS: 一時ディレクトリへの Cloud Storage パス

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery

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

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
JAVASCRIPT_FUNCTION: UDF の名前
PATH_TO_BIGQUERY_SCHEMA_JSON: スキーマ定義を含む JSON ファイルへの Cloud Storage パス
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
PATH_TO_TEXT_DATA: テキストデータセットへの Cloud Storage パス
BIGQUERY_TABLE: BigQuery テーブル名
PATH_TO_TEMP_DIR_ON_GCS: 一時ディレクトリへの Cloud Storage パス

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

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_APPEND)
                .withCustomGcsTempLocation(options.getBigQueryLoadingTemporaryDirectory()));

    pipeline.run();
  }
}

Cloud Storage Text to Datastore

Cloud Storage Text to Datastore テンプレートは、Cloud Storage に保存されたテキストファイルを読み取り、JSON にエンコードされたエンティティを Datastore に書き込むバッチパイプラインです。入力テキストファイルの各行は、指定された 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`	Datastore エンティティを書き込む Google Cloud プロジェクト ID
`errorWritePath`	処理中に発生したエラーを書き込むために使用するエラーログ出力ファイル。例: `gs://bucket-name/errors.txt`

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Text Files on Cloud Storage to Datastore template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

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=JAVASCRIPT_FUNCTION,\
datastoreWriteProjectId=PROJECT_ID,\
errorWritePath=ERROR_FILE_WRITE_PATH

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PATH_TO_INPUT_TEXT_FILES: Cloud Storage 上の入力ファイルパターン
JAVASCRIPT_FUNCTION: JavaScript 関数名
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
ERROR_FILE_WRITE_PATH: Cloud Storage 上のエラーファイルの目的のパス

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_Text_to_Datastore

POST https://dataflow.googleapis.com/v1b3/projects/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": "JAVASCRIPT_FUNCTION",
       "datastoreWriteProjectId": "PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
PATH_TO_INPUT_TEXT_FILES: Cloud Storage 上の入力ファイルパターン
JAVASCRIPT_FUNCTION: JavaScript 関数名
PATH_TO_JAVASCRIPT_UDF_FILE: JavaScript コードを含む .js ファイルへの Cloud Storage パス
ERROR_FILE_WRITE_PATH: Cloud Storage 上のエラーファイルの目的のパス

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

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();
  }
}

Cloud Storage Text to Pub/Sub (Batch)

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

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

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

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

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

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

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

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Text Files on Cloud Storage to Pub/Sub (Batch) template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

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

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
TOPIC_NAME: Pub/Sub トピック名
BUCKET_NAME: Cloud Storage バケットの名前

API

REST API からの実行

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

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_PubSub

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

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
TOPIC_NAME: Pub/Sub トピック名
BUCKET_NAME: Cloud Storage バケットの名前

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

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();
  }
}

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, NUMERIC, 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 =
      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` として解析されません。

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

Console

Cloud Console で実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Text Files on Cloud Storage to Cloud Spanner template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

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

DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
INSTANCE_ID: Cloud Spanner インスタンス ID
DATABASE_ID: Cloud Spanner データベース ID
GCS_PATH_TO_IMPORT_MANIFEST: インポートマニフェストファイルへの Cloud Storage パス
JOB_NAME: 任意のジョブ名。

API

REST API で実行する

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

gs://dataflow-templates/VERSION/GCS_Text_to_Cloud_Spanner

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

{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "importManifest": "GCS_PATH_TO_IMPORT_MANIFEST"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

PROJECT_ID: プロジェクト ID
DATAFLOW_REGION: Dataflow ジョブを実行するリージョン（例: us-central1）
INSTANCE_ID: Cloud Spanner インスタンス ID
DATABASE_ID: Cloud Spanner データベース ID
GCS_PATH_TO_IMPORT_MANIFEST: インポートマニフェストファイルへの Cloud Storage パス
JOB_NAME: 任意のジョブ名。

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

Java: SDK 2.x

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

Java Database Connectivity（JDBC）to BigQuery

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

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

リレーショナルデータベース用の JDBC ドライバが使用可能である必要があります。
パイプラインを実行する前に BigQuery テーブルが存在している必要があります。
BigQuery テーブルに互換性のあるスキーマが必要です。
リレーショナルデータベースは、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`Base64 でエンコードされ、Cloud KMS 鍵で暗号化される文字列として渡すことができます。
`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

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the JDBC to BigQuery template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

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=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名
DRIVER_PATHS: カンマで区切った JDBC ドライバの Cloud Storage パス
DRIVER_CLASS_NAME: ドライブクラス名
JDBC_CONNECTION_URL: JDBC 接続 URL
SOURCE_SQL_QUERY: ソースデータベースで実行する SQL クエリ
DATASET: BigQuery データセット。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 暗号鍵

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Jdbc_to_BigQuery

POST https://dataflow.googleapis.com/v1b3/projects/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": "PROJECT_ID:DATASET.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" }
}

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名
DRIVER_PATHS: カンマで区切った JDBC ドライバの Cloud Storage パス
DRIVER_CLASS_NAME: ドライブクラス名
JDBC_CONNECTION_URL: JDBC 接続 URL
SOURCE_SQL_QUERY: ソースデータベースで実行する SQL クエリ
DATASET: BigQuery データセット。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 暗号鍵

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

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();
  }
}

Apache Cassandra to Cloud Bigtable

Apache Cassandra to Cloud Bigtable テンプレートは、Apache Cassandra から Cloud Bigtable にテーブルをコピーします。このテンプレートに対して行う必要のある構成は最小限に抑えられており、Cassandra のテーブル構造を Cloud Bigtable で可能な限り再現します。

Apache Cassandra to Cloud Bigtable テンプレートは次の場合に役立ちます。

短いダウンタイムしか許容されない状況で Apache Cassandra データベースを移行する。
グローバルなサービス提供を目的として、Cassandra のテーブルを Cloud Bigtable に定期的に複製する。

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

パイプラインを実行する前に、複製先の Cloud Bigtable テーブルが存在していること。
Dataflow ワーカーと Apache Cassandra ノードの間のネットワーク接続。

型変換

Apache Cassandra to Cloud Bigtable テンプレートでは、Apache Cassandra のデータ型が Cloud Bigtable のデータ型に自動的に変換されます。

ほとんどのプリミティブは Cloud Bigtable と Apache Cassandra で同じように表現されますが、次のプリミティブは異なる方法で表現されます。

Date と Timestamp は DateTime オブジェクトに変換されます。
UUID は String に変換されます。
Varint は BigDecimal に変換されます。

Apache Cassandra は、Tuple、List、Set、Map などの複雑な型もネイティブにサポートしています。Apache Beam にはタプルに対応する型がないため、このパイプラインではタプルはサポートされません。

たとえば、Apache Cassandra では「mylist」という名前の List 型の列を使用し、次の表のような値を格納できます。

row	mylist
1	`(a,b,c)`

このリスト列はパイプラインによって 3 つの異なる列に展開されます（Cloud Bigtable ではこれを列修飾子といいます）。列の名前は「mylist」ですが、「mylist[0]」のようにリスト内のアイテムのインデックスがパイプラインによって追加されます。

row	mylist[0]	mylist[1]	mylist[2]
1	a	b	c

このパイプラインでは、セットもリストと同じように処理されますが、セルがキーか値かを示す接尾辞が追加されます。

row	mymap
1	`{"first_key":"first_value","another_key":"different_value"}`

変換後、テーブルは次のようになります。

row	mymap[0].key	mymap[0].value	mymap[1].key	mymap[1].value
1	first_key	first_value	another_key	different_value

主キー変換

Apache Cassandra では、主キーはデータ定義言語を使用して定義されます。主キーは、単純、複合、クラスタ化された列の複合のいずれかです。Cloud Bigtable では、バイト配列を辞書順に並べ替える行キーの手動作成がサポートされています。このパイプラインはキーの型の情報を自動で収集し、複数の値に基づいて行キーを作成するためのベストプラクティスに基づいてキーを作成します。

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

パラメータ	説明
`cassandraHosts`	Apache Cassandra ノードのホストをカンマ区切りのリストで表したもの。
`cassandraPort`	（省略可）ノード上の Apache Cassandra に到達する TCP ポート（デフォルトは `9042`）。
`cassandraKeyspace`	テーブルが配置されている Apache Cassandra キースペース。
`cassandraTable`	コピーする Apache Cassandra テーブル。
`bigtableProjectId`	Apache Cassandra テーブルがコピーされる Cloud Bigtable インスタンスの Google Cloud プロジェクト ID。
`bigtableInstanceId`	Apache Cassandra テーブルをコピーする Cloud Bigtable インスタンス ID。
`bigtableTableId`	Apache Cassandra テーブルをコピーする Cloud Bigtable テーブルの名前。
`defaultColumnFamily`	（省略可）Cloud Bigtable テーブルの列ファミリーの名前（デフォルトは `default`）。
`rowKeySeparator`	（省略可）行キーの作成に使用される区切り文字（デフォルトは `#`）。

Apache Cassandra to Cloud Bigtable テンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Apache Cassandra to Cloud Bigtable template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Cassandra_To_Cloud_Bigtable

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Cassandra_To_Cloud_Bigtable \
    --parameters\
bigtableProjectId=PROJECT_ID,\
bigtableInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableTableId=BIGTABLE_TABLE_ID,\
cassandraHosts=CASSANDRA_HOSTS,\
cassandraKeyspace=CASSANDRA_KEYSPACE,\
cassandraTable=CASSANDRA_TABLE

以下を置き換えます。

PROJECT_ID: Cloud Bigtable が配置されているプロジェクト ID
JOB_NAME: 任意のジョブ名。
BIGTABLE_INSTANCE_ID: Cloud Bigtable インスタンス ID
BIGTABLE_TABLE_ID: Cloud Bigtable テーブル名
CASSANDRA_HOSTS: Apache Cassandra のホストリスト。複数のホストがある場合は、カンマをエスケープするための手順に従ってください
CASSANDRA_KEYSPACE: テーブルが配置されている Apache Cassandra キースペース
CASSANDRA_TABLE: 移行する必要がある Apache Cassandra テーブル

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Cassandra_To_Cloud_Bigtable

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Cassandra_To_Cloud_Bigtable
{
   "jobName": "JOB_NAME",
   "parameters": {
       "bigtableProjectId": "PROJECT_ID",
       "bigtableInstanceId": "BIGTABLE_INSTANCE_ID",
       "bigtableTableId": "BIGTABLE_TABLE_ID",
       "cassandraHosts": "CASSANDRA_HOSTS",
       "cassandraKeyspace": "CASSANDRA_KEYSPACE",
       "cassandraTable": "CASSANDRA_TABLE"
   },
   "environment": { "zone": "us-central1-f" }
}

以下を置き換えます。

PROJECT_ID: Cloud Bigtable が配置されているプロジェクト ID
JOB_NAME: 任意のジョブ名。
BIGTABLE_INSTANCE_ID: Cloud Bigtable インスタンス ID
BIGTABLE_TABLE_ID: Cloud Bigtable テーブル名
CASSANDRA_HOSTS: Apache Cassandra のホストリスト。複数のホストがある場合は、カンマをエスケープするための手順に従ってください
CASSANDRA_KEYSPACE: テーブルが配置されている Apache Cassandra キースペース
CASSANDRA_TABLE: 移行する必要がある Apache Cassandra テーブル

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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.datastax.driver.core.Session;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.io.cassandra.CassandraIO;
import org.apache.beam.sdk.io.cassandra.Mapper;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
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;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.Row;

/**
 * This Dataflow Template performs a one off copy of one table from Apache Cassandra to Cloud
 * Bigtable. It is designed to require minimal configuration and aims to replicate the table
 * structure in Cassandra as closely as possible in Cloud Bigtable. To run the pipeline go to <link>
 * enter the required configuration and press <run>
 *
 * <p>The minimum required configuration required to run the pipeline is:
 *
 * <ul>
 *   <li><b>cassandraHosts:</b> The hosts of the Cassandra nodes in a comma separated value list.
 *   <li><b>cassandraPort:</b> The tcp port where Cassandra can be reached on the nodes.
 *   <li><b>cassandraKeyspace:</b> The Cassandra keyspace where the table is located.
 *   <li><b>cassandraTable:</b> The Cassandra table to be copied.
 *   <li><b>bigtableProjectId:</b> The Project ID of the Bigtable instance where the Cassandra table
 *       should be copied.
 *   <li><b>bigtableInstanceId:</b> The Bigtable Instance ID where the Cassandra table should be
 *       copied.
 *   <li><b>bigtableTableId:</b> The name of the Bigtable table where the Cassandra table should be
 *       copied.
 * </ul>
 */
final class CassandraToBigtable {

  public interface Options extends PipelineOptions {

    @Description("Cassandra hosts to read from")
    ValueProvider<String> getCassandraHosts();

    @SuppressWarnings("unused")
    void setCassandraHosts(ValueProvider<String> hosts);

    @Description("Cassandra port")
    @Default.Integer(9042)
    ValueProvider<Integer> getCassandraPort();

    @SuppressWarnings("unused")
    void setCassandraPort(ValueProvider<Integer> port);

    @Description("Cassandra keyspace to read from")
    ValueProvider<String> getCassandraKeyspace();

    @SuppressWarnings("unused")
    void setCassandraKeyspace(ValueProvider<String> keyspace);

    @Description("Cassandra table to read from")
    ValueProvider<String> getCassandraTable();

    @SuppressWarnings("unused")
    void setCassandraTable(ValueProvider<String> cassandraTable);

    @Description("Project where the destination Bigtable instance is located")
    ValueProvider<String> getBigtableProjectId();

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

    @Description("Target Bigtable instance id")
    ValueProvider<String> getBigtableInstanceId();

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

    @Description("Target Bigtable table")
    ValueProvider<String> getBigtableTableId();

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

    @Description("Default Column Family")
    @Default.String("default")
    ValueProvider<String> getDefaultColumnFamily();

    @SuppressWarnings("unused")
    void setDefaultColumnFamily(ValueProvider<String> defaultColumnFamily);

    @Description("RowKeySeparator")
    @Default.String("#")
    ValueProvider<String> getRowKeySeparator();

    @SuppressWarnings("unused")
    void setRowKeySeparator(ValueProvider<String> rowKeySeparator);

    @Description(
        "If true, a large row is split into multiple MutateRows requests. When a row is"
            + " split across requests, updates are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

  /**
   * Runs a pipeline to copy one Cassandra table to Cloud Bigtable.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {

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

    // Split the Cassandra Hosts value provider into a list value provider.
    ValueProvider.NestedValueProvider<List<String>, String> hosts =
        ValueProvider.NestedValueProvider.of(
            options.getCassandraHosts(),
            (SerializableFunction<String, List<String>>) value -> Arrays.asList(value.split(",")));

    Pipeline p = Pipeline.create(PipelineUtils.tweakPipelineOptions(options));

    // Create a factory method to inject the CassandraRowMapperFn to allow custom type mapping.
    SerializableFunction<Session, Mapper> cassandraObjectMapperFactory =
        new CassandraRowMapperFactory(options.getCassandraTable(), options.getCassandraKeyspace());

    CassandraIO.Read<Row> source =
        CassandraIO.<Row>read()
            .withHosts(hosts)
            .withPort(options.getCassandraPort())
            .withKeyspace(options.getCassandraKeyspace())
            .withTable(options.getCassandraTable())
            .withMapperFactoryFn(cassandraObjectMapperFactory)
            .withEntity(Row.class)
            .withCoder(SerializableCoder.of(Row.class));

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

    p.apply("Read from Cassandra", source)
        .apply(
            "Convert Row",
            ParDo.of(
                BeamRowToBigtableFn.createWithSplitLargeRows(
                    options.getRowKeySeparator(),
                    options.getDefaultColumnFamily(),
                    options.getSplitLargeRows(),
                    BeamRowToBigtableFn.MAX_MUTATION_PER_REQUEST)))
        .apply("Write to Bigtable", sink);
    p.run();
  }
}

Apache Hive to BigQuery

Apache Hive to BigQuery テンプレートは、Apache Hive テーブルから読み取り、BigQuery テーブルに書き込むバッチパイプラインです。

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

パイプラインを実行する前に、ターゲットの BigQuery テーブルが存在する。
Dataflow ワーカーと Apache Hive ノードの間にネットワーク接続が存在する。
Dataflow ワーカーと Apache Thrift サーバーノードの間にネットワーク接続が存在する。
パイプラインを実行する前に、BigQuery データセットが存在する。

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

パラメータ	説明
`metastoreUri`	Apache Thrift サーバー URI（`thrift://thrift-server-host:port` など）。
`hiveDatabaseName`	エクスポートするテーブルを含む Apache Hive データベース名。
`hiveTableName`	エクスポートする Apache Hive テーブル名。
`outputTableSpec`	BigQuery 出力テーブルの場所。`<my-project>:<my-dataset>.<my-table>` の形式で指定します。
`hivePartitionCols`	（省略可）Apache Hive パーティション列のカンマ区切りのリスト。
`filterString`	（省略可）入力 Apache Hive テーブルでフィルタ文字列。
`partitionType`	（省略可）BigQuery のパーティションタイプ。現在、`Time` だけがサポートされています。
`partitionCol`	（省略可）出力 BigQuery テーブルのパーティション列名。

Apache Hive to BigQuery テンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Apache Hive to BigQuery template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/Hive_To_BigQuery

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/latest/Hive_To_BigQuery \
    --parameters\
metastoreUri=METASTORE_URI,\
hiveDatabaseName=HIVE_DATABASE_NAME,\
hiveTableName=HIVE_TABLE_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
hivePartitionCols=HIVE_PARTITION_COLS,\
filterString=FILTER_STRING,\
partitionType=PARTITION_TYPE,\
partitionCol=PARTITION_COL

以下を置き換えます。

PROJECT_ID: BigQuery が配置されているプロジェクト ID
JOB_NAME: 任意のジョブ名
DATASET: BigQuery データセット
TABLE_NAME: BigQuery テーブル名
METASTORE_URI: Apache Thrift サーバー URI
HIVE_DATABASE_NAME: エクスポートするテーブルを含む Apache Hive データベース名
HIVE_TABLE_NAME: エクスポートする Apache Hive テーブル名
HIVE_PARTITION_COLS: Apache Hive パーティション列のカンマ区切りのリスト。
FILTER_STRING: Apache Hive 入力テーブルのフィルタ文字列
PARTITION_TYPE: BigQuery のパーティションタイプ
PARTITION_COL: BigQuery パーティション列の名前

API

REST API からの実行

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

gs://dataflow-templates/VERSION/Hive_To_BigQuery

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/templates:launch?gcsPath=gs://dataflow-templates/latest/Hive_To_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "metastoreUri": "METASTORE_URI",
       "hiveDatabaseName": "HIVE_DATABASE_NAME",
       "hiveTableName": "HIVE_TABLE_NAME",
       "outputTableSpec": "PROJECT_ID:DATASET.TABLE_NAME",
       "hivePartitionCols": "HIVE_PARTITION_COLS",
       "filterString": "FILTER_STRING",
       "partitionType": "PARTITION_TYPE",
       "partitionCol": "PARTITION_COL"
   },
   "environment": { "zone": "us-central1-f" }
}

以下を置き換えます。

PROJECT_ID: BigQuery が配置されているプロジェクト ID
JOB_NAME: 任意のジョブ名
DATASET: BigQuery データセット
TABLE_NAME: BigQuery テーブル名
METASTORE_URI: Apache Thrift サーバー URI
HIVE_DATABASE_NAME: エクスポートするテーブルを含む Apache Hive データベース名
HIVE_TABLE_NAME: エクスポートする Apache Hive テーブル名
HIVE_PARTITION_COLS: Apache Hive パーティション列のカンマ区切りのリスト。
FILTER_STRING: Apache Hive 入力テーブルのフィルタ文字列
PARTITION_TYPE: BigQuery のパーティションタイプ
PARTITION_COL: BigQuery パーティション列の名前

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2020 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.v2.templates;

import com.google.api.services.bigquery.model.TimePartitioning;
import com.google.cloud.teleport.v2.options.BigQueryCommonOptions.WriteOptions;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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.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.io.gcp.bigquery.BigQueryUtils;
import org.apache.beam.sdk.io.hcatalog.HCatToRow;
import org.apache.beam.sdk.io.hcatalog.HCatalogIO;
import org.apache.beam.sdk.io.hdfs.HadoopFileSystemOptions;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.apache.hadoop.conf.Configuration;

/**
 * The {@link HiveToBigQuery} pipeline is a batch pipeline which ingests data
 * from Hive and outputs the resulting records to BigQuery.
 *
 */

public final class HiveToBigQuery {

  /**
   * The {@link Options} class provides the custom execution options passed by the executor at the
   * command-line.
   */
  public interface Options extends DataflowPipelineOptions, HadoopFileSystemOptions, WriteOptions {
    @Description("hive metastore uri")
    @Required
    String getMetastoreUri();

    void setMetastoreUri(String value);

    @Description("Hive database name")
    @Required
    String getHiveDatabaseName();

    void setHiveDatabaseName(String value);

    @Description("Hive table name")
    @Required
    String getHiveTableName();

    void setHiveTableName(String value);

    @Description("comma separated list of the columns of Hive partition columns")
    List<java.lang.String> getHivePartitionCols();

    void setHivePartitionCols(List<java.lang.String> value);

    @Description("the filter details")
    String getFilterString();

    void setFilterString(String value);
  }

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for it's execution to finish. If blocking execution is required, use the {@link
   * HiveToBigQuery#run(Options)} method to start the pipeline and invoke
   * {@code
   * result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) {
    Options options =
            PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    run(options);
  }

  /**
   * Configures the HCatalog source.
   *
   * @param options The execution options.
   * @param configProperties The configuration properties like metastore URI
   * @return hiveRead Configured HCatalogIO.Read method
   */
  private static HCatalogIO.Read getHCatalogIORead(Options options, Map<String, String> configProperties) {
    HCatalogIO.Read hiveRead = HCatalogIO.read()
        .withConfigProperties(configProperties)
        .withDatabase(options.getHiveDatabaseName())
        .withTable(options.getHiveTableName());

    if (options.getHivePartitionCols() != null) {
      hiveRead = hiveRead.withPartitionCols(options.getHivePartitionCols());
    }

    if (options.getFilterString() != null) {
      hiveRead = hiveRead.withFilter(options.getFilterString());
    }
    return hiveRead;
  }

  /**
   * Configures the HCatalog source.
   *
   * @param options The execution options.
   * @param configProperties The configuration properties like metastore URI
   * @return bqWrite Configured BigQueryIO.Write method
   */
  private static BigQueryIO.Write getBigQueryIOWrite(Options options, PCollection<Row> rows) {
    BigQueryIO.Write bqWrite = BigQueryIO.<Row>write()
            .to(options.getOutputTableSpec())
            .withSchema(BigQueryUtils.toTableSchema(rows.getSchema()))
            .withFormatFunction(BigQueryUtils.toTableRow())
            .withCreateDisposition(CreateDisposition.valueOf(options.getCreateDisposition()))
            .withWriteDisposition(WriteDisposition.valueOf(options.getWriteDisposition()))
            .withExtendedErrorInfo();

    if (options.getPartitionType() == "Time") {
      bqWrite = bqWrite.withTimePartitioning(new TimePartitioning().setField(options.getPartitionCol()));
    }
    return bqWrite;
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(Options options) {
    // Set Hadoop configurations
    Configuration hadoopConf = new Configuration();
    options.setHdfsConfiguration(Collections.singletonList(hadoopConf));

    Map<String, String> configProperties = new HashMap<String, String>();
    configProperties.put("hive.metastore.uris", options.getMetastoreUri());

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    PCollection<Row> rows =
        pipeline
            /*
             * Step #1: Read hive table rows from Hive.
             */
            .apply(
                "Read from Hive source",
                    HCatToRow.fromSpec(
                        getHCatalogIORead(options, configProperties)));

    /*
     * Step #2: Write table rows out to BigQuery
     */
    rows.apply(
            "Write records to BigQuery", getBigQueryIOWrite(options, rows)
    );
    return pipeline.run();
  }
}

File Format Conversion（Avro、Parquet、CSV）

File Format Conversion テンプレートは、Cloud Storage に格納されたファイルをサポートされている形式から別の形式に変換するバッチパイプラインです。

次の形式変換がサポートされています。

CSV から Avro
CSV から Parquet
Avro から Parquet
Parquet から Avro

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

パイプラインを実行する前に、出力先の Cloud Storage バケットが存在すること。

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

パラメータ	説明
`inputFileFormat`	入力ファイル形式。`[csv, avro, parquet]` のいずれかにする必要があります。
`outputFileFormat`	出力ファイル形式。`[avro, parquet]` のいずれかにする必要があります。
`inputFileSpec`	入力ファイルの Cloud Storage パスのパターン。例: `gs://bucket-name/path/*.csv`
`outputBucket`	出力ファイルを書き込む Cloud Storage フォルダ。このパスはスラッシュで終わる必要があります。例: `gs://bucket-name/output/`
`schema`	Avro スキーマファイルへの Cloud Storage パス。例: `gs://bucket-name/schema/my-schema.avsc`
`containsHeaders`	（省略可）入力 CSV ファイルにはヘッダーレコード（true/false）が含まれています。デフォルト値は `false` です。CSV ファイルを読み込む場合にのみ必要です。
`csvFormat`	（省略可）レコードの解析に使用する CSV 形式の仕様。デフォルト値は `Default` です。詳細については、Apache Commons CSV 形式をご覧ください。
`delimiter`	（省略可）入力 CSV ファイルで使用されるフィールド区切り文字。
`outputFilePrefix`	（省略可）出力ファイルの接頭辞。デフォルト値は `output` です。
`numShards`	（省略可）出力ファイルのシャード数。

File Format Conversion テンプレートの実行

Console

Cloud Console からの実行

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

Dataflow ページに移動

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

[Dataflow テンプレート] プルダウンメニューから the Convert file formats template を選択します。
[ジョブ名] フィールドにジョブ名を入力します。
表示されるパラメータフィールドにパラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

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

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

gs://dataflow-templates/VERSION/flex/File_Format_Conversion

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --template-file-gcs-location=gs://dataflow-templates/latest/flex/File_Format_Conversion \
    --parameters \
inputFileFormat=INPUT_FORMAT,\
outputFileFormat=OUTPUT_FORMAT,\
inputFileSpec=INPUT_FILES,\
schema=SCHEMA,\
outputBucket=OUTPUT_FOLDER

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
INPUT_FORMAT: 入力ファイルのファイル形式。[csv, avro, parquet] のいずれかにする必要があります
OUTPUT_FORMAT: 出力ファイルのファイル形式。[avro, parquet] のいずれかにする必要があります
INPUT_FILES: 入力ファイルのパスのパスパターン
OUTPUT_FOLDER: 出力ファイルを格納する Cloud Storage フォルダ
SCHEMA: Avro スキーマファイルのパス
LOCATION: 実行リージョン（例: us-central1）

API

REST API からの実行

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

gs://dataflow-templates/VERSION/flex/File_Format_Conversion

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputFileFormat": "INPUT_FORMAT",
          "outputFileFormat": "OUTPUT_FORMAT",
          "inputFileSpec": "INPUT_FILES",
          "schema": "SCHEMA",
          "outputBucket": "OUTPUT_FOLDER"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/latest/flex/File_Format_Conversion",
   }
}

以下を置き換えます。

PROJECT_ID: プロジェクト ID
JOB_NAME: 任意のジョブ名。
INPUT_FORMAT: 入力ファイルのファイル形式。[csv, avro, parquet] のいずれかにする必要があります
OUTPUT_FORMAT: 出力ファイルのファイル形式。[avro, parquet] のいずれかにする必要があります
INPUT_FILES: 入力ファイルのパスのパスパターン
OUTPUT_FOLDER: 出力ファイルを格納する Cloud Storage フォルダ
SCHEMA: Avro スキーマファイルのパス
LOCATION: 実行リージョン（例: us-central1）

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

Java

GitHub で表示フィードバック

/*
 * Copyright (C) 2019 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.v2.templates;

import com.google.cloud.teleport.v2.transforms.AvroConverters.AvroOptions;
import com.google.cloud.teleport.v2.transforms.CsvConverters.CsvPipelineOptions;
import com.google.cloud.teleport.v2.transforms.ParquetConverters.ParquetOptions;
import java.io.IOException;
import java.util.EnumMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link FileFormatConversion} pipeline takes in an input file, converts it to a desired format
 * and saves it to Cloud Storage. Supported file transformations are:
 *
 * <ul>
 *   <li>Csv to Avro
 *   <li>Csv to Parquet
 *   <li>Avro to Parquet
 *   <li>Parquet to Avro
 * </ul>
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Input file exists in Google Cloud Storage.
 *   <li>Google Cloud Storage output bucket exists.
 * </ul>
 *
 * <p><b>Example Usage</b>
 *
 * <pre>
 * # Set the pipeline vars
 * PROJECT=my-project
 * BUCKET_NAME=my-bucket
 *
 * # Set containerization vars
 * IMAGE_NAME=my-image-name
 * TARGET_GCR_IMAGE=gcr.io/${PROJECT}/${IMAGE_NAME}
 * BASE_CONTAINER_IMAGE=my-base-container-image
 * BASE_CONTAINER_IMAGE_VERSION=my-base-container-image-version
 * APP_ROOT=/path/to/app-root
 * COMMAND_SPEC=/path/to/command-spec
 *
 * # Set vars for execution
 * export INPUT_FILE_FORMAT=Csv
 * export OUTPUT_FILE_FORMAT=Avro
 * export AVRO_SCHEMA_PATH=gs://path/to/avro/schema
 * export HEADERS=false
 * export DELIMITER=","
 *
 * # Build and upload image
 * mvn clean package \
 * -Dimage=${TARGET_GCR_IMAGE} \
 * -Dbase-container-image=${BASE_CONTAINER_IMAGE} \
 * -Dbase-container-image.version=${BASE_CONTAINER_IMAGE_VERSION} \
 * -Dapp-root=${APP_ROOT} \
 * -Dcommand-spec=${COMMAND_SPEC}
 *
 * # Create an image spec in GCS that contains the path to the image
 * {
 *    "docker_template_spec": {
 *       "docker_image": $TARGET_GCR_IMAGE
 *     }
 *  }
 *
 * # Execute template:
 * API_ROOT_URL="https://dataflow.googleapis.com"
 * TEMPLATES_LAUNCH_API="${API_ROOT_URL}/v1b3/projects/${PROJECT}/templates:launch"
 * JOB_NAME="csv-to-avro-`date +%Y%m%d-%H%M%S-%N`"
 *
 * time curl -X POST -H "Content-Type: application/json"     \
 *     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 *     "${TEMPLATES_LAUNCH_API}"`
 *     `"?validateOnly=false"`
 *     `"&dynamicTemplate.gcsPath=${BUCKET_NAME}/path/to/image-spec"`
 *     `"&dynamicTemplate.stagingLocation=${BUCKET_NAME}/staging" \
 *     -d '
 *      {
 *       "jobName":"'$JOB_NAME'",
 *       "parameters": {
 *            "inputFileFormat":"'$INPUT_FILE_FORMAT'",
 *            "outputFileFormat":"'$OUTPUT_FILE_FORMAT'",
 *            "inputFileSpec":"'$BUCKET_NAME/path/to/input-file'",
 *            "outputBucket":"'$BUCKET_NAME/path/to/output-location/'",
 *            "containsHeaders":"'$HEADERS'",
 *            "schema":"'$AVRO_SCHEMA_PATH'",
 *            "outputFilePrefix":"output-file",
 *            "numShards":"3",
 *            "delimiter":"'$DELIMITER'"
 *         }
 *       }
 *      '
 * </pre>
 */
public class FileFormatConversion {

  /** Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(FileFormatConversionFactory.class);

  private static EnumMap<ValidFileFormats, String> validFileFormats =
      new EnumMap<ValidFileFormats, String>(ValidFileFormats.class);

  /**
   * The {@link FileFormatConversionOptions} provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface FileFormatConversionOptions
      extends PipelineOptions, CsvPipelineOptions, AvroOptions, ParquetOptions {
    @Description("Input file format.")
    @Required
    String getInputFileFormat();

    void setInputFileFormat(String inputFileFormat);

    @Description("Output file format.")
    @Required
    String getOutputFileFormat();

    void setOutputFileFormat(String outputFileFormat);
  }

  /** The {@link ValidFileFormats} enum contains all valid file formats. */
  public enum ValidFileFormats {
    CSV,
    AVRO,
    PARQUET
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    FileFormatConversionOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(FileFormatConversionOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   * @throws RuntimeException thrown if incorrect file formats are passed.
   */
  public static PipelineResult run(FileFormatConversionOptions options) {
    String inputFileFormat = options.getInputFileFormat().toUpperCase();
    String outputFileFormat = options.getOutputFileFormat().toUpperCase();

    validFileFormats.put(ValidFileFormats.CSV, "CSV");
    validFileFormats.put(ValidFileFormats.AVRO, "AVRO");
    validFileFormats.put(ValidFileFormats.PARQUET, "PARQUET");

    try {
      if (inputFileFormat.equals(outputFileFormat)) {
        LOG.error("Input and output file format cannot be the same.");
        throw new IOException();
      }
      if (!validFileFormats.containsValue(inputFileFormat)
          || !validFileFormats.containsValue(outputFileFormat)) {
        LOG.error("Invalid input or output file format.");
        throw new IOException();
      }
    } catch (IOException e) {
      throw new RuntimeException("Provide correct input/output file format.");
    }

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    pipeline.apply(
        inputFileFormat + " to " + outputFileFormat,
        FileFormatConversionFactory.FileFormat.newBuilder()
            .setOptions(options)
            .setInputFileFormat(inputFileFormat)
            .setOutputFileFormat(outputFileFormat)
            .build());

    return pipeline.run();
  }
}

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

BigQuery to Cloud Storage TFRecords

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

BigQuery to Cloud Storage TFRecord ファイル テンプレートの実行

Console

gcloud

API

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

Java

Python

BigQuery の Parquet へのエクスポート（Storage API 経由）

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

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

Console

gcloud

API

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

Java

Cloud Bigtable to Cloud Storage Avro

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

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

Console

gcloud

API

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

Java

Cloud Bigtable to Cloud Storage Parquet

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

Cloud Bigtable to Cloud Storage Parquet ファイル テンプレートの実行

Console

gcloud

API

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

Java

Cloud Bigtable to Cloud Storage SequenceFile

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

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

Console

gcloud

API

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

Java: SDK 2.x

Datastore to Cloud Storage Text

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

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

Console

gcloud

API

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

Java

Cloud Spanner to Cloud Storage Avro

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

Cloud Spanner to Avro Files on Cloud Storage テンプレートの実行

Console

gcloud

API

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

Java

Cloud Spanner to Cloud Storage Text

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

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

Console

gcloud

API

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

Java

Cloud Storage Avro to Cloud Bigtable

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

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

Console

gcloud

API

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

Java

Cloud Storage Avro to Cloud Spanner

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

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

Console

gcloud

API

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

BigQuery to Cloud Storage TFRecord ファイルテンプレートの実行

Cloud Bigtable to Cloud Storage Parquet ファイルテンプレートの実行