Creare funzioni definite dall'utente per i modelli Dataflow

Alcuni modelli Dataflow forniti da Google supportano funzioni definite dall'utente. Le funzioni definite dall'utente consentono di estendere la funzionalità un modello senza modificare il codice del modello.

Panoramica

Per creare una funzione definita dall'utente, scrivi una funzione JavaScript o Python, a seconda nel modello. Archivia il file di codice UDF in Cloud Storage e specifica la location come parametro del modello. Per ogni elemento di input, il modello chiama la tua funzione. La funzione trasforma l'elemento o esegue altre applicazioni personalizzate e restituisce il risultato al modello.

Ad esempio, potresti utilizzare una funzione definita dall'utente per:

• Riformattare i dati di input in modo che corrispondano a uno schema target.
• Oscura i dati sensibili.
• Filtra alcuni elementi dall'output.

L'input della funzione funzione definita dall'utente è un singolo elemento di dati, serializzato come JSON stringa. La funzione restituisce come output una stringa JSON serializzata. Il formato dei dati dipende dal modello. Ad esempio, nel Sottoscrizione Pub/Sub a BigQuery l'input è costituito dai dati dei messaggi Pub/Sub serializzati come oggetto JSON, e l'output è un oggetto JSON serializzato che rappresenta un Riga della tabella BigQuery. Per ulteriori informazioni, consulta documentazione per ogni modello.

Esegui un modello con una funzione definita dall'utente

Per eseguire un modello con una funzione definita dall'utente, devi specificare il percorso di Cloud Storage del file JavaScript e il nome della funzione come parametri di modello.

Nota: i nomi dei parametri pertinenti dipendono dal modello. Consulta le documentazione per ogni modello.

Con alcuni modelli forniti da Google, puoi anche creare la funzione definita dall'utente direttamente nella Console Google Cloud, come segue:

1. Vai alla pagina Dataflow nella console Google Cloud.

   Vai alla pagina Dataflow

2. Fai clic su Crea job da modello.

3. Seleziona il modello fornito da Google che vuoi eseguire.

4. Espandi Parametri facoltativi. Se il modello supporta le funzioni definite dall'utente, per la posizione di Cloud Storage della funzione definita dall'utente e di un altro per il nome della funzione.

5. Accanto al parametro del modello, fai clic su Crea funzione definita dall'utente.

   Nota: questo pulsante non è disponibile per tutti i modelli. Se questo pulsante non è visibile, quindi crea la funzione definita dall'utente in un file locale e carica il file Cloud Storage.

6. Nel riquadro Seleziona o crea una funzione definita dall'utente:

   1. Inserisci un nome file. Esempio: my_udf.js.
   2. Seleziona una cartella di Cloud Storage. Esempio: gs://your-bucket/your-folder.
   3. Usa l'editor di codice in linea per scrivere la funzione. L'editore è precompilato con codice boilerplate, che puoi utilizzare come punto di partenza.

   4. Fai clic su Crea funzione definita dall'utente.

      La console Google Cloud salva il file della funzione definita dall'utente e compila la posizione di Cloud Storage.

   5. Inserisci il nome della funzione nel campo corrispondente.

Scrivi una funzione JavaScript definita dall'utente

Il seguente codice mostra una funzione JavaScript definita dall'utente da cui puoi iniziare:

/*
 * @param {string} inJson input JSON message (stringified)
 * @return {?string} outJson output JSON message (stringified)
 */
function process(inJson) {
  const obj = JSON.parse(inJson);

  // Example data transformations:
  // Add a field: obj.newField = 1;
  // Modify a field: obj.existingField = '';
  // Filter a record: return null;

  return JSON.stringify(obj);
}

Il codice JavaScript viene eseguito Motore JavaScript Nashorn. I nostri suggerimenti testerai la funzione definita dall'utente sul motore Nashorn prima di eseguirne il deployment. Il motore Nashorn non corrisponde esattamente all'implementazione Node.js di JavaScript. Un comune problema è l'utilizzo di console.log() o Number.isNaN(), nessuno dei due è definiti nel motore Nashorn.

Puoi testare la funzione definita dall'utente sul motore Nashorn utilizzando Cloud Shell, JDK 11 preinstallato. Avvia Nashorn in modalità interattiva nel seguente modo:

jjs --language=es6

Nella shell interattiva di Nashorn, segui questi passaggi:

1. Richiama load per caricare il file JavaScript UDF.
2. Definisci un oggetto JSON di input in base ai messaggi previsti della pipeline.
3. Utilizza la funzione JSON.stringify per serializzare l'input in una stringa JSON.
4. Chiama la funzione UDF per elaborare la stringa JSON.
5. Chiama JSON.parse per deserializzare l'output.
6. Verifica il risultato.

Esempio:

> load('my_udf.js')
> var input = {"name":"user1"}
> var output = process(JSON.stringify(input))
> print(output)

Scrivi una funzione definita dall'utente Python

Il codice seguente mostra una funzione Python no-op da cui puoi iniziare:

