Modelli batch forniti da Google

Google fornisce un insieme di modelli di flussi di dati open source. Per informazioni generali sui modelli, consulta la sezione Modelli di Dataflow. Per un elenco di tutti i modelli forniti da Google, consulta la Guida introduttiva all'utilizzo dei 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 di 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 si imposta la suddivisione del set di dati, la somma di addestramento, test e convalida deve essere pari a 1 o 100% (ad esempio, 0,6+0,2+0,2). Dataflow determina automaticamente il numero ottimale di shard per ciascun set di dati di output.

Requisiti per questa pipeline:

• Il set di dati e la tabella BigQuery devono esistere.
• Il bucket Cloud Storage di output deve esistere prima dell'esecuzione della pipeline. Non è necessario preesistere alle sottodirectory, che sono necessarie per l'addestramento, il test e la convalida.

Parametri del modello

Parametro	Descrizione
readQuery	Una query SQL BigQuery che estrae dati dall'origine. Ad esempio, select * from dataset1.sample_table.
outputDirectory	Il prefisso percorso di primo livello di Cloud Storage per scrivere i file TFRecord di addestramento, test e convalida. Ad esempio, gs://mybucket/output. Le sottodirectory per i file TFRecord di addestramento, test e convalida risultanti 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 assegnate ai test dei file TFRecord. Il valore predefinito è 0 o 0%.
validationPercentage	(Facoltativo) La percentuale di dati di 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 da BigQuery a Cloud Storage

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, 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 suddivisione percentuale decimale per il set di dati di addestramento
• TESTING_PERCENTAGE: la suddivisione percentuale decimale per il set di dati di test
• VALIDATION_PERCENTAGE: la suddivisione percentuale decimale per il set di dati di convalida
• OUTPUT_FILENAME_SUFFIX: il suffisso file di output preferito di TensorFlow

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, 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 suddivisione percentuale decimale per il set di dati di addestramento
• TESTING_PERCENTAGE: la suddivisione percentuale decimale per il set di dati di test
• VALIDATION_PERCENTAGE: la suddivisione percentuale decimale per il set di dati di convalida
• OUTPUT_FILENAME_SUFFIX: il suffisso file di output preferito di TensorFlow

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

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

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

  private static final String TEST = "test/";
  private static final String VAL = "val/";

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

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

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

  /**
   * The {@link BigQueryToTFRecord#record2Example(SchemaAndRecord)} method uses takes in a
   * SchemaAndRecord Object returned from a BigQueryIO.read() step and builds a TensorFlow Example
   * from the record.
   */
  @VisibleForTesting
  protected static byte[] record2Example(SchemaAndRecord schemaAndRecord) {
    Example.Builder example = Example.newBuilder();
    Features.Builder features = example.getFeaturesBuilder();
    GenericRecord record = schemaAndRecord.getRecord();
    for (TableFieldSchema field : schemaAndRecord.getTableSchema().getFields()) {
      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 {

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

    void setOutputDirectory(ValueProvider<String> outputDirectory);

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

    void setOutputSuffix(ValueProvider<String> outputSuffix);

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

    void setTrainingPercentage(ValueProvider<Float> trainingPercentage);

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

    void setTestingPercentage(ValueProvider<Float> testingPercentage);

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

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

Esportazione di BigQuery in Parquet (tramite API Storage)

Il modello BigQuery Export in Parquet è una pipeline batch che legge i dati da 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.
• Prima di eseguire la pipeline, deve esistere il bucket Cloud Storage di output.

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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli flex, devi disporre dell'interfaccia a riga di comando di Google Cloud 284.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGQUERY_TABLE: nome della tabella BigQuery
• OUTPUT_DIRECTORY: la tua cartella Cloud Storage per i file di output
• NUM_SHARDS: il numero desiderato di shard del file di output
• FIELDS: l'elenco di campi separati da virgole dalla tabella BigQuery di input.

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGQUERY_TABLE: nome della tabella BigQuery
• OUTPUT_DIRECTORY: la tua cartella Cloud Storage per i file di output
• NUM_SHARDS: il numero desiderato di shard del file di output
• FIELDS: l'elenco di campi separati da virgole 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 java.io.IOException;
import java.util.Arrays;
import java.util.List;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.io.parquet.ParquetIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

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

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

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

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

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

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

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

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

    void setTableRef(String tableRef);

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

    void setBucket(String bucket);

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

    void setNumShards(Integer numShards);

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

    void setFields(String fields);
  }

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

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

    return avroSchema;
  }

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

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(BigQueryToParquetOptions options) {

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

    TableReadOptions.Builder builder = TableReadOptions.newBuilder();

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

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

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

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

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

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

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

BigQuery per Elasticsearch

Beta

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

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

Requisiti per questa pipeline

• Deve essere presente la tabella BigQuery di origine.
• Un host Elasticsearch su un'istanza GCP o su Elastic Cloud con Elasticsearch versione 7.0 o successiva e dovrebbe 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 Base64 utilizzata per l'autenticazione.
index	L'indice Elasticsearch verso cui verranno inviate le richieste, ad esempio my-index.
inputTableSpec	(Facoltativo) Tabella BigQuery da cui inserire l'elemento in Elasticsearch. Devi fornire una tabella o una query. Ad esempio, projectId:datasetId.tablename.
query	(Facoltativo) Query SQL per l'estrazione di dati da BigQuery. Devi fornire una tabella o una query.
useLegacySql	(Facoltativo) Impostato su true per utilizzare l'SQL precedente (applicabile solo se fornisci la query). Valore predefinito: false.
batchSize	(Facoltativo) Dimensioni batch nel numero di documenti. Valore predefinito: 1000.
batchSizeBytes	(Facoltativo) Dimensioni batch in numero di byte. Valore predefinito: 5242880 (5 MB).
maxRetryAttempts	(Facoltativo) Numero massimo di nuovi tentativi, il valore deve essere > 0. Valore predefinito: no retries.
maxRetryDuration	(Facoltativo) La durata massima per il nuovo tentativo, espressa in millisecondi, deve essere > 0. Valore predefinito: no retries.

Esecuzione del modello BigQuery in Elasticsearch

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli flex, devi disporre dell'interfaccia a riga di comando di Google Cloud 284.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: nome della tabella BigQuery.
• CONNECTION_URL: il tuo URL Elasticsearch.
• APIKEY: la chiave API codificata in Base64 per l'autenticazione.
• INDEX: il tuo indice Elasticsearch.

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: nome della tabella BigQuery.
• CONNECTION_URL: il tuo URL 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.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.
 */
public class BigQueryToElasticsearch {

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    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();
  }
}

Bigtable da Avro a Cloud Storage

Il modello Avro da Bigtable a Cloud Storage è una pipeline che legge i dati di 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.
• Prima di eseguire la pipeline, deve esistere il bucket Cloud Storage di output.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable di 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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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-

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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.util.DualInputNestedValueProvider;
import com.google.cloud.teleport.util.DualInputNestedValueProvider.TranslatorInput;
import com.google.protobuf.ByteOutput;
import com.google.protobuf.ByteString;
import com.google.protobuf.UnsafeByteOperations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.SimpleFunction;

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

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

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

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

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

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

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

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

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

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

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

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

    PipelineResult result = run(options);

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

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

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

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

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

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

    return pipeline.run();
  }

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

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

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

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

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

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

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

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

