Modelli batch Dataflow forniti da Google

Google fornisce un set di modelli open source Dataflow.

Questi modelli Dataflow possono aiutarti a risolvere attività di dati di grandi dimensioni, tra cui l'importazione, l'esportazione e il backup dei dati, il ripristino dei dati e le operazioni API collettive, il tutto senza utilizzare un ambiente di sviluppo dedicato. I modelli sono basati su Apache Beam e utilizzano Dataflow per trasformare i dati.

Per informazioni generali sui modelli, consulta Modelli di Dataflow. Per un elenco di tutti i modelli forniti da Google, consulta la Guida introduttiva ai modelli forniti da Google.

Questa guida documenta i modelli batch.

Da TFRecord da BigQuery a Cloud Storage

Il modello TFRecords da BigQuery a Cloud Storage è una pipeline che legge i dati da una query BigQuery e li scrive in un bucket Cloud Storage in formato TFRecord. Puoi specificare le suddivisioni della percentuale di addestramento, test e convalida. Per impostazione predefinita, la suddivisione è 1 o 100% per il set di addestramento e 0 o 0% per i set di test e convalida. Quando imposti la suddivisione del set di dati, la somma dei valori di addestramento, test e convalida deve essere pari o inferiore al 100% (ad esempio 0,6+0,2+0,2). Dataflow determina automaticamente il numero ottimale di shard per ogni set di dati di output.

Requisiti per questa pipeline:

• Devono esistere la tabella e il set di dati BigQuery.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline. Non è necessario preesistere e vengono generate automaticamente le sottodirectory di addestramento, test e convalida.

Parametri del modello

Parametro	Descrizione
readQuery	Una query BigQuery BigQuery che estrae i dati dall'origine. Ad esempio, select * from dataset1.sample_table.
outputDirectory	Il prefisso del percorso di Cloud Storage di primo livello a cui scrivere i file TFRecord per l'addestramento, il test e la convalida. Ad esempio, gs://mybucket/output. Le sottodirectory per i file TFRecord di addestramento, test e convalida generati vengono generate automaticamente da outputDirectory. Ad esempio, gs://mybucket/output/train
trainingPercentage	(Facoltativo) La percentuale di dati delle query assegnati ai file TFRecord di addestramento. Il valore predefinito è 1 o 100%.
testingPercentage	(Facoltativo) La percentuale di dati delle query assegnati ai test dei file TFRecord. Il valore predefinito è 0 o 0%.
validationPercentage	(Facoltativo) La percentuale di dati delle query assegnati ai file TFRecord di convalida. Il valore predefinito è 0 o 0%.
outputSuffix	(Facoltativo) Il suffisso per i file TFRecord di addestramento, test e convalida. Il valore predefinito è .tfrecord.

Esecuzione del modello di file TFRecord di BigQuery in Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the BigQuery to TFRecords template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• READ_QUERY: la query BigQuery da eseguire
• OUTPUT_DIRECTORY: il prefisso del percorso di Cloud Storage per i set di dati di output
• TRAINING_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di addestramento
• TESTING_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di test
• VALIDATION_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di convalida
• OUTPUT_FILENAME_SUFFIX: il suffisso di file di registrazione TensorFlow preferito

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• READ_QUERY: la query BigQuery da eseguire
• OUTPUT_DIRECTORY: il prefisso del percorso di Cloud Storage per i set di dati di output
• TRAINING_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di addestramento
• TESTING_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di test
• VALIDATION_PERCENTAGE: la percentuale decimale suddivisa per il set di dati di convalida
• OUTPUT_FILENAME_SUFFIX: il suffisso di file di registrazione TensorFlow preferito

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2019 Google LLC
 *
 * 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.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.templates.BigQueryToTFRecord.Options;
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.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.
 */
