Pianifica attività Spark e Spark SQL personalizzate

Dataplex Universal Catalog supporta la pianificazione dell'esecuzione di codice personalizzato, come esecuzione una tantum, in base a una pianificazione regolare o on demand. On demand è in anteprima ed è disponibile solo tramite API. Puoi pianificare le trasformazioni dei dati dei clienti utilizzando Spark (Java), PySpark (limitato alla versione 3.2 di Spark) o Spark SQL. Il Catalogo universale Dataplex esegue il codice utilizzando l'elaborazione Spark serverless e uno scheduler serverless integrato.

Terminologia

Attività
Un'attività Dataplex Universal Catalog rappresenta il lavoro che vuoi che Dataplex Universal Catalog esegua in base a una pianificazione. Incapsula il codice, i parametri e la pianificazione.
Job

Un job rappresenta una singola esecuzione di un'attività di Dataplex Universal Catalog. Ad esempio, se un'attività è pianificata per essere eseguita quotidianamente, Dataplex Universal Catalog creerà un job ogni giorno.

Per i job creati a partire dal 10 maggio 2023, il campo Trigger mostra il tipo di trigger di esecuzione del job.

Di seguito sono riportati i tipi di attivatori di esecuzione dei job:

  • RUN_REQUEST: indica che il job è stato eseguito a causa della chiamata all'API RunTask.

  • TASK_CONFIG: indica che il job è stato eseguito a causa della TriggerSpec configurazione dell'attività.

Modalità di pianificazione

Dataplex Universal Catalog supporta le seguenti modalità di pianificazione:

Esegui una volta
Utilizza questa modalità per eseguire l'attività una sola volta. Puoi scegliere di eseguirlo immediatamente o in un orario futuro prestabilito. Se esegui l'attività immediatamente, l'esecuzione potrebbe comunque richiedere fino a due minuti.
Esegui in base a una programmazione
Utilizza questa modalità per eseguire l'attività con una frequenza ripetuta. Le ripetizioni supportate sono giornaliera, settimanale, mensile o personalizzata.
Esegui on demand

Utilizza questa modalità per eseguire una attività creata in precedenza su richiesta. La modalità di esecuzione on demand è supportata solo dall'API RunTask. Quando il job viene eseguito on demand, Dataplex Universal Catalog utilizza i parametri esistenti per creare un job. Puoi specificare gli argomenti ExecutionSpec e le etichette per eseguire il job.

Prima di iniziare

  1. Abilitare l'API Dataproc.

    Abilita l'API Dataproc

  2. Abilita l'accesso privato Google per la tua rete e la tua subnet. Abilita l'accesso privato Google sulla rete che utilizzi con le attività Dataplex Universal Catalog. Se non specifichi una rete o una subnet durante la creazione dell'attività Dataplex Universal Catalog, Dataplex Universal Catalog utilizza la subnet predefinita e devi attivare l'accesso privato Google per la subnet predefinita.

  3. Crea un service account. Per pianificare qualsiasi attività Dataplex Universal Catalog è necessario un account di servizio. Il account di servizio deve appartenere al progetto in cui esegui le attività. Il account di servizio deve disporre delle seguenti autorizzazioni:

    • Accesso ai dati BigQuery e Cloud Storage in fase di elaborazione.

    • Autorizzazione Ruolo worker Dataproc sul progetto in cui esegui l'attività.

    • Se l'attività deve leggere o aggiornare l'istanza Dataproc Metastore collegata al lake, il account di servizio deve disporre del ruolo Visualizzatore o Editor Dataproc Metastore. Questo ruolo deve essere concesso nel progetto in cui è configurato il lake Dataplex Universal Catalog.

    • Se l'attività è un job Spark SQL, devi concedere all'account di servizio il ruolo Sviluppatore catalogo universale Dataplex. Questo ruolo deve essere concesso nel progetto in cui è configurato il lake Dataplex Universal Catalog.

    • Se l'attività è un job Spark SQL, devi disporre delle autorizzazioni di amministratore Cloud Storage sul bucket in cui vengono scritti i risultati.

    • Per pianificare ed eseguire Spark SQL e attività Spark personalizzate, devi disporre dei ruoli IAM Lettore metadati del catalogo universale Dataplex (roles/dataplex.metadataReader), Visualizzatore del catalogo universale Dataplex (roles/dataplex.viewer) e Utente metadati di Dataproc Metastore (roles/metastore.metadataUser) nel tuo account di servizio.

  4. Concedi all'utente che invia il job il ruolo Utente account di servizio (roles/iam.serviceAccountUser) nel service account. Per le istruzioni, vedi Gestire l'accesso agli account di servizio.

  5. Concedi all'account di servizio del lake Dataplex Universal Catalog le autorizzazioni per utilizzare l'account di servizio. Puoi trovare l'account di servizio del lake Dataplex Universal Catalog nella pagina Dettagli lake della consoleGoogle Cloud .

  6. Se il progetto contenente il lake Dataplex Universal Catalog è diverso dal progetto in cui deve essere eseguita l'attività, concedi aaccount di serviziont del lake Dataplex Universal Catalog il ruolo Editor Dataproc nel progetto in cui esegui l'attività.

  7. Inserisci gli artefatti di codice richiesti (file JAR, Python o script SQL) o i file archiviati (.jar, .tar, .tar.gz, .tgz, .zip) in un percorso Cloud Storage.

  8. Assicurati che il account di servizio disponga dell'autorizzazione storage.objects.get necessaria per il bucket Cloud Storage che archivia questi artefatti di codice.