Bigtable su Cloud Storage Parquet

Il modello Bigtable in Cloud Storage Parquet è una pipeline che legge i dati di 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.
• Prima di eseguire la pipeline, deve esistere il bucket Cloud Storage di output.

Parametri del modello

Parametro	Descrizione
bigtableProjectId	L'ID del progetto Google Cloud dell'istanza Bigtable di 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 in Cloud Storage Parquet

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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 java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.AvroCoder;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.PCollection;

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

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

    @Description("The project that contains the table to export.")
    ValueProvider<String> getBigtableProjectId();

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

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

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

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

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

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

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

    @Description(
        "The prefix for each exported file in outputDirectory. Default file prefix is set to"
            + " \"output\".")
    @Default.String("output")
    ValueProvider<String> getFilenamePrefix();

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

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

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

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

    PipelineResult result = run(options);

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

  /**
   * Runs a pipeline to export data from a Cloud Bigtable table to Parquet file(s) in GCS.
   *
   * @param options arguments to the pipeline
   */
  public static PipelineResult run(Options options) {
    Pipeline pipeline = Pipeline.create(PipelineUtils.tweakPipelineOptions(options));
    BigtableIO.Read read =
        BigtableIO.read()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId());

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

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

    return pipeline.run();
  }

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

Bigtable in Cloud Storage SequenzaFile

Il modello Bigtable - File di sequenza di 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 dati da Bigtable in Cloud Storage.

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Prima di eseguire la pipeline, deve esistere il bucket Cloud Storage di output.

Parametri del modello

Parametro	Descrizione
bigtableProject	L'ID del progetto Google Cloud dell'istanza Bigtable di 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 dell'app, Bigtable utilizza il profilo dell'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 della sequenza. Ad esempio, output-.

Esecuzione del modello SequenzaFile da Bigtable a Cloud Storage

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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: 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-

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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: 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à è coperta dai Termini delle offerte pre-GA dei Termini di servizio di Google Cloud. Le funzionalità pre-GA potrebbero avere un supporto limitato e le modifiche alle funzionalità pre-GA potrebbero non essere compatibili con altre versioni pre-GA. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

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

Il modello di testo da Datastore a 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 funzione di questo tipo, ogni riga del 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 le entità da acquisire. Ad esempio, SELECT * FROM MyKind.
datastoreReadProjectId	L'ID progetto di Google Cloud dell'istanza di Datastore da cui 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.
textWritePrefix	Il prefisso del percorso 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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: il nome del tuo bucket Cloud Storage
• DATASTORE_PROJECT_ID: ID del progetto Cloud in cui esiste l'istanza di Datastore
• DATASTORE_KIND: tipo di entità Datastore
• DATASTORE_NAMESPACE: spazio dei nomi delle tue entità Datastore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

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

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

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: il nome del tuo bucket Cloud Storage
• DATASTORE_PROJECT_ID: ID del progetto Cloud in cui esiste l'istanza di Datastore
• DATASTORE_KIND: tipo di entità Datastore
• DATASTORE_NAMESPACE: spazio dei nomi delle tue entità Datastore
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

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

• PATH_TO_JAVASCRIPT_UDF_FILE: L'URI Cloud Storage del file .js che definisce la funzione JavaScript definita dall'utente (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.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
 */
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 Text

Il modello di testo da Firestore a Cloud Storage è 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 funzione di questo tipo, ogni riga del 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 le entità da acquisire. Ad esempio, SELECT * FROM MyKind.
firestoreReadProjectId	L'ID progetto di 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.
textWritePrefix	Il prefisso del percorso 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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: il nome del tuo bucket Cloud Storage
• FIRESTORE_PROJECT_ID: l'ID progetto Cloud in cui è presente l'istanza Firestore
• FIRESTORE_KIND: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

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

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 Spanner, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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: il nome del tuo bucket Cloud Storage
• FIRESTORE_PROJECT_ID: l'ID progetto Cloud in cui è presente l'istanza Firestore
• FIRESTORE_KIND: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

• PATH_TO_JAVASCRIPT_UDF_FILE: L'URI Cloud Storage del file .js che definisce la funzione JavaScript definita dall'utente (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.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
 */
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();
  }
}

Avro da Cloud Spanner a Cloud Storage

