Modelo de ficheiros de texto no Cloud Storage para o Pub/Sub (stream)

Este modelo cria um pipeline de streaming que procura continuamente novos ficheiros de texto carregados para o Cloud Storage, lê cada ficheiro linha a linha e publica strings num tópico do Pub/Sub. O modelo publica registos num ficheiro delimitado por newline que contém registos JSON ou num ficheiro CSV num tópico do Pub/Sub para processamento em tempo real. Pode usar este modelo para repetir dados no Pub/Sub.

O pipeline é executado indefinidamente e tem de ser terminado manualmente através de um "cancel" e não de um "drain", devido à sua utilização da transformação "Watch", que é um "SplittableDoFn" que não suporta a drenagem.

Atualmente, o intervalo de sondagem é fixo e está definido para 10 segundos. Este modelo não define nenhuma data/hora nos registos individuais, pelo que a hora do evento é igual à hora de publicação durante a execução. Se o seu pipeline depender de uma hora do evento precisa para o processamento, não deve usar este pipeline.

Este modelo é afetado pelo problema do Beam 18459 e é provável que tenha problemas de memória em contentores grandes (observados em contentores com 5 milhões de ficheiros). Seguem-se algumas soluções alternativas conhecidas:

1. Divida os ficheiros em dois ou mais contentores e use este modelo.
2. Crie um modelo personalizado, possivelmente ramificado a partir deste, que transmita nomes de ficheiros através do Pub/Sub e processe esses ficheiros.

Requisitos do pipeline

• Os ficheiros de entrada têm de estar no formato JSON ou CSV delimitado por newline. Os registos que abrangem várias linhas nos ficheiros de origem podem causar problemas a jusante, porque cada linha nos ficheiros é publicada como uma mensagem no Pub/Sub.
• O tópico Pub/Sub tem de existir antes da execução.
• O pipeline é executado indefinidamente e tem de ser terminado manualmente.

Parâmetros de modelos

Parâmetros obrigatórios

• inputFilePattern: o padrão do ficheiro de entrada a partir do qual ler. Por exemplo, gs://bucket-name/files/*.json.
• outputTopic: o tópico de entrada do Pub/Sub para o qual escrever. O nome tem de estar no formato projects/<PROJECT_ID>/topics/<TOPIC_NAME>. Por exemplo, projects/your-project-id/topics/your-topic-name.

Execute o modelo

Consola

1. Aceda à página do fluxo de dados Criar tarefa a partir de um modelo.
Aceda a Criar tarefa a partir de modelo
2. No campo Nome da tarefa, introduza um nome exclusivo para a tarefa.
3. Opcional: para Ponto final regional, selecione um valor no menu pendente. A região predefinida é us-central1.

   Para ver uma lista das regiões onde pode executar uma tarefa do Dataflow, consulte as localizações do Dataflow.

4. No menu pendente Modelo do fluxo de dados, selecione the Text Files on Cloud Storage to Pub/Sub (Stream) template.
5. Nos campos de parâmetros fornecidos, introduza os valores dos parâmetros.
6. Opcional: para mudar do processamento exatamente uma vez para o modo de streaming pelo menos uma vez, selecione Pelo menos uma vez.
7. Clique em Executar tarefa.

gcloud

Nota: para usar a CLI do Google Cloud para executar modelos classic, tem de ter a versão CLI do Google Cloud 138.0.0 ou posterior.

Na shell ou no terminal, execute o modelo:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/Stream_GCS_Text_to_Cloud_PubSub \
    --region REGION_NAME\
    --staging-location STAGING_LOCATION\
    --parameters \
inputFilePattern=gs://BUCKET_NAME/FILE_PATTERN,\
outputTopic=projects/PROJECT_ID/topics/TOPIC_NAME

Substitua o seguinte:

• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• REGION_NAME: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• STAGING_LOCATION: a localização para organizar ficheiros locais (por exemplo, gs://your-bucket/staging)
• TOPIC_NAME: o nome do seu tópico do Pub/Sub
• BUCKET_NAME: o nome do seu contentor do Cloud Storage
• FILE_PATTERN: o padrão glob do ficheiro a ler no contentor do Cloud Storage (por exemplo, path/*.csv)

API

Para executar o modelo através da API REST, envie um pedido HTTP POST. Para mais informações sobre a API e os respetivos âmbitos de autorização, consulte projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/Stream_GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "environment": {
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputFilePattern": "gs://BUCKET_NAME/FILE_PATTERN",
       "outputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME"
   }
}

Substitua o seguinte:

• PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• LOCATION: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• STAGING_LOCATION: a localização para organizar ficheiros locais (por exemplo, gs://your-bucket/staging)
• TOPIC_NAME: o nome do seu tópico do Pub/Sub
• BUCKET_NAME: o nome do seu contentor do Cloud Storage
• FILE_PATTERN: o padrão glob do ficheiro a ler no contentor do Cloud Storage (por exemplo, path/*.csv)

Código-fonte do modelo

Java

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

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.templates.TextToPubsub.Options;
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.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Watch;
import org.joda.time.Duration;

/**
 * The {@code TextToPubsubStream} is a streaming version of {@code TextToPubsub} pipeline that
 * publishes records to Cloud Pub/Sub from a set of files. The pipeline continuously polls for new
 * files, reads them row-by-row and publishes each record as a string message. The polling interval
 * is fixed and equals to 10 seconds. At the moment, publishing messages with attributes is
 * unsupported.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v1/README_Stream_GCS_Text_to_Cloud_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Stream_GCS_Text_to_Cloud_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "Text Files on Cloud Storage to Pub/Sub",
    description = {
      "This template creates a streaming pipeline that continuously polls for new text files uploaded to Cloud Storage, reads each file line by line, and publishes strings to a Pub/Sub topic. "
          + "The template publishes records in a newline-delimited file containing JSON records or CSV file to a Pub/Sub topic for real-time processing. "
          + "You can use this template to replay data to Pub/Sub.\n",
      "The pipeline runs indefinitely and needs to be terminated manually via a <a href=\"https://cloud.google.com/dataflow/docs/guides/stopping-a-pipeline#cancel\">cancel</a> and not a <a href=\"https://cloud.google.com/dataflow/docs/guides/stopping-a-pipeline#drain\">drain</a>, due to its use of the <code>Watch</code> transform, which is a <code>SplittableDoFn</code> that does not support draining.\n",
      "Currently, the polling interval is fixed and set to 10 seconds. This template does not set any timestamp on the individual records, so the event time is equal to the publishing time during execution. "
          + "If your pipeline relies on an accurate event time for processing, you should not use this pipeline."
    },
    optionsClass = Options.class,
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/text-to-pubsub-stream",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "Input files must be in newline-delimited JSON or CSV format. Records that span multiple lines in the source files can cause issues downstream, because each line within the files is published as a message to Pub/Sub.",
      "The Pub/Sub topic must exist prior to execution.",
      "The pipeline runs indefinitely and needs to be terminated manually.",
    },
    streaming = true,
    supportsAtLeastOnce = true,
    supportsExactlyOnce = true)
public class TextToPubsubStream {
  private static final Duration DEFAULT_POLL_INTERVAL = Duration.standardSeconds(10);

  /**
   * 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())
                .watchForNewFiles(DEFAULT_POLL_INTERVAL, Watch.Growth.never()))
        .apply("Write to PubSub", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

O que se segue?

• Saiba mais sobre os modelos do Dataflow.
• Consulte a lista de modelos fornecidos pela Google.