Pianifica un'attività Spark (Java o Python)

Console

  1. Nella console Google Cloud , vai alla pagina Processo di Dataplex Universal Catalog.

    Vai a Elaborazione

  2. Fai clic su Crea attività.

  3. In Crea attività Spark personalizzata, fai clic su Crea attività.

  4. Scegli un lake Dataplex Universal Catalog.

  5. Fornisci un nome per l'attività.

  6. Crea un ID per l'attività.

  7. Nella sezione Configurazione attività, per Tipo, seleziona Spark o PySpark.

  8. Inserisci gli argomenti pertinenti.

  9. Nel campo Service account, inserisci un account di servizio utente con cui può essere eseguita l'attività Spark personalizzata.

  10. Fai clic su Continua.

  11. (Facoltativo) Imposta pianificazione: seleziona Esegui una volta o Ripeti. Compila i campi obbligatori.

  12. Fai clic su Continua.

  13. (Facoltativo) Personalizza le risorse e Aggiungi impostazioni aggiuntive.

  14. Fai clic su Crea.

gcloud

Puoi pianificare un'attività Spark (Java / Python) utilizzando il comando gcloud CLI. La tabella seguente elenca i parametri obbligatori e facoltativi da utilizzare:

Parametro Descrizione
--lake L'ID lake per la risorsa lake del servizio Dataplex Universal Catalog.
--location La posizione del servizio Dataplex Universal Catalog.
--spark-main-class La classe principale di conducente. Il file jar che contiene la classe deve trovarsi nel CLASSPATH predefinito.
--spark-main-jar-file-uri L'URI Cloud Storage del file jar che contiene la classe principale.
--spark-archive-uris (Facoltativo) URI Cloud Storage degli archivi da estrarre nella directory di lavoro di ciascun esecutore. Tipi di file supportati: .jar, .tar, .tar.gz, .tgz e .zip.
--spark-file-uris (Facoltativo) URI Cloud Storage dei file da inserire nella directory di lavoro di ciascun esecutore.
--batch-executors-count (Facoltativo) Il numero totale di esecutori job. Il valore predefinito è 2.
--batch-max-executors-count (Facoltativo) Il numero massimo di esecutori configurabili. Il valore predefinito è 1000. Se batch-max-executors-count è maggiore di batch-executors-count, Dataplex Universal Catalog attiva lo scalabilità automatica.
--container-image-java-jars (Facoltativo) Un elenco di JAR Java da aggiungere al classpath. L'input valido include gli URI Cloud Storage ai file binari Jar.
Ad esempio, gs://bucket-name/my/path/to/file.jar.
--container-image-properties (Facoltativo) Chiavi delle proprietà, specificate in formato prefix:property.
Ad esempio, core:hadoop.tmp.dir.
Per saperne di più, consulta Proprietà del cluster.
--vpc-network-tags (Facoltativo) Un elenco di tag di rete da applicare al job.
--vpc-network-name (Facoltativo) La rete Virtual Private Cloud in cui viene eseguito il job. Per impostazione predefinita, Dataplex Universal Catalog utilizza la rete VPC denominata Default all'interno del progetto.
Devi utilizzare solo una delle seguenti proprietà: --vpc-network-name o --vpc-sub-network-name.
--vpc-sub-network-name (Facoltativo) La subnet VPC in cui viene eseguito il job.
Devi utilizzare solo una delle proprietà --vpc-sub-network-name o --vpc-network-name.
--trigger-type Tipo di attivatore dell'attività specificata dall'utente. I valori devono essere uno dei seguenti:
ON_DEMAND: l'attività viene eseguita una volta poco dopo la creazione.
RECURRING: l'attività viene eseguita periodicamente in base a una pianificazione.
--trigger-start-time (Facoltativo) L'ora della prima esecuzione dell'attività. Il formato è `{year}-{month}-{day}T{hour}:{min}:{sec}Z`, dove il fuso orario è UTC. Ad esempio, "2017-01-15T01:30:00Z" codifica le ore 01:30 UTC del 15 gennaio 2017. Se questo valore non è specificato, l'attività verrà eseguita dopo l'invio se il tipo di trigger è ON_DEMAND oppure in base alla pianificazione specificata se il tipo di trigger è RECURRING.
--trigger-disabled (Facoltativo) Impedisce l'esecuzione dell'attività. Questo parametro non annulla le attività già in esecuzione, ma disattiva temporaneamente le attività RECURRING.
--trigger-max-retires (Facoltativo) Il numero di tentativi prima di interrompere l'operazione. Imposta il valore su zero per non tentare mai di ripetere un'attività non riuscita.
--trigger-schedule Pianificazione cron per l'esecuzione periodica delle attività.
--description (Facoltativo) Descrizione dell'attività.
--display-name (Facoltativo) Nome visualizzato dell'attività.
--labels (Facoltativo) Elenco di coppie di etichette KEY=VALUE da aggiungere.
--execution-args (Facoltativo) Gli argomenti da passare all'attività. Gli argomenti possono essere un mix di coppie chiave-valore. Puoi trasferire un elenco di coppie chiave-valore separate da virgole come argomenti di esecuzione. Per passare argomenti posizionali, imposta la chiave su TASK_ARGS e il valore su una stringa separata da virgole di tutti gli argomenti posizionali. Per utilizzare un delimitatore diverso dalla virgola, consulta la sezione relativa all'escape.
Se key-value e gli argomenti posizionali vengono passati insieme, TASK_ARGS verrà passato come ultimo argomento.
--execution-service-account Service account da utilizzare per eseguire un'attività.
--max-job-execution-lifetime (Facoltativo) La durata massima prima della scadenza dell'esecuzione del job.
--container-image (Facoltativo) Immagine container personalizzata per l'ambiente di runtime del job. Se non specificata, verrà utilizzata un'immagine container predefinita.
--kms-key (Facoltativo) La chiave Cloud KMS da utilizzare per la crittografia, nel formato:
projects/{project_number}/locations/{location_id}/keyRings/{key-ring-name}/cryptoKeys/{key-name}