Il modello Cloud Spanner in 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 esportato.
• Uno o più file TableName.avro-#####-of-#####.

Ad esempio, quando esporti un database con due tabelle, Singers e Albums, viene creato 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 essere presente il bucket Cloud Storage di output.
• Oltre ai ruoli IAM necessari per eseguire i job Dataflow, devi disporre anche dei ruoli IAM appropriati per leggere i dati di Cloud Spanner e scrivere 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 Cloud Storage in cui vuoi esportare i file Avro. Il job di esportazione crea una nuova directory in questo percorso che contiene 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 RFC 3339 UTC &Zulu". Ad esempio, 1990-12-31T23:59:60Z. Il timestamp deve essere passato e deve essere applicata l'importanza 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 esterne). Se non sono esplicitamente elencati, il flag 'shouldExportRelatedTables' deve essere impostato per un'esportazione riuscita.
shouldExportRelatedTables	(Facoltativo) Il flag utilizzato insieme a 'tableNames' parametro 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 nel modello di file Avro su Cloud Storage

console

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

   Affinché il job venga visualizzato nella pagina Istanze di Spanner della console, 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 istanza Spanner
   • SPANNER_DATABASE_NAME: nome del database Spanner
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 di Cloud Spanner della console, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• GCS_STAGING_LOCATION: percorso per scrivere file temporanei, ad esempio gs://mybucket/temp
• INSTANCE_ID: ID dell'istanza Cloud Spanner
• DATABASE_ID: ID del database Cloud Spanner
• GCS_DIRECTORY: il percorso Cloud Storage da cui sono stati esportati i file Avro in

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• JOB_NAME: un nome job univoco a tua scelta

  Affinché il job venga visualizzato nella parte di Cloud Spanner della console, 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• GCS_STAGING_LOCATION: percorso per scrivere file temporanei, ad esempio gs://mybucket/temp
• INSTANCE_ID: ID dell'istanza Cloud Spanner
• DATABASE_ID: ID del database Cloud Spanner
• GCS_DIRECTORY: il percorso Cloud Storage da cui sono stati esportati i file Avro 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 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. */
public class ExportPipeline {

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

    void setInstanceId(ValueProvider<String> value);

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

    void setDatabaseId(ValueProvider<String> value);

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

    void setOutputDir(ValueProvider<String> value);

    @Description("Temporary Directory to store Avro files.")
    ValueProvider<String> getAvroTempDirectory();

    void setAvroTempDirectory(ValueProvider<String> value);

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

    void setTestJobId(ValueProvider<String> jobId);

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

    void setSpannerHost(ValueProvider<String> value);

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

    void setWaitUntilFinish(boolean value);

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

    void setSnapshotTime(ValueProvider<String> value);

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

    void setSpannerProjectId(ValueProvider<String> value);

    @Description(
        "If true, Timestamps are exported with timestamp-micros logical type."
            + "By default, Timestamps are exported as ISO8601 strings at nanosecond precision.")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportTimestampAsLogicalType();

    void setShouldExportTimestampAsLogicalType(ValueProvider<Boolean> value);

    @Description(
        "If provided, these tables (along with any necessary parent or foreign key"
            + " tables) are exported from Cloud Spanner.")
    @Default.String(value = "")
    ValueProvider<String> getTableNames();

    void setTableNames(ValueProvider<String> value);

    @Description(
        "If true, export not only the specified list of tables, but any related tables necessary"
            + " for the export, such as interleaved parent tables and foreign keys tables")
    @Default.Boolean(false)
    ValueProvider<Boolean> getShouldExportRelatedTables();

    void setShouldExportRelatedTables(ValueProvider<Boolean> value);

