Da protocollo Pub/Sub a modello BigQuery

Il modello da protocollo Pub/Sub a BigQuery è una pipeline in modalità flusso che importa i dati di protocollo da una sottoscrizione Pub/Sub a una tabella BigQuery. Eventuali errori che si verificano durante la scrittura nella tabella BigQuery vengono inviati in flussi a un argomento Pub/Sub non elaborato.

È possibile fornire una funzione JavaScript definita dall'utente (UDF) per trasformare i dati. Gli errori durante l'esecuzione della funzione definita dall'utente possono essere inviati a un argomento Pub/Sub separato o allo stesso argomento non elaborato degli errori di BigQuery.

Requisiti della pipeline

  • Deve esistere la sottoscrizione Pub/Sub di input.
  • Il file dello schema per i record Proto deve esistere in Cloud Storage.
  • Deve esistere l'argomento Pub/Sub di output.
  • Deve esistere il set di dati BigQuery di output.
  • Se la tabella BigQuery esiste, deve avere uno schema corrispondente ai dati di protocollo indipendentemente dal valore createDisposition.

Parametri del modello

Parametro Descrizione
protoSchemaPath Il percorso Cloud Storage del file di schema proto autonomo. Ad esempio, gs://path/to/my/file.pb. Questo file può essere generato con il flag --descriptor_set_out del comando protoc. Il flag --include_imports garantisce che il file sia indipendente.
fullMessageName Il nome completo del messaggio proto. Ad esempio, package.name.MessageName, dove package.name è il valore fornito per l'istruzione package e non per l'istruzione java_package.
inputSubscription La sottoscrizione di input Pub/Sub da cui leggere. Ad esempio, projects/<project>/subscriptions/<subscription>.
outputTopic L'argomento Pub/Sub da utilizzare per i record non elaborati. Ad esempio, projects/<project-id>/topics/<topic-name>.
outputTableSpec La posizione della tabella di output BigQuery. Ad esempio, my-project:my_dataset.my_table. A seconda del valore createDisposition specificato, la tabella di output potrebbe essere creata automaticamente utilizzando il file dello schema di input.
preserveProtoFieldNames (Facoltativo) true per mantenere il nome del campo Proto originale in JSON. false per utilizzare nomi JSON più standard. Ad esempio, false modificherebbe field_name in fieldName. (valore predefinito: false)
bigQueryTableSchemaPath (Facoltativo) Percorso Cloud Storage del percorso dello schema BigQuery. Ad esempio, gs://path/to/my/schema.json. Se non viene fornito, lo schema viene dedotto dallo schema Proto.
javascriptTextTransformGcsPath (Facoltativo) L'URI Cloud Storage del file .js che definisce la funzione definita dall'utente JavaScript che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facoltativo) Il nome della funzione definita dall'utente di JavaScript che vuoi utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di funzioni JavaScript definite, consulta gli esempi di funzioni definite dall'utente.
javascriptTextTransformReloadIntervalMinutes (Facoltativo) Specifica la frequenza di ricarica della funzione definita dall'utente, in minuti. Se il valore è maggiore di 0, Dataflow controlla periodicamente il file UDF in Cloud Storage e ricarica la funzione definita dall'utente se il file viene modificato. Questo parametro consente di aggiornare la funzione definita dall'utente mentre la pipeline è in esecuzione, senza dover riavviare il job. Se il valore è 0, il ricaricamento delle funzioni definite dall'utente è disabilitato. Il valore predefinito è 0.
udfOutputTopic (Facoltativo) L'argomento Pub/Sub in cui sono archiviati gli errori della funzione definita dall'utente. Ad esempio, projects/<project-id>/topics/<topic-name>. Se non viene specificato, gli errori delle funzioni definite dall'utente vengono inviati allo stesso argomento di outputTopic.
writeDisposition (Facoltativo) L'oggetto WriteDisposition di BigQuery. Ad esempio, WRITE_APPEND, WRITE_EMPTY o WRITE_TRUNCATE. Valore predefinito: WRITE_APPEND.
createDisposition (Facoltativo) L'oggetto CreateDisposition di BigQuery. Ad esempio, CREATE_IF_NEEDED, CREATE_NEVER. Valore predefinito: CREATE_IF_NEEDED.
useStorageWriteApi Facoltativo: se true, la pipeline utilizza l' API BigQuery Storage Write. Il valore predefinito è false. Per ulteriori informazioni, consulta la pagina relativa all' utilizzo dell'API Storage Write.
useStorageWriteApiAtLeastOnce Facoltativo: quando si utilizza l'API Storage Write, specifica la semantica di scrittura. Per utilizzare la semantica at-least-once, imposta questo parametro su true. Per usare la semantica "exactly-once", imposta il parametro su false. Questo parametro si applica solo quando useStorageWriteApi è true. Il valore predefinito è false.
numStorageWriteApiStreams Facoltativo: quando si utilizza l'API Storage Write, specifica il numero di flussi di scrittura. Se useStorageWriteApi è true e useStorageWriteApiAtLeastOnce è false, devi impostare questo parametro.
storageWriteApiTriggeringFrequencySec Facoltativo: quando utilizzi l'API Storage Write, specifica la frequenza di attivazione in secondi. Se useStorageWriteApi è true e useStorageWriteApiAtLeastOnce è false, devi impostare questo parametro.