Esempio Java:

glcoud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --trigger-type=ON_DEMAND spark-main-jar-file-uri=<gcs location to java file> --execution-service-account=<service-account-email> --trigger-start-time=<timestamp after which job starts ex. 2099-01-01T00:00:00Z> --labels=key1=value1,key2=value3,key3=value3 --execution-args=arg1=value1,arg2=value3,arg3=value3 <task-id>

Esempio PySpark:

gcloud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --trigger-type=RECURRING --trigger-schedule=<Cron schedule https://en.wikipedia.org/wiki/Cron> --spark-python-script-file=<gcs location to python script> --execution-service-account=<service-account-email> --execution-args=^::^arg1=value1::arg2=value2::TASK_ARGS="pos-arg1, pos-arg2" <task-id>

REST

Per creare un'attività, utilizza Explorer API.

Pianifica un'attività Spark SQL

gcloud

Per pianificare un'attività Spark SQL, esegui lo stesso comando gcloud CLI descritto in Pianificare un'attività Spark (Java o Python), con i seguenti parametri aggiuntivi:

Parametro Descrizione
--spark-sql-script Il testo della query SQL. È obbligatorio specificare spark-sql-script o spark-sql-script-file.
--spark-sql-script-file Un riferimento a un file di query. Questo valore può essere l'URI Cloud Storage del file di query o il percorso dei contenuti dello script SQL. È obbligatorio specificare spark-sql-script o spark-sql-script-file.
--execution-args Per le attività Spark SQL, i seguenti argomenti sono obbligatori e devono essere trasmessi come argomenti posizionali:
--output_location, <GCS uri of the output directory>
--output_format, <output file format>.
I formati supportati sono file CSV, file JSON, Parquet e ORC.
gcloud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --execution-service-account=<service-account-email> --trigger-type=ON_DEMAND --spark-sql-script=<sql-script> --execution-args=^::^TASK_ARGS="--output_location, <gcs folder location>, --output_format, json" <sql-task-id>