    @Description("The spanner priority. --spannerPriority 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 di Google Cloud del database Cloud Spanner da cui 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 sono 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 RFC 3339 UTC &Zulu". Ad esempio, 1990-12-31T23:59:60Z. Il timestamp deve essere passato e deve essere applicata l'importanza massima del timestamp.

Esecuzione del modello Cloud Text to Cloud Spanner

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• SPANNER_PROJECT_ID: ID del progetto Cloud del database Spanner da cui leggere i dati
• DATABASE_ID: ID del database Spanner
• BUCKET_NAME: il nome del tuo bucket Cloud Storage
• INSTANCE_ID: l'ID istanza Spanner
• TABLE_ID: ID della tabella Spanner

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• SPANNER_PROJECT_ID: ID del progetto Cloud del database Spanner da cui leggere i dati
• DATABASE_ID: ID del database Spanner
• BUCKET_NAME: il nome del tuo bucket Cloud Storage
• INSTANCE_ID: l'ID istanza Spanner
• TABLE_ID: 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.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.ReadOperation;
import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
import org.apache.beam.sdk.io.gcp.spanner.Transaction;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.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>
 */
public class SpannerToText {

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

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

    @Description("Temporary Directory to store Csv files.")
    ValueProvider<String> getCsvTempDirectory();

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

    @Description("The spanner priority. --spannerPriority 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 SpannerIO
     * 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 SpannerIO.readAll() instead of SpannerIO.read()
            // because ValueProvider parameters such as table name required for SpannerIO.read()
            // can be read only inside DoFn but SpannerIO.read() is of type
            // PTransform<PBegin, Struct>, which prevents prepending it with DoFn that reads these
            // parameters at the pipeline execution time.
            .apply(
                "Read all records",
                SpannerIO.readAll().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");
  }
}

Avro da Cloud Storage a Bigtable

Il modello Avro di Cloud Storage per Bigtable è una pipeline che legge i dati dai file Avro in un bucket Cloud Storage e scrive i dati in una tabella Bigtable. Puoi utilizzare il modello per copiare 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 Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del file Avro Cloud Storage sul modello Bigtable

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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 Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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 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.common.base.MoreObjects;
import com.google.common.collect.ImmutableList;
import com.google.protobuf.ByteString;
import java.nio.ByteBuffer;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.AvroIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

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

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

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

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

    @Description("If true, a large row is split into multiple MutateRows requests.")
    ValueProvider<Boolean> getSplitLargeRows();

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

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

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

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

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

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

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

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

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

    pipeline
        .apply("Read from Avro", AvroIO.read(BigtableRow.class).from(options.getInputFilePattern()))
        .apply(
            "Transform to Bigtable",
            ParDo.of(
                AvroToBigtableFn.createWithSplitLargeRows(
                    options.getSplitLargeRows(), MAX_MUTATIONS_PER_ROW)))
        .apply("Write to Bigtable", write);

    return pipeline.run();
  }

  /**
   * Translates {@link BigtableRow} to {@link Mutation}s along with a row key. The mutations are
   * {@link SetCell}s that set the value for specified cells with family name, column qualifier and
   * timestamp.
   */
  static class AvroToBigtableFn extends DoFn<BigtableRow, KV<ByteString, Iterable<Mutation>>> {
    private final ValueProvider<Boolean> splitLargeRowsFlag;
    private Boolean splitLargeRows;
    private final int maxMutationsPerRow;

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

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

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

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

    @ProcessElement
    public void processElement(
        @Element BigtableRow row, OutputReceiver<KV<ByteString, Iterable<Mutation>>> out) {
      ByteString key = toByteString(row.getKey());
      // BulkMutation doesn't split rows. Currently, if a single row contains more than 100,000
      // mutations, the service will fail the request.
      ImmutableList.Builder<Mutation> mutations = ImmutableList.builder();
      int cellsProcessed = 0;
      for (BigtableCell cell : row.getCells()) {
        SetCell setCell =
            SetCell.newBuilder()
                .setFamilyName(cell.getFamily().toString())
                .setColumnQualifier(toByteString(cell.getQualifier()))
                .setTimestampMicros(cell.getTimestamp())
                .setValue(toByteString(cell.getValue()))
                .build();

        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
        cellsProcessed++;

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

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

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

Da Cloud Storage Avro a Cloud Spanner

I file Avro Cloud Storage al 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 disporre delle 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 oppure 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 di un modello Avro da Cloud Storage a Cloud Spanner

console

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

   Affinché il job venga visualizzato nella pagina Istanze di Spanner della console, 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 istanza Spanner
   • SPANNER_DATABASE_NAME: nome del database Spanner
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INSTANCE_ID: ID dell'istanza Spanner che contiene il database
• DATABASE_ID: ID del database Spanner in cui eseguire l'importazione
• GCS_DIRECTORY: percorso di Cloud Storage da cui vengono importati i file Avro, ad esempio gs://mybucket/somefolder

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INSTANCE_ID: ID dell'istanza Spanner che contiene il database
• DATABASE_ID: ID del database Spanner in cui eseguire l'importazione
• GCS_DIRECTORY: 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 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. */
public class ImportPipeline {

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

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

    void setInstanceId(ValueProvider<String> value);

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

    void setDatabaseId(ValueProvider<String> value);

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

    void setInputDir(ValueProvider<String> value);

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

    void setSpannerHost(ValueProvider<String> value);

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

    void setWaitForIndexes(ValueProvider<Boolean> value);

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

    void setWaitForForeignKeys(ValueProvider<Boolean> value);

    @Description(
        "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);

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

    void setEarlyIndexCreateFlag(ValueProvider<Boolean> value);

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

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

    void setSpannerProjectId(ValueProvider<String> value);

    void setWaitUntilFinish(boolean value);

    @Description("DDL creation timeout.")
    @Default.Integer(30)
    ValueProvider<Integer> getDDLCreationTimeoutInMinutes();

    void setDDLCreationTimeoutInMinutes(ValueProvider<Integer> value);

    @Description("The spanner priority. --spannerPriority 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 Parquet di Cloud Storage a Bigtable è una pipeline che legge i dati dai file Parquet in un bucket Cloud Storage e scrive i dati in una tabella Bigtable. Puoi utilizzare il modello per copiare 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 Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del file Parquet di Cloud Storage su modello Bigtable

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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 Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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 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.protobuf.ByteString;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.runners.direct.DirectRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.parquet.ParquetIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.MoreObjects;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

  private static final Boolean DEFAULT_SPLIT_LARGE_ROWS = false;

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

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

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

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

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

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

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

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

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

    void setSplitLargeRows(ValueProvider<Boolean> splitLargeRows);
  }

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

    PipelineResult result = run(options);
  }

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

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

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

    return pipeline.run();
  }

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

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

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

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

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

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

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

      // BulkMutation doesn't split rows. Currently, if a single row contains more than 100,000
      // mutations, the service will fail the request.
      ImmutableList.Builder<Mutation> mutations = ImmutableList.builder();
      List<Object> cells = (List) ctx.element().get(1);
      int cellsProcessed = 0;
      for (Object element : cells) {
        Mutation.SetCell setCell = null;
        if (runner.isAssignableFrom(DirectRunner.class)) {
          setCell =
              Mutation.SetCell.newBuilder()
                  .setFamilyName(((GenericData.Record) element).get(0).toString())
                  .setColumnQualifier(
                      toByteString((ByteBuffer) ((GenericData.Record) element).get(1)))
                  .setTimestampMicros((Long) ((GenericData.Record) element).get(2))
                  .setValue(toByteString((ByteBuffer) ((GenericData.Record) element).get(3)))
                  .build();
        } else {
          BigtableCell bigtableCell = (BigtableCell) element;
          setCell =
              Mutation.SetCell.newBuilder()
                  .setFamilyName(bigtableCell.getFamily().toString())
                  .setColumnQualifier(toByteString(bigtableCell.getQualifier()))
                  .setTimestampMicros(bigtableCell.getTimestamp())
                  .setValue(toByteString(bigtableCell.getValue()))
                  .build();
        }
        mutations.add(Mutation.newBuilder().setSetCell(setCell).build());
        cellsProcessed++;

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

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

SequenzaFile Cloud Storage in Bigtable

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

Requisiti per questa pipeline:

• La tabella Bigtable deve esistere.
• Prima di eseguire la pipeline, il file di sequenza di input deve esistere in un bucket Cloud Storage.
• L'input SequenzaFile 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 dell'app, Bigtable utilizza il profilo dell'app predefinito dell'istanza.
sourcePattern	Il pattern del percorso Cloud Storage in cui si trovano i dati. Ad esempio, gs://mybucket/somefolder/prefix*.

Esecuzione del modello di sequenza di Cloud Storage in Bigtable

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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: ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• SOURCE_PATTERN: pattern del percorso Cloud Storage in cui si trovano i dati, ad esempio gs://mybucket/somefolder/prefix*

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• BIGTABLE_PROJECT_ID: l'ID del progetto Google Cloud dell'istanza Bigtable di 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: ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione
• SOURCE_PATTERN: pattern del percorso 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 Cloud Storage in BigQuery

La pipeline Cloud Text to BigQuery è una pipeline batch che consente di leggere i file di testo archiviati in Cloud Storage, trasformarli mediante 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 sia presente 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 Text to BigQuery non supporta l'importazione dei dati nei campi STRUCT (Record) della tabella BigQuery di destinazione.

  Il seguente JSON descrive un esempio di schema BigQuery:

  {
    "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 definita dall'utente JavaScript che intendi utilizzare. Ad esempio, se il codice funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di 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:// del 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 per archiviare 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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

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

• PATH_TO_BIGQUERY_SCHEMA_JSON: percorso 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 JavaScript definita dall'utente (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• PATH_TO_TEXT_DATA: percorso di Cloud Storage al tuo set di dati di testo
• BIGQUERY_TABLE: nome della tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: percorso di Cloud Storage alla directory temporanea

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• JAVASCRIPT_FUNCTION: Il nome della funzione definita dall'utente JavaScript che vuoi utilizzare

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

• PATH_TO_BIGQUERY_SCHEMA_JSON: percorso 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 JavaScript definita dall'utente (UDF) che vuoi utilizzare, ad esempio gs://my-bucket/my-udfs/my_file.js
• PATH_TO_TEXT_DATA: percorso di Cloud Storage al tuo set di dati di testo
• BIGQUERY_TABLE: nome della tabella BigQuery
• PATH_TO_TEMP_DIR_ON_GCS: percorso di Cloud Storage alla 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.templates.common.BigQueryConverters;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.JavascriptTextTransformerOptions;
import com.google.cloud.teleport.templates.common.JavascriptTextTransformer.TransformTextViaJavascript;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

    void setInputFilePattern(ValueProvider<String> value);

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

    void setJSONPath(ValueProvider<String> value);

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

    void setOutputTable(ValueProvider<String> value);

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

    void setJavascriptTextTransformGcsPath(ValueProvider<String> jsTransformPath);

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

    void setJavascriptTextTransformFunctionName(
        ValueProvider<String> javascriptTextTransformFunctionName);

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

    void setBigQueryLoadingTemporaryDirectory(ValueProvider<String> directory);
  }

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

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

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

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

                          @Override
                          public TableSchema apply(String jsonPath) {

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

                            try {

                              jsonSchema = schemaParser.parseSchema(jsonPath);

                              JSONArray bqSchemaJsonArray =
                                  jsonSchema.getJSONArray(BIGQUERY_SCHEMA);

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

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

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

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

    pipeline.run();
  }
}

Testo Cloud Storage in Datastore [Deprecato]

Beta

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

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

Il modello Cloud Storage Text in Datastore è una pipeline batch che legge dai file di testo archiviati in Cloud Storage e scrive 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 inseriti nel 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.
datastoreWriteProjectId	L'ID progetto Google Cloud in cui scrivere le entità di Datastore
datastoreHintNumWorkers	(Facoltativo) Suggerimento per il numero previsto di worker nel passaggio di limitazione della velocità effettiva di 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 Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome univoco.
3. (Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. L'endpoint regionale predefinito è us-central1.

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• PATH_TO_INPUT_TEXT_FILES: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

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

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• PATH_TO_INPUT_TEXT_FILES: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

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

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.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.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.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.values.TupleTag;

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

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

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

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

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

    Pipeline pipeline = Pipeline.create(options);

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

    pipeline.run();
  }
}

Testo Cloud Storage in Firestore

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

Requisiti per questa pipeline:

• Firestore 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 inseriti nel 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.
firestoreWriteProjectId	L'ID progetto Google Cloud in cui scrivere le entità Firestore
firestoreHintNumWorkers	(Facoltativo) Suggerimento per il numero previsto di worker nel passaggio di limitazione della rampa di Firestore. 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 in Firestore

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/GCS_Text_to_Firestore \
    --region REGION_NAME \
    --parameters \
textReadPattern=PATH_TO_INPUT_TEXT_FILES,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
firestoreWriteProjectId=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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• PATH_TO_INPUT_TEXT_FILES: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

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

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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_Firestore
{
   "jobName": "JOB_NAME",
   "parameters": {
       "textReadPattern": "PATH_TO_INPUT_TEXT_FILES",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "firestoreWriteProjectId": "PROJECT_ID",
       "errorWritePath": "ERROR_FILE_WRITE_PATH"
   },
   "environment": { "zone": "us-central1-f" }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• PATH_TO_INPUT_TEXT_FILES: il 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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.

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

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.templates.common.DatastoreConverters.DatastoreWriteOptions;
import com.google.cloud.teleport.templates.common.DatastoreConverters.WriteJsonEntities;
import com.google.cloud.teleport.templates.common.ErrorConverters.ErrorWriteOptions;
import com.google.cloud.teleport.templates.common.ErrorConverters.LogErrors;
import com.google.cloud.teleport.templates.common.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.FilesystemReadOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.values.TupleTag;

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

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

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

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

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

    Pipeline pipeline = Pipeline.create(options);

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

    pipeline.run();
  }
}

Testo Cloud Storage in Pub/Sub (batch)

Questo modello crea una pipeline batch che legge i record dei file di testo archiviati in Cloud Storage e li pubblica in un argomento Pub/Sub. Il modello può essere utilizzato per pubblicare record in un file delimitato da nuova riga contenente record JSON o file CSV in un argomento Pub/Sub per l'elaborazione in tempo reale. Puoi utilizzare questo modello per riprodurre i dati in Pub/Sub.

Questo modello non imposta un timestamp per i singoli record. Il tempo dell'evento equivale al tempo di pubblicazione durante l'esecuzione. Se la tua pipeline si basa su un tempo di evento preciso per l'elaborazione, non devi utilizzare questa pipeline.

Requisiti per questa pipeline:

• I file da leggere devono essere in formato JSON delimitato da nuova riga o in formato CSV. I record che coprono più righe nei file di origine potrebbero causare problemi a valle perché ogni riga all'interno dei file verrà pubblicata come messaggio in Pub/Sub.
• L'argomento Pub/Sub deve esistere prima dell'esecuzione della pipeline.

Parametri del modello

Parametro	Descrizione
inputFilePattern	Il pattern del file di input da cui leggere. Ad esempio, gs://bucket-name/files/*.json.
outputTopic	L'argomento di input di Pub/Sub in cui scrivere. Il nome deve essere nel formato projects/<project-id>/topics/<topic-name>.

Esecuzione del modello di testo Cloud Storage in Pub/Sub (batch)

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• TOPIC_NAME: nome dell'argomento Pub/Sub
• BUCKET_NAME: il nome del tuo bucket Cloud Storage

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern": "gs://BUCKET_NAME/files/*.json",
       "outputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME"
   },
   "environment": { "zone": "us-central1-f" }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• TOPIC_NAME: nome dell'argomento Pub/Sub
• BUCKET_NAME: il nome del tuo bucket Cloud Storage

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 org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.options.ValueProvider;

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

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

    void setInputFilePattern(ValueProvider<String> value);

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

    void setOutputTopic(ValueProvider<String> value);
  }

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

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

    run(options);
  }

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

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