Funzione definita dall'utente

Facoltativamente, puoi estendere questo modello scrivendo una funzione definita dall'utente (UDF). Il modello chiama la funzione definita dall'utente per ogni elemento di input. I payload degli elementi sono serializzati come stringhe JSON. Per ulteriori informazioni, consulta Creare funzioni definite dall'utente per i modelli Dataflow.

Specifica della funzione

La funzione definita dall'utente ha le seguenti specifiche:

  • Input: il campo dei dati del messaggio Pub/Sub, serializzato come stringa JSON.
  • Output: una stringa JSON corrispondente allo schema della tabella di destinazione BigQuery.

    • Esegui il modello

    Console

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

      Per un elenco delle regioni in cui puoi eseguire un job Dataflow, vedi Località Dataflow.

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

    gcloud

    Nella shell o nel terminale, esegui il modello:

    gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/PubSub_Proto_to_BigQuery_Flex \
    --parameters \
schemaPath=SCHEMA_PATH,\
fullMessageName=PROTO_MESSAGE_NAME,\
inputSubscription=SUBSCRIPTION_NAME,\
outputTableSpec=BIGQUERY_TABLE,\
outputTopic=UNPROCESSED_TOPIC

    Sostituisci quanto segue:

    • JOB_NAME: un nome job univoco a tua scelta
    • REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
    • VERSION: la versione del modello che vuoi utilizzare

      Puoi utilizzare i seguenti valori:

    • SCHEMA_PATH: il percorso Cloud Storage del file di schema Proto (ad esempio, gs://MyBucket/file.pb)
    • PROTO_MESSAGE_NAME: il nome del messaggio Proto (ad esempio, package.name.MessageName)
    • SUBSCRIPTION_NAME: nome della sottoscrizione di input Pub/Sub
    • BIGQUERY_TABLE: nome della tabella di output BigQuery
    • UNPROCESSED_TOPIC: l'argomento Pub/Sub da utilizzare per la coda non elaborata

    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/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/PubSub_Proto_to_BigQuery_Flex",
      "parameters": {
          "schemaPath": "SCHEMA_PATH",
          "fullMessageName": "PROTO_MESSAGE_NAME",
          "inputSubscription": "SUBSCRIPTION_NAME",
          "outputTableSpec": "BIGQUERY_TABLE",
          "outputTopic": "UNPROCESSED_TOPIC"
      }
   }
}

    Sostituisci quanto segue:

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

      Puoi utilizzare i seguenti valori:

    • SCHEMA_PATH: il percorso Cloud Storage del file di schema Proto (ad esempio, gs://MyBucket/file.pb)
    • PROTO_MESSAGE_NAME: il nome del messaggio Proto (ad esempio, package.name.MessageName)
    • SUBSCRIPTION_NAME: nome della sottoscrizione di input Pub/Sub
    • BIGQUERY_TABLE: nome della tabella di output BigQuery
    • UNPROCESSED_TOPIC: l'argomento Pub/Sub da utilizzare per la coda non elaborata

    Passaggi successivi