@Template(
    name = "Cloud_BigQuery_to_GCS_TensorFlow_Records",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery to TensorFlow Records",
    description =
        "A pipeline that reads rows from BigQuery and writes them as TFRecords in Cloud Storage. (NOTE: Nested BigQuery columns are currently not supported and should be unnested within the SQL query.)",
    optionsClass = Options.class,
    optionsOrder = {BigQueryReadOptions.class, Options.class},
    contactInformation = "https://cloud.google.com/support")
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()) {
      Object fieldValue = record.get(field.getName());
      if (fieldValue != null) {
        Feature feature = buildFeature(fieldValue, 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 {

    @TemplateParameter.GcsWriteFolder(
        order = 1,
        description = "Output Cloud Storage directory.",
        helpText = "Cloud Storage directory to store output TFRecord files.",
        example = "gs://your-bucket/your-path")
    ValueProvider<String> getOutputDirectory();

    void setOutputDirectory(ValueProvider<String> outputDirectory);

    @TemplateParameter.Text(
        order = 2,
        optional = true,
        regexes = {"^[A-Za-z_0-9.]*"},
        description = "The output suffix for TFRecord files",
        helpText = "File suffix to append to TFRecord files. Defaults to .tfrecord")
    @Default.String(".tfrecord")
    ValueProvider<String> getOutputSuffix();

    void setOutputSuffix(ValueProvider<String> outputSuffix);

    @TemplateParameter.Text(
        order = 3,
        optional = true,
        regexes = {"(^\\.[1-9]*$)|(^[01]*)"},
        description = "Percentage of data to be in the training set ",
        helpText = "Defaults to 1 or 100%. Should be decimal between 0 and 1 inclusive")
    @Default.Float(1)
    ValueProvider<Float> getTrainingPercentage();

    void setTrainingPercentage(ValueProvider<Float> trainingPercentage);

    @TemplateParameter.Text(
        order = 4,
        optional = true,
        regexes = {"(^\\.[1-9]*$)|(^[01]*)"},
        description = "Percentage of data to be in the testing set ",
        helpText = "Defaults to 0 or 0%. Should be decimal between 0 and 1 inclusive")
    @Default.Float(0)
    ValueProvider<Float> getTestingPercentage();

    void setTestingPercentage(ValueProvider<Float> testingPercentage);

    @TemplateParameter.Text(
        order = 5,
        optional = true,
        regexes = {"(^\\.[1-9]*$)|(^[01]*)"},
        description = "Percentage of data to be in the validation set ",
        helpText = "Defaults to 0 or 0%. Should be decimal between 0 and 1 inclusive")
    @Default.Float(0)
    ValueProvider<Float> getValidationPercentage();

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

Esportazione in BigQuery in Parquet (tramite API Storage)

Il modello BigQuery Export in Parquet è una pipeline batch che legge i dati di una tabella BigQuery e li scrive in un bucket Cloud Storage in formato Parquet. Questo modello utilizza l'API BigQuery Storage per esportare i dati.

Requisiti per questa pipeline:

• La tabella BigQuery di input deve esistere prima dell'esecuzione della pipeline.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
tableRef	La posizione della tabella di input di BigQuery. Ad esempio, <my-project>:<my-dataset>.<my-table>.
bucket	La cartella Cloud Storage in cui scrivere i file Parquet. Ad esempio, gs://mybucket/exports.
numShards	(Facoltativo) Il numero di shard del file di output. Il valore predefinito è 1.
fields	(Facoltativo) Un elenco di campi separati da virgole da selezionare nella tabella BigQuery di input.

Esecuzione del modello Parquet da BigQuery a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the BigQuery export to Parquet (via Storage API) template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli flex, devi disporre di Google Cloud CLI 284.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGQUERY_TABLE: il nome della tua tabella BigQuery
• OUTPUT_DIRECTORY: la tua cartella Cloud Storage per i file di output
• NUM_SHARDS: numero desiderato di shard del file di output
• FIELDS: l'elenco di campi separati da virgole che puoi selezionare dalla tabella BigQuery di input

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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/VERSION/flex/BigQuery_to_Parquet",
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGQUERY_TABLE: il nome della tua tabella BigQuery
• OUTPUT_DIRECTORY: la tua cartella Cloud Storage per i file di output
• NUM_SHARDS: numero desiderato di shard del file di output
• FIELDS: l'elenco di campi separati da virgole che puoi selezionare dalla tabella BigQuery di input

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2019 Google LLC
 *
 * 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 com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.templates.BigQueryToParquet.BigQueryToParquetOptions;
import com.google.common.base.Splitter;
import com.google.common.base.Strings;
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.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 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>
 */
@Template(
    name = "BigQuery_to_Parquet",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery export to Parquet (via Storage API)",
    description =
        "A pipeline to export a BigQuery table into Parquet files using the BigQuery Storage API.",
    optionsClass = BigQueryToParquetOptions.class,
    flexContainerName = "bigquery-to-parquet",
    contactInformation = "https://cloud.google.com/support")
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 {
    @TemplateParameter.BigQueryTable(
        order = 1,
        description = "BigQuery table to export",
        helpText = "BigQuery table location to export in the format <project>:<dataset>.<table>.",
        example = "your-project:your-dataset.your-table-name")
    @Required
    String getTableRef();

    void setTableRef(String tableRef);

    @TemplateParameter.GcsWriteFile(
        order = 2,
        description = "Output Cloud Storage file(s)",
        helpText = "Path and filename prefix for writing output files.",
        example = "gs://your-bucket/export/")
    @Required
    String getBucket();

    void setBucket(String bucket);

    @TemplateParameter.Integer(
        order = 3,
        optional = true,
        description = "Maximum output shards",
        helpText =
            "The maximum number of output shards produced when writing. A higher number of shards"
                + " means higher throughput for writing to Cloud Storage, but potentially higher"
                + " data aggregation cost across shards when processing output Cloud Storage"
                + " files.")
    @Default.Integer(0)
    Integer getNumShards();

    void setNumShards(Integer numShards);

    @TemplateParameter.Text(
        order = 4,
        optional = true,
        description = "List of field names",
        helpText = "Comma separated list of fields to select from the table.")
    String getFields();

    void setFields(String fields);

    @TemplateParameter.Text(
        order = 5,
        optional = true,
        description = "Row restrictions/filter.",
        helpText =
            "Read only rows which match the specified filter, which must be a SQL expression"
                + " compatible with Google standard SQL"
                + " (https://cloud.google.com/bigquery/docs/reference/standard-sql). If no value is"
                + " specified, then all rows are returned.")
    String getRowRestriction();

    void setRowRestriction(String restriction);
  }

  /**
   * 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) {
    UncaughtExceptionLogger.register();

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

    // Add row restrictions/filter if any.
    if (!Strings.isNullOrEmpty(options.getRowRestriction())) {
      readFromBQ = readFromBQ.withRowRestriction(options.getRowRestriction());
    }

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

Da BigQuery a Elasticsearch

beta

Questa funzionalità è regolata dai Termini delle offerte pre-GA dei Termini di servizio di Google Cloud. Le funzionalità pre-GA potrebbero avere supporto limitato e le modifiche a queste funzionalità potrebbero non essere compatibili con altre versioni pre-GA. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Il modello BigQuery in Elasticsearch è una pipeline in modalità batch che importa i dati da una tabella BigQuery a Elasticsearch come documenti. Il modello può leggere l'intera tabella o leggere record specifici utilizzando una query fornita.

Requisiti per questa pipeline

• Deve esistere la tabella BigQuery di origine.
• Un host Elasticsearch su un'istanza Google Cloud o su Elastic Cloud con Elasticsearch versione 7.0 o successiva e deve essere accessibile dalle macchine worker Dataflow.

Parametri del modello

Parametro	Descrizione
connectionUrl	URL Elasticsearch nel formato https://hostname:[port] o specifica CloudID se utilizzi Elastic Cloud.
apiKey	Chiave API codificata in Base64 utilizzata per l'autenticazione.
index	L'indice Elasticsearch per il quale verranno inviate le richieste, ad esempio my-index.
inputTableSpec	(Facoltativo) Tabella BigQuery da leggere per inserire in Elasticsearch. È necessario indicare la tabella o la query. Ad esempio, projectId:datasetId.tablename.
query	(Facoltativo) Query SQL per l'estrazione di dati da BigQuery. È necessario indicare la tabella o la query.
useLegacySql	(Facoltativo) Impostalo su vero per utilizzare SQL precedente (applicabile solo se fornisci la query). Valore predefinito: false.
batchSize	(Facoltativo) Dimensioni del batch relative al numero di documenti. Valore predefinito: 1000.
batchSizeBytes	(Facoltativo) Dimensioni del batch in numero di byte. Valore predefinito: 5242880 (5 MB).
maxRetryAttempts	(Facoltativo) Numero massimo di nuovi tentativi, deve essere maggiore di 0. Valore predefinito: no retries.
maxRetryDuration	(Facoltativo) La durata massima dei nuovi tentativi in millisecondi deve essere maggiore o uguale a 0. Valore predefinito: no retries.
propertyAsIndex	(Facoltativo) Una proprietà nel documento che viene indicizzata il cui valore specificherà i metadati _index da includere nel documento nella richiesta collettiva (ha la precedenza su un _index UDF). Valore predefinito: nessuno.
propertyAsId	(Facoltativo) Una proprietà nel documento che viene indicizzata il cui valore specificherà i metadati _id da includere nel documento nella richiesta collettiva (ha la precedenza su un _id UDF). Valore predefinito: nessuno.
javaScriptIndexFnGcsPath	(Facoltativo) Percorso di Cloud Storage dell'origine JavaScript dell'UDF per una funzione che specifica i metadati _index da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptIndexFnName	(Facoltativo) Nome della funzione JavaScript dell'UDF per la funzione che specifica i metadati _index da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptIdFnGcsPath	(Facoltativo) Percorso di Cloud Storage dell'origine JavaScript dell'UDF per una funzione che specifica i metadati _id da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptIdFnName	(Facoltativo) Nome della funzione JavaScript dell'UDF per la funzione che specifica i metadati _id da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptTypeFnGcsPath	(Facoltativo) Percorso di Cloud Storage dell'origine JavaScript dell'UDF per una funzione che specifica i metadati _type da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptTypeFnName	(Facoltativo) Nome della funzione JavaScript dell'UDF per la funzione che specifica i metadati _type da includere nel documento nella richiesta collettiva. Valore predefinito: nessuno.
javaScriptIsDeleteFnGcsPath	(Facoltativo) Il percorso Cloud Storage dell'origine UDF JavaScript della funzione che determina se il documento deve essere eliminato anziché inserito o aggiornato. La funzione deve restituire il valore stringa "true" o "false". Valore predefinito: nessuno.
javaScriptIsDeleteFnName	(Facoltativo) Nome della funzione JavaScript di UDF per la funzione che determina se il documento deve essere eliminato anziché inserito o aggiornato. La funzione deve restituire il valore stringa "true" o "false". Valore predefinito: nessuno.
usePartialUpdate	(Facoltativo) Indica se utilizzare aggiornamenti parziali (aggiornare invece di creare o indicizzare i documenti parziali) con le richieste Elasticsearch. Valore predefinito: false.
bulkInsertMethod	(Facoltativo) Indica se utilizzare INDEX (indice, possibilità di aumento) o CREATE (creazione, errori su _id duplicato) con richieste collettive Elasticsearch. Valore predefinito: CREATE.

Esecuzione del modello BigQuery to Elasticsearch

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the BigQuery to Elasticsearch template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli flex, devi disporre di Google Cloud CLI 284.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/BigQuery_to_Elasticsearch \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• CONNECTION_URL: l'URL dell'Elasticsearch.
• APIKEY: la chiave API codificata in base64 per l'autenticazione.
• INDEX: il tuo indice Elasticsearch.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/BigQuery_to_Elasticsearch",
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• CONNECTION_URL: l'URL dell'Elasticsearch.
• APIKEY: la chiave API codificata in base64 per l'autenticazione.
• INDEX: il tuo indice Elasticsearch.

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2021 Google LLC
 *
 * 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.elasticsearch.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.BigQueryToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.ReadBigQuery;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.TableRowToJsonFn;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ParDo;

/**
 * The {@link BigQueryToElasticsearch} pipeline exports data from a BigQuery table to Elasticsearch.
 *
 * <p>Please refer to <b><a href=
 * "https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/master/v2/googlecloud-to-elasticsearch/docs/BigQueryToElasticsearch/README.md">
 * README.md</a></b> for further information.
 */
@Template(
    name = "BigQuery_to_Elasticsearch",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery to Elasticsearch",
    description =
        "A pipeline which sends BigQuery records into an Elasticsearch instance as json documents.",
    optionsClass = BigQueryToElasticsearchOptions.class,
    flexContainerName = "bigquery-to-elasticsearch",
    contactInformation = "https://cloud.google.com/support")
public class BigQueryToElasticsearch {
  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    BigQueryToElasticsearchOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigQueryToElasticsearchOptions.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(BigQueryToElasticsearchOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);
    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Create json string from Table Row.
     *        3) Write records to Elasticsearch.
     *
     *
     * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
     */
    pipeline
        .apply(
            "ReadFromBigQuery",
            ReadBigQuery.newBuilder()
                .setOptions(options.as(BigQueryToElasticsearchOptions.class))
                .build())

        /*
         * Step #2: Convert table rows to JSON documents.
         */
        .apply("TableRowsToJsonDocument", ParDo.of(new TableRowToJsonFn()))

        /*
         * Step #3: Write converted records to Elasticsearch
         */
        .apply(
            "WriteToElasticsearch",
            WriteToElasticsearch.newBuilder()
                .setOptions(options.as(BigQueryToElasticsearchOptions.class))
                .build());

    return pipeline.run();
  }
}

Da BigQuery a MongoDB

beta

Questa funzionalità è regolata dai Termini delle offerte pre-GA dei Termini di servizio di Google Cloud. Le funzionalità pre-GA potrebbero avere supporto limitato e le modifiche a queste funzionalità potrebbero non essere compatibili con altre versioni pre-GA. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Il modello BigQuery in MongoDB è una pipeline batch che legge le righe da BigQuery e le scrive in MongoDB sotto forma di documenti. Attualmente ogni riga viene archiviata come documento.

Requisiti per questa pipeline

• Deve esistere la tabella BigQuery di origine.
• L'istanza MongoDB di destinazione deve essere accessibile dalle macchine worker di Dataflow.

Parametri del modello

Parametro	Descrizione
mongoDbUri	URI di connessione MongoDB nel formato mongodb+srv://:@.
database	Database in MongoDB per archiviare la raccolta. Ad esempio: my-db.
collection	Nome della raccolta nel database MongoDB. Ad esempio: my-collection.
inputTableSpec	Tabella BigQuery da cui leggere. Ad esempio, bigquery-project:dataset.input_table.

Esecuzione del modello BigQuery in MongoDB

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the BigQuery to MongoDB template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli flex, devi disporre di Google Cloud CLI 284.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

  gcloud beta dataflow flex-template run JOB_NAME \
      --project=PROJECT_ID \
      --region=REGION_NAME \
      --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/BigQuery_to_MongoDB \
      --parameters \
  inputTableSpec=INPUT_TABLE_SPEC,\
  mongoDbUri=MONGO_DB_URI,\
  database=DATABASE,\
  collection=COLLECTION
  

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery di origine.
• MONGO_DB_URI: l'URI MongoDB.
• DATABASE: il tuo database MongoDB.
• COLLECTION: la tua raccolta MongoDB.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

  POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
  {
     "launch_parameter": {
        "jobName": "JOB_NAME",
        "parameters": {
            "inputTableSpec": "INPUT_TABLE_SPEC",
            "mongoDbUri": "MONGO_DB_URI",
            "database": "DATABASE",
            "collection": "COLLECTION"
        },
        "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/BigQuery_to_MongoDB",
     }
  }

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery di origine.
• MONGO_DB_URI: l'URI MongoDB.
• DATABASE: il tuo database MongoDB.
• COLLECTION: la tua raccolta MongoDB.

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

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

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.mongodb.options.BigQueryToMongoDbOptions.BigQueryReadOptions;
import com.google.cloud.teleport.v2.mongodb.options.BigQueryToMongoDbOptions.MongoDbOptions;
import com.google.cloud.teleport.v2.mongodb.templates.BigQueryToMongoDb.Options;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.mongodb.MongoDbIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.bson.Document;

/**
 * The {@link BigQueryToMongoDb} pipeline is a batch pipeline which reads data from BigQuery and
 * outputs the resulting records to MongoDB.
 */
@Template(
    name = "BigQuery_to_MongoDB",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery to MongoDB",
    description =
        "A batch pipeline which reads data rows from BigQuery and writes them to MongoDB as"
            + " documents.",
    optionsClass = Options.class,
    flexContainerName = "bigquery-to-mongodb",
    contactInformation = "https://cloud.google.com/support")
public class BigQueryToMongoDb {
  /**
   * Options supported by {@link BigQueryToMongoDb}
   *
   * <p>Inherits standard configuration options.
   */
  public interface Options extends PipelineOptions, MongoDbOptions, BigQueryReadOptions {}

  private static class ParseAsDocumentsFn extends DoFn<String, Document> {

    @ProcessElement
    public void processElement(ProcessContext context) {
      context.output(Document.parse(context.element()));
    }
  }

  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

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

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

    pipeline
        .apply(BigQueryIO.readTableRows().withoutValidation().from(options.getInputTableSpec()))
        .apply(
            "bigQueryDataset",
            ParDo.of(
                new DoFn<TableRow, Document>() {
                  @ProcessElement
                  public void process(ProcessContext c) {
                    Document doc = new Document();
                    TableRow row = c.element();
                    row.forEach(
                        (key, value) -> {
                          if (key != "_id") {
                            doc.append(key, value);
                          }
                        });
                    c.output(doc);
                  }
                }))
        .apply(
            MongoDbIO.write()
                .withUri(options.getMongoDbUri())
                .withDatabase(options.getDatabase())
                .withCollection(options.getCollection()));
    pipeline.run();
    return true;
  }
}

Avro da Bigtable a Cloud Storage

Il modello Avro da Bigtable a Cloud Storage è una pipeline che legge i dati da una tabella Bigtable e li scrive in un bucket Cloud Storage in formato Avro. Puoi utilizzare il modello per spostare dati da Bigtable a Cloud Storage.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da esportare.
outputDirectory	Percorso di Cloud Storage in cui sono scritti i dati. Ad esempio, gs://mybucket/somefolder.
filenamePrefix	Il prefisso del nome file Avro. Ad esempio, output-.

Esecuzione del modello di file Avro da Bigtable a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Bigtable to Avro Files on Cloud Storage template .
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• OUTPUT_DIRECTORY: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file di Avro, ad esempio output-

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• OUTPUT_DIRECTORY: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file di Avro, ad esempio output-

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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.bigtable.BigtableToAvro.Options;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
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.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.
 */
@Template(
    name = "Cloud_Bigtable_to_GCS_Avro",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Bigtable to Avro Files in Cloud Storage",
    description =
        "A pipeline which reads in Cloud Bigtable table and writes it to Cloud Storage in Avro format.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
public class BigtableToAvro {

  /** Options for the export pipeline. */
  public interface Options extends PipelineOptions {
    @TemplateParameter.ProjectId(
        order = 1,
        description = "Project ID",
        helpText =
            "The ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from")
    ValueProvider<String> getBigtableProjectId();

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

    @TemplateParameter.Text(
        order = 2,
        regexes = {"[a-z][a-z0-9\\-]+[a-z0-9]"},
        description = "Instance ID",
        helpText = "The ID of the Cloud Bigtable instance that contains the table")
    ValueProvider<String> getBigtableInstanceId();

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

    @TemplateParameter.Text(
        order = 3,
        regexes = {"[_a-zA-Z0-9][-_.a-zA-Z0-9]*"},
        description = "Table ID",
        helpText = "The ID of the Cloud Bigtable table to read")
    ValueProvider<String> getBigtableTableId();

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

    @TemplateParameter.GcsWriteFolder(
        order = 4,
        description = "Output file directory in Cloud Storage",
        helpText =
            "The path and filename prefix for writing output files. Must end with a slash. DateTime formatting is used to parse directory path for date & time formatters.",
        example = "gs://your-bucket/your-path")
    ValueProvider<String> getOutputDirectory();

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

    @TemplateParameter.Text(
        order = 5,
        description = "Avro file prefix",
        helpText = "The prefix of the Avro file name. For example, \"table1-\"")
    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();
    }
  }
}

Parquet da Bigtable a Cloud Storage

Il modello da Bigtable a Cloud Storage è una pipeline che legge i dati da una tabella Bigtable e li scrive in un bucket Cloud Storage in formato Parquet. Puoi utilizzare il modello per spostare dati da Bigtable a Cloud Storage.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da esportare.
outputDirectory	Percorso di Cloud Storage in cui sono scritti i dati. Ad esempio, gs://mybucket/somefolder.
filenamePrefix	Il prefisso del nome file Parquet. Ad esempio, output-.
numShards	Il numero di shard del file di output. Ad esempio: 2.

Esecuzione del modello di file Bigtable da Parquet a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Bigtable to Parquet Files on Cloud Storage template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• OUTPUT_DIRECTORY: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file Parquet, ad esempio output-
• NUM_SHARDS: il numero di file Parquet da visualizzare, ad esempio 1

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• OUTPUT_DIRECTORY: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file Parquet, ad esempio output-
• NUM_SHARDS: il numero di file Parquet da visualizzare, ad esempio 1

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2019 Google LLC
 *
 * 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 com.google.cloud.teleport.bigtable.BigtableToParquet.Options;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
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.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.
 */
@Template(
    name = "Cloud_Bigtable_to_GCS_Parquet",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Bigtable to Parquet Files on Cloud Storage",
    description =
        "A pipeline which reads in Cloud Bigtable table and writes it to Cloud Storage in Parquet format.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
public class BigtableToParquet {

  /** Options for the export pipeline. */
  public interface Options extends PipelineOptions {

    @TemplateParameter.ProjectId(
        order = 1,
        description = "Project ID",
        helpText =
            "The ID of the Google Cloud project of the Cloud Bigtable instance that you want to read data from")
    ValueProvider<String> getBigtableProjectId();

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

    @TemplateParameter.Text(
        order = 2,
        regexes = {"[a-z][a-z0-9\\-]+[a-z0-9]"},
        description = "Instance ID",
        helpText = "The ID of the Cloud Bigtable instance that contains the table")
    ValueProvider<String> getBigtableInstanceId();

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

    @TemplateParameter.Text(
        order = 3,
        regexes = {"[_a-zA-Z0-9][-_.a-zA-Z0-9]*"},
        description = "Table ID",
        helpText = "The ID of the Cloud Bigtable table to export")
    ValueProvider<String> getBigtableTableId();

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

    @TemplateParameter.GcsWriteFolder(
        order = 4,
        description = "Output file directory in Cloud Storage",
        helpText =
            "The path and filename prefix for writing output files. Must end with a slash. DateTime formatting is used to parse directory path for date & time formatters.",
        example = "gs://your-bucket/your-path")
    ValueProvider<String> getOutputDirectory();

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

    @TemplateParameter.Text(
        order = 5,
        description = "Parquet file prefix",
        helpText = "The prefix of the Parquet file name. For example, \"table1-\"")
    @Default.String("output")
    ValueProvider<String> getFilenamePrefix();

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

    @TemplateParameter.Integer(
        order = 6,
        optional = true,
        description = "Maximum output shards",
        helpText =
            "The maximum number of output shards produced when writing. A higher number of "
                + "shards means higher throughput for writing to Cloud Storage, but potentially higher "
                + "data aggregation cost across shards when processing output Cloud Storage files. "
                + "Default value is decided by the runner.")
    @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();
    }
  }
}

SequenzaFile da Bigtable a Cloud Storage

Il modello quifile da Bigtable a Cloud Storage è una pipeline che legge i dati da una tabella Bigtable e li scrive in un bucket Cloud Storage in formato SequenzaFile. Puoi utilizzare il modello per copiare i dati da Bigtable in Cloud Storage.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
bigtableProject	L'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da esportare.
bigtableAppProfileId	L'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione. Se non specifichi un profilo app, Bigtable utilizza il profilo app predefinito dell'istanza.
destinationPath	Percorso di Cloud Storage in cui sono scritti i dati. Ad esempio, gs://mybucket/somefolder.
filenamePrefix	Il prefisso del nome file SequenzaFile. Ad esempio, output-.

Esecuzione del modello SequenzaFile da Bigtable a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Bigtable to SequenceFile Files on Cloud Storage template .
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• APPLICATION_PROFILE_ID: l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• DESTINATION_PATH: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file SequenzaFile, ad esempio output-

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• APPLICATION_PROFILE_ID: l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• DESTINATION_PATH: percorso di Cloud Storage in cui sono scritti i dati, ad esempio gs://mybucket/somefolder
• FILENAME_PREFIX: il prefisso del nome file SequenzaFile, ad esempio output-

Codice sorgente del modello

Java

Il codice sorgente di questo modello si trova nel repository GoogleCloudPlatform/cloud-bigtable-client su GitHub.

Testo da Datastore a Cloud Storage [deprecato]

beta

Questa funzionalità è regolata dai Termini delle offerte pre-GA dei Termini di servizio di Google Cloud. Le funzionalità pre-GA potrebbero avere supporto limitato e le modifiche a queste funzionalità potrebbero non essere compatibili con altre versioni pre-GA. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Questo modello è deprecato e verrà rimosso nel primo trimestre del 2022. Esegui la migrazione al modello Firestore to Cloud Storage Text.

Il modello di testo Datastore da Cloud Storage è una pipeline batch che legge le entità Datastore e le scrive in Cloud Storage come file di testo. Puoi fornire una funzione per elaborare ciascuna entità come stringa JSON. Se non fornisci una tale funzione, ogni riga nel file di output sarà un'entità con serie JSON.

Requisiti per questa pipeline:

Datastore deve essere configurato nel progetto prima di eseguire la pipeline.

Parametri del modello

Parametro	Descrizione
datastoreReadGqlQuery	Una query GQL che specifica quali entità acquisire. Ad esempio, SELECT * FROM MyKind.
datastoreReadProjectId	L'ID del progetto Google Cloud dell'istanza Datastore da cui vuoi leggere i dati.
datastoreReadNamespace	Lo spazio dei nomi delle entità richieste. Per utilizzare lo spazio dei nomi predefinito, lascia vuoto questo parametro.
javascriptTextTransformGcsPath	(Facoltativo) L'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName	(Facoltativo) Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.
textWritePrefix	Il prefisso del percorso di Cloud Storage per specificare dove sono scritti i dati. Ad esempio: gs://mybucket/somefolder/.

Esecuzione del modello di testo da Datastore a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Datastore to Text Files on Cloud Storage template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• BUCKET_NAME: nome del bucket Cloud Storage
• DATASTORE_PROJECT_ID: l'ID del progetto Cloud in cui esiste l'istanza di Datastore
• DATASTORE_KIND: tipo di entità dei Datastore
• DATASTORE_NAMESPACE: lo spazio dei nomi delle entità Datastore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• BUCKET_NAME: nome del bucket Cloud Storage
• DATASTORE_PROJECT_ID: l'ID del progetto Cloud in cui esiste l'istanza di Datastore
• DATASTORE_KIND: tipo di entità dei Datastore
• DATASTORE_NAMESPACE: lo spazio dei nomi delle entità Datastore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.templates.DatastoreToText.DatastoreToTextOptions;
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.FirestoreNestedValueProvider;
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;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * 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
 */
@Template(
    name = "Datastore_to_GCS_Text",
    category = TemplateCategory.BATCH,
    displayName = "Datastore to Text Files on Cloud Storage [Deprecated]",
    description =
        "Batch pipeline. Reads Datastore entities and writes them to Cloud Storage as text files.",
    optionsClass = DatastoreToTextOptions.class,
    skipOptions = {"firestoreReadNamespace", "firestoreReadGqlQuery", "firestoreReadProjectId"},
    contactInformation = "https://cloud.google.com/support")
@Template(
    name = "Firestore_to_GCS_Text",
    category = TemplateCategory.BATCH,
    displayName = "Firestore (Datastore mode) to Text Files on Cloud Storage",
    description =
        "Batch pipeline. Reads Firestore entities and writes them to Cloud Storage as text files.",
    optionsClass = DatastoreToTextOptions.class,
    skipOptions = {"datastoreReadNamespace", "datastoreReadGqlQuery", "datastoreReadProjectId"},
    contactInformation = "https://cloud.google.com/support")
public class DatastoreToText {

  public static ValueProvider<String> selectProvidedInput(
      ValueProvider<String> datastoreInput, ValueProvider<String> firestoreInput) {
    return new FirestoreNestedValueProvider(datastoreInput, firestoreInput);
  }

  /** 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(
                    selectProvidedInput(
                        options.getDatastoreReadGqlQuery(), options.getFirestoreReadGqlQuery()))
                .setProjectId(
                    selectProvidedInput(
                        options.getDatastoreReadProjectId(), options.getFirestoreReadProjectId()))
                .setNamespace(
                    selectProvidedInput(
                        options.getDatastoreReadNamespace(), options.getFirestoreReadNamespace()))
                .build())
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(TextIO.write().to(options.getTextWritePrefix()).withSuffix(".json"));

    pipeline.run();
  }
}

Da Firestore a Cloud Storage

Il modello Firestore to Cloud Storage Text è una pipeline batch che legge le entità Firestore e le scrive in Cloud Storage sotto forma di file di testo. Puoi fornire una funzione per elaborare ciascuna entità come stringa JSON. Se non fornisci una tale funzione, ogni riga nel file di output sarà un'entità con serie JSON.

Requisiti per questa pipeline:

Firestore deve essere configurato nel progetto prima di eseguire la pipeline.

Parametri del modello

Parametro	Descrizione
firestoreReadGqlQuery	Una query GQL che specifica quali entità acquisire. Ad esempio, SELECT * FROM MyKind.
firestoreReadProjectId	L'ID del progetto Google Cloud dell'istanza Firestore da cui vuoi leggere i dati.
firestoreReadNamespace	Lo spazio dei nomi delle entità richieste. Per utilizzare lo spazio dei nomi predefinito, lascia vuoto questo parametro.
javascriptTextTransformGcsPath	(Facoltativo) L'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName	(Facoltativo) Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.
textWritePrefix	Il prefisso del percorso di Cloud Storage per specificare dove sono scritti i dati. Ad esempio: gs://mybucket/somefolder/.

Esecuzione del modello di testo da Firestore a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Firestore to Text Files on Cloud Storage template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Firestore_to_GCS_Text \
    --region REGION_NAME \
    --parameters \
firestoreReadGqlQuery="SELECT * FROM FIRESTORE_KIND",\
firestoreReadProjectId=FIRESTORE_PROJECT_ID,\
firestoreReadNamespace=FIRESTORE_NAMESPACE,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
textWritePrefix=gs://BUCKET_NAME/output/

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• BUCKET_NAME: nome del bucket Cloud Storage
• FIRESTORE_PROJECT_ID: l'ID del progetto Cloud in cui esiste l'istanza di Firestore
• FIRESTORE_KIND: tipo di entità Firestore
• FIRESTORE_NAMESPACE: lo spazio dei nomi delle tue entità Firestore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Firestore_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "parameters": {
       "firestoreReadGqlQuery": "SELECT * FROM FIRESTORE_KIND"
       "firestoreReadProjectId": "FIRESTORE_PROJECT_ID",
       "firestoreReadNamespace": "FIRESTORE_NAMESPACE",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "textWritePrefix": "gs://BUCKET_NAME/output/"
   },
   "environment": { "zone": "us-central1-f" }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• BUCKET_NAME: nome del bucket Cloud Storage
• FIRESTORE_PROJECT_ID: l'ID del progetto Cloud in cui esiste l'istanza di Firestore
• FIRESTORE_KIND: tipo di entità Firestore
• FIRESTORE_NAMESPACE: lo spazio dei nomi delle tue entità Firestore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.templates.DatastoreToText.DatastoreToTextOptions;
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.FirestoreNestedValueProvider;
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;
import org.apache.beam.sdk.options.ValueProvider;

/**
 * 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
 */
@Template(
    name = "Datastore_to_GCS_Text",
    category = TemplateCategory.BATCH,
    displayName = "Datastore to Text Files on Cloud Storage [Deprecated]",
    description =
        "Batch pipeline. Reads Datastore entities and writes them to Cloud Storage as text files.",
    optionsClass = DatastoreToTextOptions.class,
    skipOptions = {"firestoreReadNamespace", "firestoreReadGqlQuery", "firestoreReadProjectId"},
    contactInformation = "https://cloud.google.com/support")
@Template(
    name = "Firestore_to_GCS_Text",
    category = TemplateCategory.BATCH,
    displayName = "Firestore (Datastore mode) to Text Files on Cloud Storage",
    description =
        "Batch pipeline. Reads Firestore entities and writes them to Cloud Storage as text files.",
    optionsClass = DatastoreToTextOptions.class,
    skipOptions = {"datastoreReadNamespace", "datastoreReadGqlQuery", "datastoreReadProjectId"},
    contactInformation = "https://cloud.google.com/support")
public class DatastoreToText {

  public static ValueProvider<String> selectProvidedInput(
      ValueProvider<String> datastoreInput, ValueProvider<String> firestoreInput) {
    return new FirestoreNestedValueProvider(datastoreInput, firestoreInput);
  }

  /** 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(
                    selectProvidedInput(
                        options.getDatastoreReadGqlQuery(), options.getFirestoreReadGqlQuery()))
                .setProjectId(
                    selectProvidedInput(
                        options.getDatastoreReadProjectId(), options.getFirestoreReadProjectId()))
                .setNamespace(
                    selectProvidedInput(
                        options.getDatastoreReadNamespace(), options.getFirestoreReadNamespace()))
                .build())
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())
        .apply(TextIO.write().to(options.getTextWritePrefix()).withSuffix(".json"));

    pipeline.run();
  }
}

Da Cloud Spanner a Cloud Storage Avro

Il modello da Cloud Spanner ad Avro Files su Cloud Storage è una pipeline batch che esporta un intero database Cloud Spanner in Cloud Storage in formato Avro. L'esportazione di un database Cloud Spanner crea una cartella nel bucket selezionato. La cartella contiene:

• Un file spanner-export.json.
• Un file TableName-manifest.json per ogni tabella nel database che hai esportato.
• Uno o più file TableName.avro-#####-of-#####.

Ad esempio, l'esportazione di un database con due tabelle, Singers e Albums, crea il seguente set di file:

• 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

Requisiti per questa pipeline:

• Deve esistere il database Cloud Spanner.
• Deve esistere il bucket Cloud Storage di output.
• Oltre ai ruoli IAM necessari per eseguire i job Dataflow, devi disporre anche dei ruoli IAM appropriati per la lettura dei dati di Cloud Spanner e la scrittura nel bucket Cloud Storage.

Parametri del modello

Parametro	Descrizione
instanceId	L'ID istanza del database Cloud Spanner che vuoi esportare.
databaseId	L'ID del database Cloud Spanner che vuoi esportare.
outputDir	Il percorso di Cloud Storage in cui vuoi esportare i file Avro. Il job di esportazione crea una nuova directory in questo percorso contenente i file esportati.
snapshotTime	(Facoltativo) Il timestamp corrispondente alla versione del database Cloud Spanner che vuoi leggere. Il timestamp deve essere specificato in base al formato "Zulu" RFC 3339. Ad esempio, 1990-12-31T23:59:60Z. Il timestamp deve essere passato e deve essere applicata la precedenza massima del timestamp.
tableNames	(Facoltativo) Un elenco di tabelle separate da virgole che specifica il sottoinsieme del database Cloud Spanner da esportare. L'elenco deve includere tutte le tabelle correlate (tabelle principali, tabelle di riferimento delle chiavi esterne). Se non sono elencati in modo esplicito, il flag " shouldExportrelatedTables" deve essere impostato per una corretta esportazione.
shouldExportRelatedTables	(Facoltativo) Il flag utilizzato in combinazione con il parametro "tableNames" per includere tutte le tabelle correlate da esportare.
spannerProjectId	(Facoltativo) L'ID progetto Google Cloud del database Cloud Spanner da cui vuoi leggere i dati.

Esecuzione di Cloud Spanner in un file Avro Files on Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.

   Affinché il job venga visualizzato nella pagina Istanze di Spanner della console Google Cloud, il nome del job deve corrispondere al seguente formato:

   cloud-spanner-export-SPANNER_INSTANCE_ID-SPANNER_DATABASE_NAME

   Sostituisci quanto segue:

   • SPANNER_INSTANCE_ID: ID della tua istanza Spanner
   • SPANNER_DATABASE_NAME: nome del database Spanner
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Spanner to Avro Files on Cloud Storage template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta

  Affinché il job venga visualizzato nella parte Cloud Spanner della console Google Cloud, il nome del job deve corrispondere al formato cloud-spanner-export-INSTANCE_ID-DATABASE_ID.

• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• GCS_STAGING_LOCATION: il percorso di scrittura dei file temporanei, ad esempio gs://mybucket/temp
• INSTANCE_ID: il tuo ID istanza di Cloud Spanner
• DATABASE_ID: il tuo ID database di Cloud Spanner
• GCS_DIRECTORY: il percorso di Cloud Storage da cui i file Avro vengono esportati in

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta

  Affinché il job venga visualizzato nella parte Cloud Spanner della console Google Cloud, il nome del job deve corrispondere al formato cloud-spanner-export-INSTANCE_ID-DATABASE_ID.

• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• GCS_STAGING_LOCATION: il percorso di scrittura dei file temporanei, ad esempio gs://mybucket/temp
• INSTANCE_ID: il tuo ID istanza di Cloud Spanner
• DATABASE_ID: il tuo ID database di Cloud Spanner
• GCS_DIRECTORY: il percorso di Cloud Storage da cui i file Avro vengono esportati in

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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 com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateCreationParameter;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.spanner.ExportPipeline.ExportPipelineOptions;
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;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;

/** Dataflow template that exports a Cloud Spanner database to Avro files in GCS. */
@Template(
    name = "Cloud_Spanner_to_GCS_Avro",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Spanner to Avro Files on Cloud Storage",
    description =
        "A pipeline to export a Cloud Spanner database to a set of Avro files in Cloud Storage.",
    optionsClass = ExportPipelineOptions.class,
    contactInformation = "https://cloud.google.com/support")
public class ExportPipeline {

  /** Options for Export pipeline. */
  public interface ExportPipelineOptions extends PipelineOptions {
    @TemplateParameter.Text(
        order = 1,
        regexes = {"[a-z][a-z0-9\\-]*[a-z0-9]"},
        description = "Cloud Spanner instance id",
        helpText = "The instance id of the Cloud Spanner database that you want to export.")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @TemplateParameter.Text(
        order = 2,
        regexes = {"[a-z][a-z0-9_\\-]*[a-z0-9]"},
        description = "Cloud Spanner database id",
        helpText = "The database id of the Cloud Spanner database that you want to export.")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @TemplateParameter.GcsWriteFolder(
        order = 3,
        description = "Cloud Storage output directory",
        helpText =
            "The Cloud Storage path where the Avro files should be exported to. A new directory will be created under this path that contains the export.",
        example = "gs://your-bucket/your-path")
    ValueProvider<String> getOutputDir();

    void setOutputDir(ValueProvider<String> value);

    @TemplateParameter.GcsWriteFolder(
        order = 4,
        optional = true,
        description = "Cloud Storage temp directory for storing Avro files",
        helpText =
            "The Cloud Storage path where the temporary Avro files can be created. Ex: gs://your-bucket/your-path")
    ValueProvider<String> getAvroTempDirectory();

    void setAvroTempDirectory(ValueProvider<String> value);

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

    void setTestJobId(ValueProvider<String> jobId);

    @TemplateParameter.Text(
        order = 6,
        optional = true,
        description = "Cloud Spanner Endpoint to call",
        helpText = "The Cloud Spanner endpoint to call in the template. Only used for testing.",
        example = "https://batch-spanner.googleapis.com")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

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

    void setWaitUntilFinish(boolean value);

    @TemplateParameter.Text(
        order = 7,
        optional = true,
        regexes = {
          "^([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):(([0-9]{2})(\\.[0-9]+)?)Z$"
        },
        description = "Snapshot time",
        helpText =
            "Specifies the snapshot time as RFC 3339 format in UTC time without the timezone offset(always ends in 'Z'). Timestamp must be in the past and Maximum timestamp staleness applies. See https://cloud.google.com/spanner/docs/timestamp-bounds#maximum_timestamp_staleness",
        example = "1990-12-31T23:59:59Z")
    @Default.String(value = "")
    ValueProvider<String> getSnapshotTime();

    void setSnapshotTime(ValueProvider<String> value);

    @TemplateParameter.ProjectId(
        order = 8,
        optional = true,
        description = "Cloud Spanner Project Id",
        helpText = "The project id of the Cloud Spanner instance.")
    ValueProvider<String> getSpannerProjectId();

    void setSpannerProjectId(ValueProvider<String> value);

    @TemplateParameter.Boolean(
        order = 9,
        optional = true,
        description = "Export Timestamps as Timestamp-micros type",
        helpText =
            "If true, Timestamps are exported as timestamp-micros type. Timestamps are exported as ISO8601 strings at nanosecond precision by default.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportTimestampAsLogicalType();

    void setShouldExportTimestampAsLogicalType(ValueProvider<Boolean> value);

    @TemplateParameter.Text(
        order = 10,
        optional = true,
        regexes = {"^[a-zA-Z0-9_]+(,[a-zA-Z0-9_]+)*$"},
        description = "Cloud Spanner table name(s).",
        helpText =
            "If provided, only this comma separated list of tables are exported. Ancestor tables and tables that are referenced via foreign keys are required. If not explicitly listed, the `shouldExportRelatedTables` flag must be set for a successful export.")
    @Default.String(value = "")
    ValueProvider<String> getTableNames();

    void setTableNames(ValueProvider<String> value);

    @TemplateParameter.Boolean(
        order = 11,
        optional = true,
        description = "Export necessary Related Spanner tables.",
        helpText =
            "Used in conjunction with `tableNames`. If true, add related tables necessary for the export, such as interleaved parent tables and foreign keys tables.  If `tableNames` is specified but doesn't include related tables, this option must be set to true for a successful export.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportRelatedTables();

    void setShouldExportRelatedTables(ValueProvider<Boolean> value);

    @TemplateParameter.Enum(
        order = 12,
        enumOptions = {"LOW", "MEDIUM", "HIGH"},
        optional = true,
        description = "Priority for Spanner RPC invocations",
        helpText =
            "The request priority for Cloud Spanner calls. The value must be one of: [HIGH,MEDIUM,LOW].")
    ValueProvider<RpcPriority> getSpannerPriority();

    void setSpannerPriority(ValueProvider<RpcPriority> 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()
            // Temporary fix explicitly setting SpannerConfig.projectId to the default project
            // if spannerProjectId is not provided as a parameter. Required as of Beam 2.38,
            // which no longer accepts null label values on metrics, and SpannerIO#setup() has
            // a bug resulting in the label value being set to the original parameter value,
            // with no fallback to the default project.
            // TODO: remove NestedValueProvider when this is fixed in Beam.
            .withProjectId(
                NestedValueProvider.of(
                    options.getSpannerProjectId(),
                    (SerializableFunction<String, String>)
                        input -> input != null ? input : SpannerOptions.getDefaultProjectId()))
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId())
            .withRpcPriority(options.getSpannerPriority());
    p.begin()
        .apply(
            "Run Export",
            new ExportTransform(
                spannerConfig,
                options.getOutputDir(),
                options.getTestJobId(),
                options.getSnapshotTime(),
                options.getTableNames(),
                options.getShouldExportRelatedTables(),
                options.getShouldExportTimestampAsLogicalType(),
                options.getAvroTempDirectory()));
    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();
    }
  }
}

Testo da Cloud Spanner a Cloud Storage

Il modello Cloud Spanner to Cloud Storage Text è una pipeline batch che legge i dati di una tabella Cloud Spanner e li scrive in Cloud Storage sotto forma di file di testo CSV.

Requisiti per questa pipeline:

• La tabella Spanner di input deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
spannerProjectId	L'ID progetto Google Cloud del database Cloud Spanner da cui vuoi leggere i dati.
spannerDatabaseId	L'ID database della tabella richiesta.
spannerInstanceId	L'ID istanza della tabella richiesta.
spannerTable	La tabella da cui leggere i dati.
textWritePrefix	La directory in cui vengono scritti i file di testo di output. Aggiungi / alla fine. Ad esempio: gs://mybucket/somefolder/.
spannerSnapshotTime	(Facoltativo) Il timestamp corrispondente alla versione del database Cloud Spanner che vuoi leggere. Il timestamp deve essere specificato in base al formato "Zulu" RFC 3339. Ad esempio, 1990-12-31T23:59:60Z. Il timestamp deve essere passato e deve essere applicata la precedenza massima del timestamp.

Esecuzione del modello di testo da Cloud Spanner a Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Spanner to Text Files on Cloud Storage template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Spanner_to_GCS_Text \
    --region REGION_NAME \
    --parameters \
spannerProjectId=SPANNER_PROJECT_ID,\
spannerDatabaseId=DATABASE_ID,\
spannerInstanceId=INSTANCE_ID,\
spannerTable=TABLE_ID,\
textWritePrefix=gs://BUCKET_NAME/output/

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• SPANNER_PROJECT_ID: l'ID del progetto Cloud del database Spanner da cui vuoi leggere i dati
• DATABASE_ID: l'ID del database Spanner
• BUCKET_NAME: nome del bucket Cloud Storage
• INSTANCE_ID: l'ID istanza di Spanner
• TABLE_ID: l'ID della tabella Spanner

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• SPANNER_PROJECT_ID: l'ID del progetto Cloud del database Spanner da cui vuoi leggere i dati
• DATABASE_ID: l'ID del database Spanner
• BUCKET_NAME: nome del bucket Cloud Storage
• INSTANCE_ID: l'ID istanza di Spanner
• TABLE_ID: l'ID della tabella Spanner

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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 static com.google.cloud.teleport.util.ValueProviderUtils.eitherOrValueProvider;

import com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.templates.SpannerToText.SpannerToTextOptions;
import com.google.cloud.teleport.templates.common.SpannerConverters;
import com.google.cloud.teleport.templates.common.SpannerConverters.CreateTransactionFnWithTimestamp;
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.fs.ResourceId;
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.io.gcp.spanner.Transaction;
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.Create;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.values.PBegin;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
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 \
 *                --spannerSnapshotTime=snapshot_time \
 *                --textWritePrefix=gcsOutputPath"
 * </pre>
 */
@Template(
    name = "Spanner_to_GCS_Text",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Spanner to Text Files on Cloud Storage",
    description =
        "A pipeline which reads in Cloud Spanner table and writes it to Cloud Storage as CSV text files.",
    optionsClass = SpannerToTextOptions.class,
    contactInformation = "https://cloud.google.com/support")
public class SpannerToText {

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

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

    @TemplateParameter.GcsWriteFolder(
        order = 1,
        optional = true,
        description = "Cloud Storage temp directory for storing CSV files",
        helpText = "The Cloud Storage path where the temporary CSV files can be stored.",
        example = "gs://your-bucket/your-path")
    ValueProvider<String> getCsvTempDirectory();

    @SuppressWarnings("unused")
    void setCsvTempDirectory(ValueProvider<String> value);

    @TemplateParameter.Enum(
        order = 2,
        enumOptions = {"LOW", "MEDIUM", "HIGH"},
        optional = true,
        description = "Priority for Spanner RPC invocations",
        helpText =
            "The request priority for Cloud Spanner calls. The value must be one of: [HIGH,MEDIUM,LOW].")
    ValueProvider<RpcPriority> getSpannerPriority();

    void setSpannerPriority(ValueProvider<RpcPriority> value);
  }

  /**
   * Runs a pipeline which reads in Records from Spanner, 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())
            .withRpcPriority(options.getSpannerPriority());

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

    /* CreateTransaction and CreateTransactionFn classes in LocalSpannerIO
     * only take a timestamp object for exact staleness which works when
     * parameters are provided during template compile time. They do not work with
     * a Timestamp valueProvider which can take parameters at runtime. Hence a new
     * ParDo class CreateTransactionFnWithTimestamp had to be created for this
     * purpose.
     */
    PCollectionView<Transaction> tx =
        pipeline
            .apply("Setup for Transaction", Create.of(1))
            .apply(
                "Create transaction",
                ParDo.of(
                    new CreateTransactionFnWithTimestamp(
                        spannerConfig, options.getSpannerSnapshotTime())))
            .apply("As PCollectionView", View.asSingleton());

    PCollection<String> csv =
        pipeline
            .apply("Create export", spannerExport)
            // We need to use LocalSpannerIO.readAll() instead of LocalSpannerIO.read()
            // because ValueProvider parameters such as table name required for
            // LocalSpannerIO.read() can be read only inside DoFn but LocalSpannerIO.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().withTransaction(tx).withSpannerConfig(spannerConfig))
            .apply(
                "Struct To Csv",
                MapElements.into(TypeDescriptors.strings())
                    .via(struct -> (new SpannerConverters.StructCsvPrinter()).print(struct)));

    ValueProvider<ResourceId> tempDirectoryResource =
        ValueProvider.NestedValueProvider.of(
            eitherOrValueProvider(options.getCsvTempDirectory(), options.getTextWritePrefix()),
            (SerializableFunction<String, ResourceId>) s -> FileSystems.matchNewResource(s, true));

    csv.apply(
        "Write to storage",
        TextIO.write()
            .to(options.getTextWritePrefix())
            .withSuffix(".csv")
            .withTempDirectory(tempDirectoryResource));

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

Da Cloud Storage Avro a Bigtable

Il modello Cloud Storage Avro to Bigtable è una pipeline che legge i dati dai file Avro in un bucket Cloud Storage e li scrive in una tabella Bigtable. Puoi utilizzare il modello per copiare i dati da Cloud Storage in Bigtable.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere e avere le stesse famiglie di colonne esportate nei file Avro.
• I file Avro di input devono esistere in un bucket Cloud Storage prima di eseguire la pipeline.
• Bigtable prevede uno schema specifico dai file Avro di input.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable in cui vuoi scrivere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da importare.
inputFilePattern	Il pattern del percorso di Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del file Avro di Cloud Storage nel modello Bigtable

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Avro Files on Cloud Storage to Cloud Bigtable template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• INPUT_FILE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• INPUT_FILE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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.cloud.teleport.bigtable.AvroToBigtable.Options;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
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.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".
 */
@Template(
    name = "GCS_Avro_to_Cloud_Bigtable",
    category = TemplateCategory.BATCH,
    displayName = "Avro Files on Cloud Storage to Cloud Bigtable",
    description =
        "A pipeline which reads data from Avro files in Cloud Storage and writes it to Cloud Bigtable table.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
public 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 {
    @TemplateParameter.ProjectId(
        order = 1,
        description = "Project ID",
        helpText =
            "The ID of the Google Cloud project of the Cloud Bigtable instance that you want to write data to")
    ValueProvider<String> getBigtableProjectId();

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

    @TemplateParameter.Text(
        order = 2,
        regexes = {"[a-z][a-z0-9\\-]+[a-z0-9]"},
        description = "Instance ID",
        helpText = "The ID of the Cloud Bigtable instance that contains the table")
    ValueProvider<String> getBigtableInstanceId();

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

    @TemplateParameter.Text(
        order = 4,
        regexes = {"[_a-zA-Z0-9][-_.a-zA-Z0-9]*"},
        description = "Table ID",
        helpText = "The ID of the Cloud Bigtable table to write")
    ValueProvider<String> getBigtableTableId();

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

    @TemplateParameter.GcsReadFile(
        order = 5,
        description = "Input Cloud Storage File(s)",
        helpText = "The Cloud Storage location of the files you'd like to process.",
        example = "gs://your-bucket/your-files/*.avro")
    ValueProvider<String> getInputFilePattern();

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

    @TemplateParameter.Boolean(
        order = 6,
        optional = true,
        description = "If true, large rows will be split into multiple MutateRows requests",
        helpText =
            "The flag for enabling splitting of large rows into multiple MutateRows requests. Note that when a large row is split between multiple API calls, the updates to the row are not atomic. ")
    ValueProvider<Boolean> getSplitLargeRows();

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

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

Da Cloud Storage Avro a Cloud Spanner

I file Avro di Cloud Storage nel modello Cloud Spanner sono una pipeline batch che legge i file Avro da Cloud Spanner archiviati in Cloud Storage e li importa in un database Cloud Spanner.

Requisiti per questa pipeline:

• Il database Cloud Spanner di destinazione deve esistere e deve essere vuoto.
• Devi avere le autorizzazioni di lettura per il bucket Cloud Storage e le autorizzazioni di scrittura per il database Cloud Spanner di destinazione.
• Il percorso di Cloud Storage di input deve esistere e deve includere un file spanner-export.json contenente una descrizione JSON dei file da importare.

Avviso: se il database di destinazione di Cloud Spanner non è vuoto, i nuovi dati potrebbero sovrascrivere i dati preesistenti o l'importazione potrebbe non riuscire.

Parametri del modello

Parametro	Descrizione
instanceId	L'ID istanza del database Cloud Spanner.
databaseId	L'ID del database Cloud Spanner.
inputDir	Percorso di Cloud Storage da cui vengono importati i file Avro.

Esecuzione del modello Avro in Cloud Spanner da Cloud Storage

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.

   Affinché il job venga visualizzato nella pagina Istanze di Spanner della console Google Cloud, il nome del job deve corrispondere al seguente formato:

   cloud-spanner-import-SPANNER_INSTANCE_ID-SPANNER_DATABASE_NAME

   Sostituisci quanto segue:

   • SPANNER_INSTANCE_ID: ID della tua istanza Spanner
   • SPANNER_DATABASE_NAME: nome del database Spanner
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Avro Files on Cloud Storage to Cloud Spanner template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• INSTANCE_ID: l'ID dell'istanza Spanner che contiene il database
• DATABASE_ID: l'ID del database Spanner in cui eseguire l'importazione
• GCS_DIRECTORY: il percorso di Cloud Storage da cui vengono importati i file Avro, ad esempio gs://mybucket/somefolder

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• INSTANCE_ID: l'ID dell'istanza Spanner che contiene il database
• DATABASE_ID: l'ID del database Spanner in cui eseguire l'importazione
• GCS_DIRECTORY: il percorso di Cloud Storage da cui vengono importati i file Avro, ad esempio gs://mybucket/somefolder

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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 com.google.cloud.spanner.Options.RpcPriority;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateCreationParameter;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.spanner.ImportPipeline.Options;
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;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;

/** Avro to Cloud Spanner Import pipeline. */
@Template(
    name = "GCS_Avro_to_Cloud_Spanner",
    category = TemplateCategory.BATCH,
    displayName = "Avro Files on Cloud Storage to Cloud Spanner",
    description =
        "A pipeline to import a Cloud Spanner database from a set of Avro files in Cloud Storage.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
public class ImportPipeline {

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

    @TemplateParameter.Text(
        order = 1,
        regexes = {"^[a-z0-9\\-]+$"},
        description = "Cloud Spanner instance id",
        helpText = "The instance id of the Cloud Spanner database that you want to import to.")
    ValueProvider<String> getInstanceId();

    void setInstanceId(ValueProvider<String> value);

    @TemplateParameter.Text(
        order = 2,
        regexes = {"^[a-z_0-9\\-]+$"},
        description = "Cloud Spanner database id",
        helpText =
            "The database id of the Cloud Spanner database that you want to import into (must already exist).")
    ValueProvider<String> getDatabaseId();

    void setDatabaseId(ValueProvider<String> value);

    @TemplateParameter.GcsReadFolder(
        order = 3,
        description = "Cloud storage input directory",
        helpText = "The Cloud Storage path where the Avro files should be imported from.")
    ValueProvider<String> getInputDir();

    void setInputDir(ValueProvider<String> value);

    @TemplateParameter.Text(
        order = 4,
        optional = true,
        description = "Cloud Spanner Endpoint to call",
        helpText = "The Cloud Spanner endpoint to call in the template. Only used for testing.",
        example = "https://batch-spanner.googleapis.com")
    @Default.String("https://batch-spanner.googleapis.com")
    ValueProvider<String> getSpannerHost();

    void setSpannerHost(ValueProvider<String> value);

    @TemplateParameter.Boolean(
        order = 5,
        optional = true,
        description = "Wait for Indexes",
        helpText =
            "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);

    @TemplateParameter.Boolean(
        order = 6,
        optional = true,
        description = "Wait for Foreign Keys",
        helpText =
            "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);

    @TemplateParameter.Boolean(
        order = 7,
        optional = true,
        description = "Wait for Foreign Keys",
        helpText =
            "By default the import pipeline is blocked on change stream creation. If false, it may"
                + " complete with change streams still being created in the background.")
    @Default.Boolean(true)
    ValueProvider<Boolean> getWaitForChangeStreams();

    void setWaitForChangeStreams(ValueProvider<Boolean> value);

    @TemplateParameter.Boolean(
        order = 8,
        optional = true,
        description = "Create Indexes early",
        helpText =
            "Flag to turn off early index creation if there are many indexes. 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);

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

    @TemplateParameter.ProjectId(
        order = 9,
        optional = true,
        description = "Cloud Spanner Project Id",
        helpText = "The project id of the Cloud Spanner instance.")
    ValueProvider<String> getSpannerProjectId();

    void setSpannerProjectId(ValueProvider<String> value);

    void setWaitUntilFinish(boolean value);

    @TemplateParameter.Text(
        order = 10,
        optional = true,
        regexes = {"[0-9]+"},
        description = "DDL Creation timeout in minutes",
        helpText = "DDL Creation timeout in minutes.")
    @Default.Integer(30)
    ValueProvider<Integer> getDDLCreationTimeoutInMinutes();

    void setDDLCreationTimeoutInMinutes(ValueProvider<Integer> value);

    @TemplateParameter.Enum(
        order = 11,
        enumOptions = {"LOW", "MEDIUM", "HIGH"},
        optional = true,
        description = "Priority for Spanner RPC invocations",
        helpText =
            "The request priority for Cloud Spanner calls. The value must be one of: [HIGH,MEDIUM,LOW].")
    ValueProvider<RpcPriority> getSpannerPriority();

    void setSpannerPriority(ValueProvider<RpcPriority> 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()
            // Temporary fix explicitly setting SpannerConfig.projectId to the default project
            // if spannerProjectId is not provided as a parameter. Required as of Beam 2.38,
            // which no longer accepts null label values on metrics, and SpannerIO#setup() has
            // a bug resulting in the label value being set to the original parameter value,
            // with no fallback to the default project.
            // TODO: remove NestedValueProvider when this is fixed in Beam.
            .withProjectId(
                NestedValueProvider.of(
                    options.getSpannerProjectId(),
                    (SerializableFunction<String, String>)
                        input -> input != null ? input : SpannerOptions.getDefaultProjectId()))
            .withHost(options.getSpannerHost())
            .withInstanceId(options.getInstanceId())
            .withDatabaseId(options.getDatabaseId())
            .withRpcPriority(options.getSpannerPriority());

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

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

Da Cloud Storage Parquet a Bigtable

Il modello da spazio di archiviazione Cloud Storage a Bigtable è una pipeline che legge i dati dai file Parquet in un bucket Cloud Storage e li scrive in una tabella Bigtable. Puoi utilizzare il modello per copiare i dati da Cloud Storage in Bigtable.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere e avere le stesse famiglie di colonne esportate nei file Parquet.
• I file Parquet di input devono esistere in un bucket Cloud Storage prima di eseguire la pipeline.
• Bigtable prevede uno schema specifico dai file Parquet di input.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable in cui vuoi scrivere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da importare.
inputFilePattern	Il pattern del percorso di Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del file Parquet di Cloud Storage in base al modello Bigtable

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Parquet Files on Cloud Storage to Cloud Bigtable template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• INPUT_FILE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• INPUT_FILE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2019 Google LLC
 *
 * 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.cloud.teleport.bigtable.ParquetToBigtable.Options;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
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.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>
 */
@Template(
    name = "GCS_Parquet_to_Cloud_Bigtable",
    category = TemplateCategory.BATCH,
    displayName = "Parquet Files on Cloud Storage to Cloud Bigtable",
    description =
        "A pipeline which reads data from Parquet files in Cloud Storage and writes it to Cloud Bigtable table.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
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 {
    @TemplateParameter.ProjectId(
        order = 1,
        description = "Project ID",
        helpText =
            "The ID of the Google Cloud project of the Cloud Bigtable instance that you want to write data to")
    ValueProvider<String> getBigtableProjectId();

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

    @TemplateParameter.Text(
        order = 2,
        regexes = {"[a-z][a-z0-9\\-]+[a-z0-9]"},
        description = "Instance ID",
        helpText = "The ID of the Cloud Bigtable instance that contains the table")
    ValueProvider<String> getBigtableInstanceId();

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

    @TemplateParameter.Text(
        order = 3,
        regexes = {"[_a-zA-Z0-9][-_.a-zA-Z0-9]*"},
        description = "Table ID",
        helpText = "The ID of the Cloud Bigtable table to write")
    ValueProvider<String> getBigtableTableId();

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

    @TemplateParameter.GcsReadFile(
        order = 4,
        description = "Input Cloud Storage File(s)",
        helpText = "The Cloud Storage location of the files you'd like to process.",
        example = "gs://your-bucket/your-files/*.parquet")
    ValueProvider<String> getInputFilePattern();

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

    @TemplateParameter.Boolean(
        order = 5,
        optional = true,
        description = "If true, large rows will be split into multiple MutateRows requests",
        helpText =
            "The flag for enabling splitting of large rows into multiple MutateRows requests. Note that when a large row is split between multiple API calls, the updates to the row 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));
      }
    }
  }
}

SequenzaFile di Cloud Storage in Bigtable

Il modello SequenzaFile di Cloud Storage in Bigtable è una pipeline che legge i dati da SecondsFile in un bucket Cloud Storage e scrive i dati in una tabella Bigtable. Puoi utilizzare il modello per copiare i dati da Cloud Storage in Bigtable.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Il SequenzaFile di input deve esistere in un bucket Cloud Storage prima di eseguire la pipeline.
• Il file di sequenza di input deve essere stato esportato da Bigtable o HBase.

Parametri del modello

Parametro	Descrizione
bigtableProject	L'ID del progetto Google Cloud dell'istanza Bigtable in cui vuoi scrivere i dati.
bigtableInstanceId	L'ID dell'istanza Bigtable che contiene la tabella.
bigtableTableId	L'ID della tabella Bigtable da importare.
bigtableAppProfileId	L'ID del profilo dell'applicazione Bigtable da utilizzare per l'importazione. Se non specifichi un profilo app, Bigtable utilizza il profilo app predefinito dell'istanza.
sourcePattern	Il pattern del percorso di Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del modello SequenzaFile di Cloud Storage in Bigtable

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the SequenceFile Files on Cloud Storage to Cloud Bigtable template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• APPLICATION_PROFILE_ID: l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• SOURCE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/GCS_SequenceFile_to_Cloud_Bigtable
{
   "jobName": "JOB_NAME",
   "parameters": {
       "bigtableProject": "BIGTABLE_PROJECT_ID",
       "bigtableInstanceId": "INSTANCE_ID",
       "bigtableTableId": "TABLE_ID",
       "bigtableAppProfileId": "APPLICATION_PROFILE_ID",
       "sourcePattern": "SOURCE_PATTERN",
   },
   "environment": { "zone": "us-central1-f" }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable da cui vuoi leggere i dati
• INSTANCE_ID: l'ID dell'istanza Bigtable che contiene la tabella
• TABLE_ID: l'ID della tabella Bigtable da esportare
• APPLICATION_PROFILE_ID: l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• SOURCE_PATTERN: pattern del percorso di Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Codice sorgente del modello

Java

Il codice sorgente di questo modello si trova nel repository GoogleCloudPlatform/cloud-bigtable-client su GitHub.

Testo di Cloud Storage in BigQuery

La pipeline Cloud Text Text to BigQuery è una pipeline batch che ti consente di leggere i file di testo archiviati in Cloud Storage, trasformarli utilizzando una funzione definita dall'utente JavaScript (UDF) da te fornita e aggiungere il risultato a una tabella BigQuery.

Nota: se vuoi sovrascrivere i dati nella tabella BigQuery invece di aggiungerla, aggiorna WriteDisposition nel modello di codice sorgente da WRITE_APPEND a WRITE_TRUNCATE.

Requisiti per questa pipeline:

• Crea un file JSON che descriva lo schema BigQuery.

  Assicurati che esista un array JSON di primo livello denominato BigQuery Schema e che i suoi contenuti seguano il pattern {"name": "COLUMN_NAME", "type": "DATA_TYPE"}.

  Il modello batch di Cloud Storage per il testo in BigQuery non supporta l'importazione di dati nei campi STRUCT (record) della tabella BigQuery di destinazione.

  Il seguente JSON descrive uno schema BigQuery di esempio:

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

• Crea un file JavaScript (.js) con la funzione UDF che fornisca la logica per trasformare le righe di testo. La funzione deve restituire una stringa JSON.

  Ad esempio, questa funzione suddivide ogni riga di un file CSV e restituisce una stringa JSON dopo aver trasformato i valori.

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

Parametri del modello

Parametro	Descrizione
javascriptTextTransformFunctionName	Il nome della funzione JavaScript definita dall'utente (UDF) che vuoi utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.
JSONPath	Il percorso gs:// al file JSON che definisce lo schema BigQuery, archiviato in Cloud Storage. Ad esempio, gs://path/to/my/schema.json.
javascriptTextTransformGcsPath	L'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
inputFilePattern	Il percorso gs:// che rimanda al testo in Cloud Storage che vuoi elaborare. Ad esempio, gs://path/to/my/text/data.txt.
outputTable	Il nome della tabella BigQuery in cui vuoi creare un archivio per i dati elaborati. Se riutilizzi una tabella BigQuery esistente, i dati vengono aggiunti alla tabella di destinazione. Ad esempio, my-project-name:my-dataset.my-table.
bigQueryLoadingTemporaryDirectory	La directory temporanea per il processo di caricamento di BigQuery. Ad esempio, gs://my-bucket/my-files/temp_dir.

Esecuzione del modello Cloud Storage Text to BigQuery

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Text Files on Cloud Storage to BigQuery (Batch) template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/GCS_Text_to_BigQuery \
    --region REGION_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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_BIGQUERY_SCHEMA_JSON: il percorso di Cloud Storage del file JSON contenente la definizione dello schema
• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• PATH_TO_TEXT_DATA: il tuo percorso di Cloud Storage verso il tuo set di dati di testo
• BIGQUERY_TABLE: il nome della tua tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: il tuo percorso di Cloud Storage nella directory temporanea

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• LOCATION: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_BIGQUERY_SCHEMA_JSON: il percorso di Cloud Storage del file JSON contenente la definizione dello schema
• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• PATH_TO_TEXT_DATA: il tuo percorso di Cloud Storage verso il tuo set di dati di testo
• BIGQUERY_TABLE: il nome della tua tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: il tuo percorso di Cloud Storage nella directory temporanea

Codice sorgente del modello

Java

Visualizza su GitHub Feedback

/*
 * Copyright (C) 2018 Google LLC
 *
 * 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.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.templates.TextIOToBigQuery.Options;
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.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.
 */
@Template(
    name = "GCS_Text_to_BigQuery",
    category = TemplateCategory.BATCH,
    displayName = "Text Files on Cloud Storage to BigQuery",
    description =
        "Batch pipeline. Reads text files stored in Cloud Storage, transforms them using a JavaScript user-defined function (UDF), and outputs the result to BigQuery.",
    optionsClass = Options.class,
    contactInformation = "https://cloud.google.com/support")
public class TextIOToBigQuery {

  /** Options supported by {@link TextIOToBigQuery}. */
  public interface Options extends DataflowPipelineOptions, JavascriptTextTransformerOptions {

    @TemplateParameter.GcsReadFile(
        order = 1,
        description = "Cloud Storage Input File(s)",
        helpText = "Path of the file pattern glob to read from.",
        example = "gs://your-bucket/path/*.csv")
    ValueProvider<String> getInputFilePattern();

    void setInputFilePattern(ValueProvider<String> value);

    @TemplateParameter.GcsReadFile(
        order = 2,
        description = "Cloud Storage location of your BigQuery schema file, described as a JSON",
        helpText =
            "JSON file with BigQuery Schema description. JSON Example: {\n"
                + "\t\"BigQuery Schema\": [\n"
                + "\t\t{\n"
                + "\t\t\t\"name\": \"location\",\n"
                + "\t\t\t\"type\": \"STRING\"\n"
                + "\t\t},\n"
                + "\t\t{\n"
                + "\t\t\t\"name\": \"name\",\n"
                + "\t\t\t\"type\": \"STRING\"\n"
                + "\t\t},\n"
                + "\t\t{\n"
                + "\t\t\t\"name\": \"age\",\n"
                + "\t\t\t\"type\": \"STRING\"\n"
                + "\t\t},\n"
                + "\t\t{\n"
                + "\t\t\t\"name\": \"color\",\n"
                + "\t\t\t\"type\": \"STRING\"\n"
                + "\t\t},\n"
                + "\t\t{\n"
                + "\t\t\t\"name\": \"coffee\",\n"
                + "\t\t\t\"type\": \"STRING\"\n"
                + "\t\t}\n"
                + "\t]\n"
                + "}")
    ValueProvider<String> getJSONPath();

    void setJSONPath(ValueProvider<String> value);

    @TemplateParameter.BigQueryTable(
        order = 3,
        description = "BigQuery output table",
        helpText =
            "BigQuery table location to write the output to. The table's schema must match the "
                + "input objects.")
    ValueProvider<String> getOutputTable();

    void setOutputTable(ValueProvider<String> value);

    @TemplateParameter.GcsWriteFolder(
        order = 6,
        description = "Temporary directory for BigQuery loading process",
        helpText = "Temporary directory for BigQuery loading process",
        example = "gs://your-bucket/your-files/temp_dir")
    @Validation.Required
    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();
  }
}

Testo di Cloud Storage in Datastore [deprecato]

beta

Questa funzionalità è regolata dai Termini delle offerte pre-GA dei Termini di servizio di Google Cloud. Le funzionalità pre-GA potrebbero avere supporto limitato e le modifiche a queste funzionalità potrebbero non essere compatibili con altre versioni pre-GA. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Questo modello è deprecato e verrà rimosso nel primo trimestre del 2022. Esegui la migrazione al modello Cloud Storage Text to Firestore.

Il modello Cloud Storage Text to Datastore è una pipeline batch che legge dai file di testo archiviati in Cloud Storage e scrive le entità con codifica JSON in Datastore. Ogni riga nei file di testo di input deve essere nel formato JSON specificato.

Requisiti per questa pipeline:

• Datastore deve essere abilitato nel progetto di destinazione.

Parametri del modello

Parametro	Descrizione
textReadPattern	Un pattern del percorso Cloud Storage che specifica la posizione dei file di dati di testo. Ad esempio, gs://mybucket/somepath/*.json.
javascriptTextTransformGcsPath	(Facoltativo) L'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName	(Facoltativo) Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.
datastoreWriteProjectId	L'ID del progetto Google Cloud di dove scrivere le entità Datastore
datastoreHintNumWorkers	(Facoltativo) Suggerimento per il numero previsto di worker nel passaggio di limitazione del datastore. Il valore predefinito è 500.
errorWritePath	Il file di output del log degli errori da utilizzare per gli errori di scrittura che si verificano durante l'elaborazione. Ad esempio, gs://bucket-name/errors.txt.

Esecuzione del modello Cloud Storage Text to Datastore

Console

1. Vai alla pagina Crea job da modello di Dataflow.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco per il job.
3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. L'endpoint della regione predefinito è us-central1.

   Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta le località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Text Files on Cloud Storage to Datastore template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

Nota:per utilizzare Google Cloud CLI per eseguire i modelli classic, devi disporre di Google Cloud CLI 138.0.0 o versioni successive.

Nella shell o nel terminale, esegui il modello:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/GCS_Text_to_Datastore \
    --region REGION_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

Sostituisci quanto segue:

• JOB_NAME: un nome job univoco a tua scelta
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale senza data del bucket: gs://dataflow-templates/latest/
  • Il nome della versione, come 2021-09-20-00_RC00, per utilizzare una versione specifica del modello, che può essere nidificata nella rispettiva cartella padre con data nel bucket: gs://dataflow-templates/
  Attenzione: l'ultima versione dei modelli potrebbe essere aggiornata a causa di modifiche che provocano un errore. I tuoi ambienti di produzione dovrebbero utilizzare i modelli conservati nella cartella padre datata più recente per evitare che queste modifiche che provocano un errore influiscano sui tuoi flussi di lavoro di produzione.
• REGION_NAME: l'endpoint a livello di area geografica in cui vuoi eseguire il deployment del tuo job Dataflow, ad esempio us-central1
• PATH_TO_INPUT_TEXT_FILES: pattern dei file di input su Cloud Storage
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

  Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per un esempio di UDF JavaScript, consulta Esempi di funzioni UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• ERROR_FILE_WRITE_PATH: percorso verso il file di errore su Cloud Storage

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui suoi ambiti di autorizzazione, consulta projects.templates.launch.

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID del progetto Cloud in cui vuoi es