    return pipeline.run();
  }
}

Testo Cloud Storage in Cloud Spanner

Il modello Cloud Storage Text to Cloud Spanner è una pipeline batch che legge i file di testo CSV da Cloud Storage e li importa in un database Cloud Spanner.

Requisiti per questa pipeline:

• Deve esistere il database e la tabella Cloud Spanner di destinazione.
• Devi disporre delle autorizzazioni di lettura per il bucket Cloud Storage e le autorizzazioni di scrittura per il database Cloud Spanner di destinazione.
• Deve esistere il percorso di Cloud Storage di input che contiene i file CSV.
• Devi creare un file manifest di importazione contenente una descrizione JSON dei file CSV e archiviarlo in Cloud Storage.
• Se il database Cloud Spanner di destinazione ha già uno schema, le colonne specificate nel file manifest devono avere gli stessi tipi di dati delle colonne corrispondenti nello schema del database di destinazione.

• Il file manifest, codificato in ASCII o UTF-8, deve corrispondere al seguente formato:

  Formato ed esempio del file manifest

  Il formato del file manifest corrisponde al seguente tipo di messaggio, mostrato qui nel formato buffer di protocollo:

  message ImportManifest {
    // The per-table import manifest.
    message TableManifest {
      // Required. The name of the destination table.
      string table_name = 1;
      // Required. The CSV files to import. This value can be either a filepath or a glob pattern.
      repeated string file_patterns = 2;
      // The schema for a table column.
      message Column {
        // Required for each Column that you specify. The name of the column in the
        // destination table.
        string column_name = 1;
        // Required for each Column that you specify. The type of the column.
        string type_name = 2;
      }
      // Optional. The schema for the table columns.
      repeated Column columns = 3;
    }
    // Required. The TableManifest of the tables to be imported.
    repeated TableManifest tables = 1;
  
    enum ProtoDialect {
      GOOGLE_STANDARD_SQL = 0;
      POSTGRESQL = 1;
    }
    // Optional. The dialect of the receiving database. Defaults to GOOGLE_STANDARD_SQL.
    ProtoDialect dialect = 2;
  }

  L'esempio seguente mostra un file manifest per l'importazione di tabelle chiamate Albums e Singers in un database del dialetto SQL standard di Google. La tabella Albums utilizza lo schema di colonne che il job recupera dal database, mentre la tabella Singers utilizza lo schema specificato dal file manifest:

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