REST

Per creare un'attività, utilizza Explorer API.

Monitorare l'attività

Console

  1. Nella console Google Cloud , vai alla pagina Processo di Dataplex Universal Catalog.

    Vai a Elaborazione

  2. Nella scheda Attività, è presente un elenco di attività, filtrate per tipi di modelli di attività.

  3. Nella colonna Nome, fai clic su qualsiasi attività che vuoi visualizzare.

  4. Fai clic sull'ID job dell'attività che vuoi visualizzare.

    Nella consoleGoogle Cloud si apre la pagina Dataproc, che consente di visualizzare i dettagli di monitoraggio e output.

gcloud

La tabella seguente elenca i comandi gcloud CLI per monitorare le tue attività.

Azione Comando gcloud CLI
Elenco delle attività gcloud dataplex tasks list --project=<project-name> --location=<location> --lake=<lake-id>
Visualizzare i dettagli dell'attività gcloud dataplex tasks describe --project=<project-name> --location=<location> --lake=<lake-id> <task-id>
Elenco dei job di un'attività gcloud dataplex tasks jobs list --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id>
Visualizzazione dei dettagli del job gcloud dataplex tasks jobs describe --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>

Dataplex Universal Catalog esegue i job su Dataproc Serverless (batch). Per visualizzare i log di esecuzione di un job Dataplex Universal Catalog, segui questi passaggi:

  1. Recupera l'ID job Dataproc Serverless (batch). Esegui questo comando:

    gcloud dataplex tasks jobs describe --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>
    
  2. Visualizza i log. Esegui questo comando utilizzando l'ID job ottenuto dall'esecuzione del comando precedente:

    gcloud beta dataproc batches wait --project=<project-name> --region=<location> <job-id>
    

REST

Per get o list un task o un job, utilizza Explorer API.

Gestire la programmazione

Nella console Google Cloud , all'interno di Dataplex Universal Catalog, puoi modificare la pianificazione di un'attività, eliminarla o annullare un job in corso. La tabella seguente elenca i comandi gcloud CLI per queste azioni.

Azione Comando gcloud CLI
Modificare la pianificazione dell'attività gcloud dataplex tasks update --project=<project-name> --location=<location> --lake=<lake-id> --trigger-schedule=<updated-schedule> <task-id>
Eliminare un'attività gcloud dataplex tasks delete --project=<project-name> --location=<location> --lake=<lake-id> <task-id>
Annullare un job gcloud dataplex tasks jobs cancel --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>

Passaggi successivi