Il modello BigQuery-Elasticsearch è una pipeline batch che importa i dati da una tabella BigQuery in Elasticsearch come documenti. Il modello può leggere l'intera tabella o record specifici utilizzando una query fornita.
Requisiti della pipeline
- La tabella BigQuery di origine deve esistere.
- Un host Elasticsearch su un'istanza Google Cloud o su Elastic Cloud con Elasticsearch versione 7.0 o successive. Deve essere accessibile dalle macchine worker di Dataflow.
Parametri del modello
Parametri obbligatori
- connectionUrl: l'URL di Elasticsearch nel formato
https://hostname:[port]
. Se utilizzi Elastic Cloud, specifica CloudID. Ad esempio:https://elasticsearch-host:9200
. - apiKey: la chiave API codificata in Base64 da utilizzare per l'autenticazione.
- index: l'indice Elasticsearch a cui vengono inviate le richieste. Ad esempio,
my-index
.
Parametri facoltativi
- inputTableSpec: la tabella BigQuery da cui leggere. Se specifichi
inputTableSpec
, il modello legge i dati direttamente dallo spazio di archiviazione BigQuery utilizzando l'API BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Per informazioni sulle limitazioni dell'API Storage di lettura, consulta https://cloud.google.com/bigquery/docs/reference/storage#limitations. Devi specificareinputTableSpec
oquery
. Se imposti entrambi i parametri, il modello utilizza il parametroquery
. Ad esempio,<BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>
. - outputDeadletterTable: la tabella BigQuery per i messaggi che non sono riusciti a raggiungere la tabella di output. Se una tabella non esiste, viene creata durante l'esecuzione della pipeline. Se non specificato, viene utilizzato
<outputTableSpec>_error_records
. Ad esempio:<PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>
. - query: la query SQL da utilizzare per leggere i dati da BigQuery. Se il set di dati BigQuery si trova in un progetto diverso dal job Dataflow, specifica il nome completo del set di dati nella query SQL, ad esempio <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Per impostazione predefinita, il parametro
query
utilizza GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a meno cheuseLegacySql
non siatrue
. Devi specificareinputTableSpec
oquery
. Se imposti entrambi i parametri, il modello utilizza il parametroquery
. Ad esempio,select * from sampledb.sample_table
. - useLegacySql: impostalo su
true
per utilizzare l'SQL precedente. Questo parametro si applica solo quando viene utilizzato il parametroquery
. Il valore predefinito èfalse
. - queryLocation: necessario per la lettura da una vista autorizzata senza l'autorizzazione della tabella sottostante. Ad esempio,
US
. - queryTempDataset: con questa opzione, puoi impostare un set di dati esistente per creare la tabella temporanea in cui archiviare i risultati della query. Ad esempio,
temp_dataset
. - KMSEncryptionKey: se leggi da BigQuery utilizzando l'origine query, utilizza questa chiave Cloud KMS per criptare le tabelle temporanee create. Ad esempio,
projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key
. - elasticsearchUsername: il nome utente Elasticsearch per l'autenticazione. Se specificato, il valore di
apiKey
viene ignorato. - elasticsearchPassword: la password di Elasticsearch per l'autenticazione. Se specificato, il valore di
apiKey
viene ignorato. - batchSize: le dimensioni del batch in numero di documenti. Il valore predefinito è
1000
. - batchSizeBytes: le dimensioni del batch in numero di byte. Il valore predefinito è
5242880
(5 MB). - maxRetryAttempts: il numero massimo di tentativi. Deve essere maggiore di zero. Il valore predefinito è
no retries
. - maxRetryDuration: la durata massima dei tentativi in millisecondi. Deve essere maggiore di zero. Il valore predefinito è
no retries
. - propertyAsIndex: la proprietà nel documento in fase di indicizzazione il cui valore specifica i metadati
_index
da includere con il documento nelle richieste collettive. Ha la precedenza su una UDF_index
. Il valore predefinito ènone
. - javaScriptIndexFnGcsPath: il percorso Cloud Storage all'origine della funzione definita dall'utente JavaScript per una funzione che specifica i metadati
_index
da includere con il documento nelle richieste collettive. Il valore predefinito ènone
. - javaScriptIndexFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati
_index
da includere nel documento nelle richieste collettive. Il valore predefinito ènone
. - propertyAsId: una proprietà nel documento in fase di indicizzazione il cui valore specifica i metadati
_id
da includere nel documento nelle richieste collettive. Ha la precedenza su una UDF_id
. Il valore predefinito ènone
. - javaScriptIdFnGcsPath: il percorso Cloud Storage all'origine della funzione JavaScript definita dall'utente per la funzione che specifica i metadati
_id
da includere nel documento nelle richieste collettive. Il valore predefinito ènone
. - javaScriptIdFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati
_id
da includere con il documento nelle richieste collettive. Il valore predefinito ènone
. - javaScriptTypeFnGcsPath: il percorso Cloud Storage all'origine della funzione definita dall'utente JavaScript per una funzione che specifica i metadati
_type
da includere con i documenti nelle richieste collettive. Il valore predefinito ènone
. - javaScriptTypeFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati
_type
da includere nel documento nelle richieste collettive. Il valore predefinito ènone
. - javaScriptIsDeleteFnGcsPath: il percorso Cloud Storage all'origine della funzione definita dall'utente JavaScript per la funzione che determina se eliminare il documento anziché inserirlo o aggiornarlo. La funzione restituisce un valore stringa di
true
ofalse
. Il valore predefinito ènone
. - javaScriptIsDeleteFnName: il nome della funzione JavaScript definita dall'utente che determina se eliminare il documento anziché inserirlo o aggiornarlo. La funzione restituisce un valore stringa di
true
ofalse
. Il valore predefinito ènone
. - usePartialUpdate: indica se utilizzare aggiornamenti parziali (aggiornamento anziché creazione o indicizzazione, consentendo documenti parziali) con le richieste Elasticsearch. Il valore predefinito è
false
. - bulkInsertMethod: indica se utilizzare
INDEX
(indice, consente upsert) oCREATE
(crea, errori in caso di _id duplicato) con le richieste collettive Elasticsearch. Il valore predefinito èCREATE
. - trustSelfSignedCerts: indica se considerare attendibile o meno il certificato autofirmato. Un'istanza Elasticsearch installata potrebbe avere un certificato autofirmato. Imposta questo valore su true per ignorare la convalida del certificato SSL. (Il valore predefinito è
false
). - disableCertificateValidation: se
true
, considera attendibile il certificato SSL autofirmato. Un'istanza Elasticsearch potrebbe avere un certificato autofirmato. Per ignorare la convalida del certificato, imposta questo parametro sutrue
. Il valore predefinito èfalse
. - apiKeyKMSEncryptionKey: la chiave Cloud KMS per decriptare la chiave API. Questo parametro è obbligatorio se
apiKeySource
è impostato suKMS
. Se viene fornito questo parametro, trasmetti una stringaapiKey
criptata. Cripta i parametri utilizzando l'endpoint di crittografia dell'API KMS. Per la chiave, utilizza il formatoprojects/<PROJECT_ID>/locations/<KEY_REGION>/keyRings/<KEY_RING>/cryptoKeys/<KMS_KEY_NAME>
. Consulta: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt. Ad esempio,projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name
. - apiKeySecretId: l'ID secret di Secret Manager per l'apiKey. Se
apiKeySource
è impostato suSECRET_MANAGER
, fornisci questo parametro. Utilizza il formatoprojects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example,
projects/your-project-id/secrets/your-secret/versions/your-secret-version`. - apiKeySource: l'origine della chiave API. I valori consentiti sono
PLAINTEXT
,KMS
eSECRET_MANAGER
. Questo parametro è obbligatorio quando utilizzi Secret Manager o KMS. SeapiKeySource
è impostato suKMS
, devono essere fornitiapiKeyKMSEncryptionKey
e l'apiKey criptata. SeapiKeySource
è impostato suSECRET_MANAGER
, deve essere fornitoapiKeySecretId
. SeapiKeySource
è impostato suPLAINTEXT
, deve essere fornitoapiKey
. Il valore predefinito è PLAINTEXT. - socketTimeout: se impostato, sovrascrive il timeout massimo predefinito per i tentativi e il timeout predefinito del socket (30.000 ms) in Elastic RestClient.
- javascriptTextTransformGcsPath: l'URI Cloud Storage del file .js che definisce la funzione definita dall'utente (UDF) JavaScript da utilizzare. Ad 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, vedi Esempi di UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
Funzioni definite dall'utente
Questo modello supporta le funzioni definite dall'utente (UDF) in diversi punti della pipeline, descritti di seguito. Per ulteriori informazioni, consulta Creare funzioni definite dall'utente per i modelli Dataflow.
Funzione Index
Restituisce l'indice a cui appartiene il documento.
Parametri del modello:
javaScriptIndexFnGcsPath
: l'URI Cloud Storage del file JavaScript.javaScriptIndexFnName
: il nome della funzione JavaScript.
Specifiche della funzione:
- Input: il documento Elasticsearch, serializzato come stringa JSON.
- Output: il valore del campo dei metadati
_index
del documento.
Funzione ID documento
Restituisce l'ID documento.
Parametri del modello:
javaScriptIdFnGcsPath
: l'URI Cloud Storage del file JavaScript.javaScriptIdFnName
: il nome della funzione JavaScript.
Specifiche della funzione:
- Input: il documento Elasticsearch, serializzato come stringa JSON.
- Output: il valore del campo dei metadati
_id
del documento.
Funzione di eliminazione dei documenti
Specifica se eliminare un documento. Per utilizzare questa funzione, imposta la modalità di inserimento collettivo su INDEX
e fornisci una funzione ID documento.
Parametri del modello:
javaScriptIsDeleteFnGcsPath
: l'URI Cloud Storage del file JavaScript.javaScriptIsDeleteFnName
: il nome della funzione JavaScript.
Specifiche della funzione:
- Input: il documento Elasticsearch, serializzato come stringa JSON.
- Output: restituisci la stringa
"true"
per eliminare il documento o"false"
per inserirlo.
Funzione Tipo di mappatura
Restituisce il tipo di mappatura del documento.
Parametri del modello:
javaScriptTypeFnGcsPath
: l'URI Cloud Storage del file JavaScript.javaScriptTypeFnName
: il nome della funzione JavaScript.
Specifiche della funzione:
- Input: il documento Elasticsearch, serializzato come stringa JSON.
- Output: il valore del campo dei metadati
_type
del documento.
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, consulta Località di Dataflow.
- Dal menu a discesa Modello di dataflow, seleziona the BigQuery to Elasticsearch template.
- Nei campi dei parametri forniti, inserisci i valori dei parametri.
- Fai clic su Esegui job.
gcloud
Nella shell o nel terminale, esegui il modello:
gcloud dataflow flex-template run JOB_NAME \ --project=PROJECT_ID \ --region=REGION_NAME \ --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/BigQuery_to_Elasticsearch \ --parameters \ inputTableSpec=INPUT_TABLE_SPEC,\ connectionUrl=CONNECTION_URL,\ apiKey=APIKEY,\ index=INDEX
Sostituisci quanto segue:
PROJECT_ID
: l'ID progetto Google Cloud in cui vuoi eseguire il job DataflowJOB_NAME
: un nome univoco del job a tua sceltaREGION_NAME
: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempious-central1
VERSION
: la versione del modello che vuoi utilizzarePuoi utilizzare i seguenti valori:
latest
per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel 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 si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
INPUT_TABLE_SPEC
: il nome della tabella BigQuery.CONNECTION_URL
: l'URL di Elasticsearch.APIKEY
: la chiave API con codifica Base64 per l'autenticazione.INDEX
: il tuo indice Elasticsearch.
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 { "launch_parameter": { "jobName": "JOB_NAME", "parameters": { "inputTableSpec": "INPUT_TABLE_SPEC", "connectionUrl": "CONNECTION_URL", "apiKey": "APIKEY", "index": "INDEX" }, "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/BigQuery_to_Elasticsearch", } }
Sostituisci quanto segue:
PROJECT_ID
: l'ID progetto Google Cloud in cui vuoi eseguire il job DataflowJOB_NAME
: un nome univoco del job a tua sceltaLOCATION
: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempious-central1
VERSION
: la versione del modello che vuoi utilizzarePuoi utilizzare i seguenti valori:
latest
per utilizzare l'ultima versione del modello, disponibile nella cartella principale senza data nel 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 si trova nidificata nella rispettiva cartella principale con data nel bucket: gs://dataflow-templates-REGION_NAME/
INPUT_TABLE_SPEC
: il nome della tabella BigQuery.CONNECTION_URL
: l'URL di Elasticsearch.APIKEY
: la chiave API con codifica Base64 per l'autenticazione.INDEX
: il tuo indice Elasticsearch.
Passaggi successivi
- Scopri di più sui modelli Dataflow.
- Consulta l'elenco dei modelli forniti da Google.