import json
def process(value):
  # Load the JSON string into a dictionary.
  data = json.loads(value)

  # Transform the data in some way.
  data['new_field'] = 'new_value'

  # Serialize the data back to JSON.
  return json.dumps(data)

Le funzioni definite dall'utente di Python supportano i pacchetti di dipendenze standard di Python Apache Beam. Non può utilizzare pacchetti di terze parti.

Gestione degli errori

In genere, quando si verifica un errore durante l'esecuzione della funzione definita dall'utente, l'errore viene scritto in una posizione dei messaggi non recapitabili. I dettagli dipendono dal modello. Ad esempio, Sottoscrizione Pub/Sub a BigQuery modello crea una tabella _error_records e scrive gli errori lì. Funzione definita dall'utente per il runtime a causa di errori di sintassi o eccezioni non rilevate. Per verificare errori di sintassi, testa la funzione definita dall'utente localmente.

Puoi generare in modo programmatico un'eccezione per un elemento che non dovrebbe essere elaborati. In questo caso, l'elemento viene scritto nella posizione dei messaggi non recapitabili, se il modello ne supporta uno. Per un esempio di questo approccio, consulta Eventi di percorso.

Esempi di casi d'uso

Questa sezione descrive alcuni pattern comuni per le funzioni definite dall'utente, basati su un uso reale d'uso diversi.

Arricchisci gli eventi

Utilizzare una funzione definita dall'utente per arricchire gli eventi con nuovi campi per ulteriori informazioni contestuali.

Esempio:

 function process(inJson) {
  const data = JSON.parse(inJson);

  // Add new field to track data source
  data.source = "source1";
  return JSON.stringify(data);
}

Trasformare gli eventi

Utilizza una funzione definita dall'utente per trasformare l'intero formato dell'evento a seconda di ciò che la destinazione desiderata.

L'esempio seguente ripristina una voce di log di Cloud Logging (LogEntry) al log originale quando disponibile. (a seconda della sorgente log, la stringa di log originale è a volte compilate nel campo textPayload. Puoi usare questo pattern per inviare i log non elaborati nel formato originale, invece di inviare l'intero LogEntry da Cloud Logging.

 function process(inJson) {
  const data = JSON.parse(inJson);

  if (data.textPayload) {
    return data.textPayload; // Return string value, and skip JSON.stringify
  }
 return JSON.stringify(obj);
}

Oscura o rimuovi i dati sugli eventi

Utilizza una funzione definita dall'utente per oscurare o rimuovere una parte dell'evento.

L'esempio seguente oscura il nome del campo sensitiveField sostituendo il relativo e rimuove completamente il campo denominato redundantField.

 function process(inJson) {
  const data = JSON.parse(inJson);

  // Normalize existing field values
  data.source = (data.source && data.source.toLowerCase()) || "unknown";

  // Redact existing field values
  if (data.sensitiveField) {
    data.sensitiveField = "REDACTED";
  }

  // Remove existing fields
  if (data.redundantField) {
    delete(data.redundantField);
  }

  return JSON.stringify(data);
}

Eventi di percorso

Utilizza una funzione definita dall'utente per indirizzare gli eventi a destinazioni separate nel sink a valle.

L'esempio seguente, in base al Da Pub/Sub a Splunk del modello, indirizza ogni evento all'indice Splunk corretto. Viene chiamato un modello funzione locale per mappare gli eventi agli indici.

function process(inJson) {
  const obj = JSON.parse(inJson);
  
  // Set index programmatically for data segregation in Splunk
  obj._metadata = {
    index: splunkIndexLookup(obj)
  }
  return JSON.stringify(obj);
}  

L'esempio successivo instrada gli eventi non riconosciuti alla coda dei messaggi non recapitabili, presupponendo che il modello supporta una coda di messaggi non recapitabili. (Ad esempio, vedi il Da Pub/Sub a JDBC template.) Potresti usare questo pattern per filtrare le voci impreviste prima scrivendo nella destinazione.

 function process(inJson) {
  const data = JSON.parse(inJson);

  // Route unrecognized events to the deadletter topic
  if (!data.hasOwnProperty('severity')) {
    throw new Error("Unrecognized event. eventId='" + data.Id + "'");
  }

  return JSON.stringify(data);

Filtra eventi

Utilizzare una funzione definita dall'utente per filtrare gli eventi indesiderati o non riconosciuti dall'output.

L'esempio seguente elimina gli eventi dove data.severity è uguale a "DEBUG".

 function process(inJson) {
  const data = JSON.parse(inJson);

  // Drop events with certain field values
  if (data.severity == "DEBUG") {
    return null;
  }

  return JSON.stringify(data);
}

Passaggi successivi

• Modelli forniti da Google
• Creazione ed esecuzione di un modello flessibile
• Esecuzione di modelli classici
• Estendi il tuo modello Dataflow con le funzioni definite dall'utente (post del blog)
• UDF di esempio (GitHub)