Questo modello crea una pipeline in modalità flusso che esegue continuamente il polling dei nuovi file di testo caricati in Cloud Storage, legge ogni file riga per riga e pubblica stringhe in un argomento Pub/Sub. Il modello pubblica i 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.
La pipeline viene eseguita a tempo indeterminato e deve essere terminata manualmente tramite "cancel" e non "drain", a causa dell'utilizzo della trasformazione "Watch", che è un "SplittableDoFn" che non supporta lo svuotamento.
Attualmente, l'intervallo di polling è fisso e impostato su 10 secondi. Questo modello non imposta alcun timestamp sui singoli record, quindi l'ora dell'evento corrisponde a quella di pubblicazione durante l'esecuzione. Se la tua pipeline si basa su una data/ora degli eventi precisa per l'elaborazione, non devi utilizzarla.
Requisiti della pipeline
- I file di input devono essere in formato JSON o CSV delimitato da nuova riga. I record su più righe dei file di origine possono causare problemi downstream, poiché ogni riga all'interno dei file viene pubblicata come un messaggio in Pub/Sub.
- L'argomento Pub/Sub deve esistere prima dell'esecuzione.
- La pipeline è in esecuzione a tempo indeterminato e deve essere arrestata manualmente.
Parametri del modello
Parametri obbligatori
- inputFilePattern : il pattern del file di input da cui leggere. (Esempio: gs://nome-bucket/files/*.json).
- outputTopic : l'argomento di input Pub/Sub in cui scrivere. Il nome deve essere nel formato
projects/<PROJECT_ID>/topics/<TOPIC_NAME>
. ad esempio projects/your-project-id/topics/your-topic-name.
Esegui il modello
Console
- Vai alla pagina Crea job da modello di Dataflow. Vai a Crea job da modello
- Nel campo Nome job, inserisci un nome univoco per il job.
- (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. La regione predefinita è
us-central1
.Per un elenco delle regioni in cui puoi eseguire un job Dataflow, vedi Località Dataflow.
- Nel menu a discesa Modello Dataflow, seleziona the Text Files on Cloud Storage to Pub/Sub (Stream) template.
- Nei campi dei parametri forniti, inserisci i valori dei parametri.
- (Facoltativo) Per passare dall'elaborazione "exactly-once" alla modalità di streaming "at-least-once", seleziona Almeno una volta.
- Fai clic su Esegui job.
gcloud
Nella shell o nel terminale, esegui il modello:
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
Sostituisci quanto segue:
JOB_NAME
: un nome job univoco a tua sceltaREGION_NAME
: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempious-central1
STAGING_LOCATION
: la posizione per la gestione temporanea dei file locali (ad esempio,gs://your-bucket/staging
)TOPIC_NAME
: nome dell'argomento Pub/SubBUCKET_NAME
: il nome del bucket Cloud StorageFILE_PATTERN
: il glob del pattern di file da cui leggere nel bucket Cloud Storage (ad esempio,path/*.csv
)
API
Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui relativi ambiti di autorizzazione, consulta 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" } }
Sostituisci quanto segue:
PROJECT_ID
: l'ID del progetto Google Cloud in cui vuoi eseguire il job DataflowJOB_NAME
: un nome job univoco a tua sceltaLOCATION
: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempious-central1
STAGING_LOCATION
: la posizione per la gestione temporanea dei file locali (ad esempio,gs://your-bucket/staging
)TOPIC_NAME
: nome dell'argomento Pub/SubBUCKET_NAME
: il nome del bucket Cloud StorageFILE_PATTERN
: il glob del pattern di file da cui leggere nel bucket Cloud Storage (ad esempio,path/*.csv
)
Passaggi successivi
- Scopri di più sui modelli Dataflow.
- Consulta l'elenco dei modelli forniti da Google.