Da Pub/Sub a modello Redis

Il modello da Pub/Sub a Redis è una pipeline di flusso che legge i messaggi da un di sottoscrizione Pub/Sub e scrive il payload del messaggio su Redis. L'uso più comune di questo modello è l'esportazione dei log in Redis Enterprise per l'analisi avanzata dei log basata sulla ricerca in tempo reale.

  • Prima di scrivere in Redis, puoi applicare una funzione definita dall'utente JavaScript al payload del messaggio.
  • I messaggi che presentano errori di elaborazione vengono inoltrati a un argomento Pub/Sub non elaborato per un'ulteriore risoluzione dei problemi e rielaborazione.
  • Per maggiore sicurezza, abilita una connessione SSL durante la configurazione della connessione dell'endpoint del database. Questo modello non supporta TLS reciprocamente.

Requisiti della pipeline

  • La sottoscrizione Pub/Sub di origine deve esistere prima dell'esecuzione della pipeline.
  • L'argomento Pub/Sub non elaborato deve esistere prima dell'esecuzione della pipeline.
  • L'endpoint del database Redis deve essere accessibile dal server Dataflow una subnet.

Parametri del modello

Parametri obbligatori

  • inputSubscription : l'abbonamento Pub/Sub da cui leggere l'input, nel formato projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_ID>. ad esempio projects/your-project-id/subscriptions/your-subscription-name).
  • redisHost : l'host del database Redis. ad esempio your.cloud.db.redislabs.com. Il valore predefinito è 127.0.0.1.
  • redisPort: la porta del database Redis. Esempio: 12345. Il valore predefinito è 6379.
  • redisPassword : la password del database Redis. Il campo predefinito è vuoto.

Parametri facoltativi

  • sslEnabled: il parametro SSL del database Redis. Il valore predefinito è false.
  • redisSinkType : il sink Redis. I valori supportati sono STRING_SINK, HASH_SINK, STREAMS_SINK, and LOGGING_SINK. (Esempio: STRING_SINK). Il valore predefinito è: STRING_SINK.
  • connectionTimeout: il timeout della connessione Redis in millisecondi. (ad es. 2000). Il valore predefinito è 2000.
  • ttl : la scadenza della chiave in secondi. Il valore predefinito di ttl per HASH_SINK è -1, il che significa che non scade mai.
  • javascriptTextTransformGcsPath: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente (UDF) JavaScript da utilizzare. Esempio: gs://my-bucket/my-udfs/my_file.js.
  • javascriptTextTransformFunctionName : il nome della funzione definita dall'utente (UDF) JavaScript da 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 (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
  • javascriptTextTransformReloadIntervalMinutes: specifica la frequenza con cui ricaricare la UDF, 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 ti consente di aggiornare la UDF durante l'esecuzione della pipeline, senza dover riavviare il job. Se il valore è 0, il ricaricamento della funzione definita dall'utente viene disabilitato. Il valore predefinito è 0.

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 vengono serializzati come stringhe JSON. Per ulteriori informazioni, vedi Crea funzioni definite dall'utente per i modelli Dataflow.

Specifica della funzione

La UDF ha la seguente specifica:

  • Input: stringa JSON
  • Output: una stringa o un oggetto JSON con formato di stringa

Esegui il modello

Console

  1. Vai alla pagina Crea job da modello di Dataflow.
  2. Vai a Crea job da modello
  3. Nel campo Nome job, inserisci un nome univoco per il job.
  4. (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.

  5. Nel menu a discesa Modello di flusso di dati, seleziona the Pub/Sub to Redis template.
  6. Inserisci i valori parametro negli appositi campi.
  7. Fai clic su Esegui job.

gcloud

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Cloud_PubSub_to_Redis \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       inputSubscription=INPUT_SUBSCRIPTION,\
       redisHost=REDIS_HOST,\
       redisPort=REDIS_PORT,\
       redisPassword=REDIS_PASSWORD,\

Sostituisci quanto segue:

  • JOB_NAME: un nome job univoco di tua scelta
  • VERSION: la versione del modello che vuoi utilizzare

    Puoi utilizzare i seguenti valori:

  • REGION_NAME: la regione in cui vuoi di eseguire il deployment del job Dataflow, ad esempio us-central1
  • INPUT_SUBSCRIPTION: la sottoscrizione di input Pub/Sub
  • REDIS_HOST: l'host del database Redis
  • REDIS_PORT: la porta del database Redis
  • REDIS_PASSWORD: la password del database Redis

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "inputSubscription": "INPUT_SUBSCRIPTION",
       "redisHost": "REDIS_HOST",
       "redisPort": "REDIS_PORT",
       "redisPassword": "REDIS_PASSWORD",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Cloud_PubSub_to_Redis",
     "environment": { "maxWorkers": "10" }
  }
}

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto Google Cloud in cui vuoi eseguire il job Dataflow
  • JOB_NAME: un nome job univoco di tua scelta
  • VERSION: la versione del modello che vuoi utilizzare

    Puoi utilizzare i seguenti valori:

  • LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
  • INPUT_SUBSCRIPTION: la sottoscrizione di input Pub/Sub
  • REDIS_HOST: l'host del database Redis
  • REDIS_PORT: la porta del database Redis
  • REDIS_PASSWORD: la password del database Redis

Passaggi successivi