• I file di testo da importare devono essere in formato CSV, con codifica ASCII o UTF-8. Ti consigliamo di non utilizzare il byte order mark (BOM) nei file con codifica UTF-8.
• I dati devono corrispondere a uno dei seguenti tipi:

  SQL standard di Google

      BOOL
      INT64
      FLOAT64
      NUMERIC
      STRING
      DATE
      TIMESTAMP
      BYPES
      JSON

  PostgreSQL

      boolean
      bigint
      double precision
      numeric
      character varying, text
      date
      timestamp with time zone
      bytea

Nota: se non specifichi i nomi di colonna e i tipi di dati della tabella di destinazione nel file manifest di importazione, le colonne nei file CSV devono essere nello stesso ordine delle colonne nel database di destinazione. Puoi vedere l'ordine delle colonne nella tabella eseguendo la seguente query:

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

Avviso: se il database di Cloud Spanner di destinazione non è vuoto, i nuovi dati potrebbero sovrascrivere i dati preesistenti.

Parametri del modello

Parametro	Descrizione
instanceId	L'ID istanza del database Cloud Spanner.
databaseId	L'ID del database Cloud Spanner.
importManifest	Percorso di Cloud Storage al file manifest di importazione.
columnDelimiter	Il delimitatore di colonna utilizzato dal file di origine. Il valore predefinito è ,.
fieldQualifier	Il carattere che deve racchiudere qualsiasi valore nel file di origine che contiene columnDelimiter. Il valore predefinito è ".
trailingDelimiter	Specifica se le righe nei file di origine hanno delimitatori finali (ovvero, se il carattere columnDelimiter appare alla fine di ogni riga, dopo l'ultimo valore della colonna). Il valore predefinito è true.
escape	Il carattere di escape utilizzato dal file di origine. Per impostazione predefinita, questo parametro non è impostato e il modello non utilizza il carattere di escape.
nullString	La stringa che rappresenta un valore NULL. Per impostazione predefinita, questo parametro non è impostato e il modello non utilizza la stringa null.
dateFormat	Il formato utilizzato per analizzare le colonne della data. Per impostazione predefinita, la pipeline tenta di analizzare le colonne della data come yyyy-M-d[' 00:00:00'], ad esempio 2019-01-31 o 2019-1-1 00:00:00. Se il formato della data è diverso, specifica il formato utilizzando i pattern java.time.format.DateTimeFormatter.
timestampFormat	Il formato utilizzato per analizzare le colonne del timestamp. Se il timestamp è un numero intero lungo, viene analizzato come ora Unix. In caso contrario, viene analizzato come una stringa utilizzando il formato java.time.format.DateTimeFormatter.ISO_INSTANT. Per altri casi, specifica la tua stringa di pattern, ad esempio utilizzando MMM dd yyyy HH:mm:ss.SSSVV per i timestamp nel formato "Jan 21 1998 01:02:03.456+08:00".

Se devi utilizzare formati di data e timestamp personalizzati, assicurati che siano modelli java.time.format.DateTimeFormatter validi. La tabella seguente mostra altri esempi di formati personalizzati per le colonne delle date e dei timestamp:

Tipo	Valore di input	Formato	Commento
DATE	2011-3-31		Per impostazione predefinita, il modello può analizzare questo formato. Non è necessario specificare il parametro dateFormat.
DATE	2022-3-31 00:00:00		Per impostazione predefinita, il modello può analizzare questo formato. Non devi specificare il formato. Se vuoi, puoi usare yyyy-M-d' 00:00:00'.
DATE	1 aprile 18	gg MMM, yy
DATE	Mercoledì 3 aprile 2019 AD	EEEE, LLLL g, aaaa G
TIMESTAMP	2019-01-02T11:22:33Z 2019-01-02T11:22:33.123Z 2019-01-02T11:22:33.12356789Z		Il formato predefinito ISO_INSTANT può analizzare questo tipo di timestamp. Non è necessario fornire il parametro timestampFormat.
TIMESTAMP	1568402363		Per impostazione predefinita, il modello può analizzare questo tipo di timestamp e considerarlo come tempo Unix.
TIMESTAMP	mar 3 giu 2008 11:05:30 GMT	EEE, g MMM aaaa HH:mm:ss VV
TIMESTAMP	31/12/2018 110530.123PST	aaaa/mm/gg HHmmss.SSSz
TIMESTAMP	2019-01-02T11:22:33Z o 2019-01-02T11:22:33.123Z	aaaa-MM-gg'T'HH:mm:ss[.SSS]VV	Se la colonna di input è una combinazione di 2019-01-02T11:22:33Z e 2019-01-02T11:22:33.123Z, il formato predefinito può analizzare questo tipo di timestamp. Non è necessario fornire un parametro di formattazione personalizzato. Puoi utilizzare yyyy-MM-dd'T'HH:mm:ss[.SSS]VV per gestire entrambi i casi. Non puoi utilizzare yyyy-MM-dd'T'HH:mm:ss[.SSS]'Z', perché postfix 'Z' deve essere analizzato come ID del fuso orario, non come valore letterale carattere. Internamente, la colonna timestamp è convertita in java.time.Instant. Di conseguenza, deve essere specificato nel fuso orario UTC o avere informazioni sul fuso orario associate. La data e l'ora locali, ad esempio 2019-01-02 11:22:33, non possono essere analizzate come valori java.time.Instant validi.

Esecuzione dei file di testo da Cloud Storage a un modello Cloud Spanner

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

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

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli classic, devi disporre dell'interfaccia a riga di comando di Google Cloud 138.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

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

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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INSTANCE_ID: ID dell'istanza Cloud Spanner
• DATABASE_ID: ID del database Cloud Spanner
• GCS_PATH_TO_IMPORT_MANIFEST: percorso Cloud Storage del file manifest di importazione

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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_Cloud_Spanner
{
   "jobName": "JOB_NAME",
   "parameters": {
       "instanceId": "INSTANCE_ID",
       "databaseId": "DATABASE_ID",
       "importManifest": "GCS_PATH_TO_IMPORT_MANIFEST"
   },
   "environment": {
       "machineType": "n1-standard-2"
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INSTANCE_ID: ID dell'istanza Cloud Spanner
• DATABASE_ID: ID del database Cloud Spanner
• GCS_PATH_TO_IMPORT_MANIFEST: percorso Cloud Storage del file manifest di importazione

Codice sorgente del modello

Java

Il codice sorgente di questo modello si trova nel repository GoogleCloudPlatform/DataflowTemplates su GitHub.

Da Cloud Storage a Elasticsearch

Beta

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

Il modello Cloud Storage to Elasticsearch è una pipeline batch che legge i dati dei file CSV archiviati in un bucket Cloud Storage e li scrive in documenti Elasticsearch come documenti JSON.

Requisiti per questa pipeline:

• Deve esistere il bucket Cloud Storage.
• Deve esistere un host Elasticsearch su un'istanza GCP o su Elasticsearch Cloud accessibile da Dataflow.
• Deve esistere una tabella BigQuery per l'output di errore.

Parametri del modello

Parametro	Descrizione
inputFileSpec	Il pattern di file Cloud Storage per cercare file CSV. Esempio: gs://mybucket/test-*.csv.
connectionUrl	URL Elasticsearch nel formato https://hostname:[port] o specifica CloudID se utilizzi Elastic Cloud.
apiKey	Chiave API codificata Base64 utilizzata per l'autenticazione.
index	L'indice Elasticsearch per il quale saranno inviate le richieste, ad esempio my-index.
deadletterTable	Tabella Dead BigQuery di BigQuery per l'invio di inserimenti non riusciti. Esempio: <your-project>:<your-dataset>.<your-table-name>.
containsHeaders	(Facoltativo) Valore booleano che indica se le intestazioni sono incluse nel file CSV. Valore predefinito: true.
delimiter	(Facoltativo) Il delimitatore utilizzato dal file CSV. Esempio: ,
csvFormat	(Facoltativo) Il formato CSV in base al formato CSV Apache Commons. Valore predefinito: Default.
jsonSchemaPath	(Facoltativo) Percorso dello schema JSON. Valore predefinito: null.
largeNumFiles	(Facoltativo) Imposta su true se il numero di file è compreso tra decine di migliaia. Valore predefinito: false.
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 funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di UDF JavaScript di esempio, consulta Esempi di UDF.
batchSize	(Facoltativo) Dimensioni batch nel numero di documenti. Valore predefinito: 1000.
batchSizeBytes	(Facoltativo) Dimensioni batch in numero di byte. Valore predefinito: 5242880 (5 MB).
maxRetryAttempts	(Facoltativo) Numero massimo di nuovi tentativi, il valore deve essere > 0. Predefinito: nessun nuovo tentativo.
maxRetryDuration	(Facoltativo) La durata massima per il nuovo tentativo, espressa in millisecondi, deve essere > 0. Predefinito: nessun nuovo tentativo.
csvFileEncoding	(Facoltativo) Codifica dei file CSV.

Esecuzione del modello Cloud Storage to Elasticsearch

console

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

   Per un elenco delle aree geografiche in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.

4. Dal menu a discesa Modello Dataflow, seleziona the Cloud Storage to Elasticsearch template.
5. Inserisci i valori dei parametri nei campi forniti.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare l'interfaccia a riga di comando di Google Cloud per eseguire i modelli flex, devi disporre dell'interfaccia a riga di comando di Google Cloud 284.0.0 o versioni successive.

Esegui il modello nella shell o nel terminale:

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID\
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/GCS_To_Elasticsearch \
    --parameters \
inputFileSpec=INPUT_FILE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX,\
deadletterTable=DEADLETTER_TABLE,\

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs:// Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INPUT_FILE_SPEC: pattern del file Cloud Storage.
• CONNECTION_URL: il tuo URL Elasticsearch.
• APIKEY: la chiave API codificata in Base64 per l'autenticazione.
• INDEX: il tuo indice Elasticsearch.
• DEADLETTER_TABLE: la tabella BigQuery.

Server

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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": {
          "inputFileSpec": "INPUT_FILE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX",
          "deadletterTable": "DEADLETTER_TABLE"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/GCS_To_Elasticsearch",
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto cloud in cui vuoi eseguire il job Spanner
• 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 puoi trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://Dataflow-templates/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. I tuoi ambienti di produzione devono utilizzare 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 Spanner, ad esempio us-central1
• INPUT_FILE_SPEC: pattern del file Cloud Storage.
• CONNECTION_URL: il tuo URL Elasticsearch.
• APIKEY: la chiave API codificata in Base64 per l'autenticazione.
• INDEX: il tuo indice Elasticsearch.
• DEADLETTER_TABLE: la tabella BigQuery.

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 static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;

import com.google.cloud.teleport.v2.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.elasticsearch.options.GCSToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.CsvConverters;
import com.google.cloud.teleport.v2.transforms.ErrorConverters.WriteStringMessageErrors;
import com.google.cloud.teleport.v2.utils.SchemaUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.NullableCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.WithTimestamps;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link GCSToElasticsearch} pipeline exports data from one or more CSV files in Cloud Storage
 * to Elasticsearch.
 *
 * <p>Please refer to <b><a href=
 * "https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/master/v2/googlecloud-to-elasticsearch/docs/GCSToElasticsearch/README.md">
 * README.md</a></b> for further information.
 */
public class GCSToElasticsearch {

  /** The tag for the headers of the CSV if required. */
  static final TupleTag<String> CSV_HEADERS = new TupleTag<String>() {};

  /** The tag for the lines of the CSV. */
  static final TupleTag<String> CSV_LINES = new TupleTag<String>() {};

  /** The tag for the dead-letter output of the udf. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** The tag for the main output for the UDF. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

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

  /** String/String Coder for FailsafeElement. */
  private static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(
          NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(StringUtf8Coder.of()));

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

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  private static PipelineResult run(GCSToElasticsearchOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    // Register the coder for pipeline
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    // Throw error if containsHeaders is true and a schema or Udf is also set.
    if (options.getContainsHeaders()) {
      checkArgument(
          options.getJavascriptTextTransformGcsPath() == null
              && options.getJsonSchemaPath() == null,
          "Cannot parse file containing headers with UDF or Json schema.");
    }

    // Throw error if only one retry configuration parameter is set.
    checkArgument(
        (options.getMaxRetryAttempts() == null && options.getMaxRetryDuration() == null)