Modello da protocollo Pub/Sub a BigQuery

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

È possibile fornire una funzione JavaScript definita dall'utente 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 di 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 del protocollo, indipendentemente dal valore createDisposition.

Parametri del modello

Parametro Descrizione
protoSchemaPath Il percorso di 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 autonomo.
fullMessageName Il nome completo del messaggio di protocollo. 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 di schema di input.
preserveProtoFieldNames (Facoltativo) true per conservare il nome del campo Proto originale in JSON. false per utilizzare più nomi JSON standard. Ad esempio, false cambierebbe 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 JavaScript definita dall'utente (UDF) che vuoi utilizzare. Ad esempio: gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facoltativo) Il nome della funzione definita dall'utente (UDF) 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 dall'utente, 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 delle funzioni definite dall'utente 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 della funzione definita dall'utente viene disabilitato. Il valore predefinito è 0.
udfOutputTopic (Facoltativo) L'argomento Pub/Sub in cui sono archiviati gli errori delle funzioni definite dall'utente. Ad esempio, projects/<project-id>/topics/<topic-name>. Se non viene fornito, gli errori delle funzioni definite dall'utente vengono inviati allo stesso argomento di outputTopic.
writeDisposition (Facoltativo) WriteDisposition di BigQuery. Ad esempio, WRITE_APPEND, WRITE_EMPTY o WRITE_TRUNCATE. Valore predefinito: WRITE_APPEND.
createDisposition (Facoltativo) 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 Writer. Il valore predefinito è false. Per ulteriori informazioni, consulta Utilizzo dell'API StorageWrite.
useStorageWriteApiAtLeastOnce (Facoltativo) Quando utilizzi l'API StorageWrite, specifica la semantica della scrittura. Per utilizzare la semantica almeno una volta, imposta questo parametro su true. Per utilizzare 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 Scrivi, specifica il numero di flussi di scrittura. Se useStorageWriteApi è true e useStorageWriteApiAtLeastOnce è false, devi impostare questo parametro.
storageWriteApiTriggeringFrequencySec (Facoltativo) Quando utilizzi l'API Storage Scrivi, 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 dall'utente. 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 la seguente specifica:

  • Input: il campo dati del messaggio Pub/Sub, serializzato come stringa JSON.
  • Output: una stringa JSON che corrisponde 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 di regioni in cui è possibile eseguire un job Dataflow, consulta Località di Dataflow.

    4. Dal menu a discesa Modello Dataflow, seleziona the Pub/Sub Proto to BigQuery template.
    5. Inserisci i valori parametro negli appositi campi.
    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:

      • latest per utilizzare la versione più recente del modello, disponibile nella cartella padre non con data del bucket: gs://dataflow-templates-REGION_NAME/latest/
      • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che è possibile trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
    • 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: il 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:

      • latest per utilizzare la versione più recente del modello, disponibile nella cartella padre non con data del bucket: gs://dataflow-templates-REGION_NAME/latest/
      • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che è possibile trovare nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
    • 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: il 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