Importare i metadati da dbt Core

Per gli ingegneri dei dati, gli ingegneri dell'analisi e i responsabili della gestione dei dati, la centralizzazione dei metadati è fondamentale per la scoperta e la governance dei dati aziendali. Quando i team utilizzano dbt per la trasformazione dei dati, vengono generati metadati operativi, semantici e di tracciabilità preziosi, che spesso rimangono isolati all'interno dell'ecosistema dbt.

Per integrare queste informazioni nel catalogo centralizzato, puoi importare i metadati da dbt Core, dbt Cloud e MetricFlow in Knowledge Catalog (precedentemente Dataplex Universal Catalog).

Poiché dbt Core funziona come motore di trasformazione anziché come sistema di archiviazione come Oracle o PostgreSQL, l'importazione dei relativi metadati consente diversi casi d'uso. Importi i metadati Oracle o PostgreSQL per rispondere alla domanda "Quali dati non elaborati abbiamo?" e importi i metadati dbt Core per rispondere alla domanda "Come vengono trasformati i nostri dati, sono affidabili e cosa significano per l'azienda?"

Questo documento descrive come importare i metadati utilizzando il comando Google Cloud CLI e i file degli artefatti dbt.

Quando esegui l'integrazione di dbt, acquisisci i seguenti metadati:

• Metadati tecnici: scopri i dati aziendali esplorando le risorse chiave (origini, seed, modelli) e le loro proprietà tecniche (nomi delle colonne, tipi di dati, conteggi delle righe).
• Metadati aziendali e semantici: forniscono il contesto per gli strumenti di BI e gli agenti AI esplorando le definizioni e la logica aziendali basate su dbt MetricFlow, come modelli semantici, metriche e query salvate.
• Metadati operativi e di qualità dei dati: monitora l'integrità della pipeline e risolvi i problemi relativi ai dati esplorando i metadati di esecuzione, come tempistiche, stato di riuscita o errore, aggiornamento dei dati e risultati dei test.
• Metadati di tracciabilità e relazione: consentono l'analisi dell'impatto a valle e il tracciamento della causa principale esplorando i grafi di trasformazione (DAG) e le dipendenze tra le risorse dbt, la tracciabilità fisica che monitora e collega i blocchi di trasformazione fisica, le chiavi di unione e le unioni dinamiche e le relazioni padre-figlio.
• Metadati di consumo: risolvi i problemi relativi al modo in cui le applicazioni downstream consumano i dati trasformati esplorando i metadati acquisiti nelle esposizioni che mappano il modo in cui i dati vengono utilizzati al di fuori di dbt.

Limitazioni

• Supporta dbt Core v1 (convalida rispetto alle versioni 1.11 e 1.12), dbt Core v2 e dbt Fusion.
• gcloud CLI versione 586.0.0 e successive supportano l'integrazione di dbt e BigQuery. Per installare o aggiornare la CLI, vedi Installa Google Cloud CLI.
• Non esiste una connessione diretta a dbt Cloud. Per importare i metadati da un job dbt Cloud, recupera prima gli artefatti del job. Consulta Importare i metadati dalle esecuzioni di dbt Cloud.
• Gli schemi molto grandi o nidificati in profondità vengono troncati: un singolo aspetto non può superare il limite di dimensioni per aspetto, quindi gli schemi nidificati in profondità potrebbero perdere i campi finali.
• --aspects-only può aggiungere e aggiornare i metadati, ma non può rimuoverli. L'eliminazione di una risorsa dbt richiede un'esecuzione completa.
• Questa integrazione supporta solo gli eventi di derivazione dbt sulle risorse BigQuery nell'API e nel grafico Data Lineage. Le voci dbt (origini, seed, modelli) per origini esterne di terze parti non vengono acquisite nella derivazione dei dati.
  • Per importare tutti gli eventi di derivazione dbt nell'API Data Lineage, utilizza l'integrazione dbt OpenLineage. Poi, integra OpenLineage con Knowledge Catalog per importare e visualizzare la tracciabilità dei dati da dbt.

Prima di iniziare

Prima di poter importare i metadati da dbt Core e MetricFlow, completa le seguenti attività:

1. Concedi i ruoli e le autorizzazioni richiesti.
2. Abilita l'API Knowledge Catalog.
3. Soddisfare i prerequisiti di dbt.
4. Crea il gruppo di voci di destinazione se non esiste già.
5. Comprendi i ruoli Cloud Storage.

Ruoli e autorizzazioni IAM

Per creare e gestire un job del connettore Knowledge Catalog, devi disporre di ruoli Identity and Access Management (IAM) che concedono autorizzazioni per Knowledge Catalog e Cloud Storage.

Per ottenere le autorizzazioni necessarie per configurare un connettore dbt, chiedi all'amministratore di concederti i seguenti ruoli IAM:

• Per creare e gestire gruppi di voci e link alle voci: Dataplex Catalog Admin (roles/dataplex.catalogAdmin), Dataplex Catalog Editor (roles/dataplex.catalogEditor) o Dataplex Entry Group Owner (roles/dataplex.entryGroupOwner) nel progetto.

• Per eseguire il comando dbt gcloud e creare job di importazione dei metadati: segui il principio del privilegio minimo e assegna i seguenti ruoli:

  • Dataplex Metadata Job Owner (roles/dataplex.metadataJobOwner) nel progetto.
  • Dataplex Entry Group Importer (roles/dataplex.entryGroupImporter) sul gruppo di voci di destinazione o sul progetto. Se importi anche i link alle voci, concedi Dataplex Entry Group Owner (roles/dataplex.entryGroupOwner) al progetto. Concedi anche Proprietario voce Dataplex (roles/dataplex.entryOwner) in ogni progetto che contiene le tabelle BigQuery in cui scrivono i tuoi modelli dbt. Per i ruoli personalizzati, le autorizzazioni dei link di accesso sono dataplex.entryGroups.useReferenceEntryLink, dataplex.entryGroups.useSchemaJoinEntryLink e dataplex.entryLinks.reference.

  In alternativa, puoi concedere il ruolo Dataplex Catalog Admin (roles/dataplex.catalogAdmin) e il ruolo Dataplex Metadata Job Owner (roles/dataplex.metadataJobOwner) nel progetto.

• Per caricare i metadati trasformati nel bucket di staging di output (--storage-uri): Creatore oggetti Storage (roles/storage.objectCreator) o Amministratore oggetti Storage (roles/storage.objectAdmin) nel bucket di staging.

• Per leggere gli artefatti dbt da un bucket Cloud Storage di input (--artifacts-path, se utilizzi Cloud Storage): Visualizzatore oggetti Storage (roles/storage.objectViewer) o Amministratore oggetti Storage (roles/storage.objectAdmin) sul bucket degli artefatti di input. Se disponi del ruolo Storage Object Admin, il ruolo Storage Object Viewer non è necessario.

• Per visualizzare i metadati dbt: Dataplex Catalog Viewer (roles/dataplex.catalogViewer) sul progetto.

• Per visualizzare i log in Cloud Logging: Visualizzatore log (roles/logging.viewer) sul progetto.

Se disponi delle autorizzazioni necessarie per gestire l'accesso IAM nel tuo progetto, puoi concedere questi ruoli al tuo account utente eseguendo i seguenti comandi gcloud:

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="user:USER_EMAIL" \
    --role="roles/dataplex.metadataJobOwner"

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="user:USER_EMAIL" \
    --role="roles/dataplex.entryGroupOwner"

gcloud storage buckets add-iam-policy-binding gs://STAGING_BUCKET \
    --member="user:USER_EMAIL" \
    --role="roles/storage.objectCreator"

Se esegui l'importazione utilizzando un account di servizio, ad esempio in una pipeline CI/CD automatizzata, puoi concedere questi ruoli al account di servizio eseguendo i seguenti comandi gcloud:

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_EMAIL" \
    --role="roles/dataplex.metadataJobOwner"

gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:SERVICE_ACCOUNT_EMAIL" \
    --role="roles/dataplex.entryGroupOwner"

gcloud storage buckets add-iam-policy-binding gs://STAGING_BUCKET \
    --member="serviceAccount:SERVICE_ACCOUNT_EMAIL" \
    --role="roles/storage.objectCreator"

Inoltre, devi concedere all'agente di servizio Knowledge Catalog (service-PROJECT_NUMBER@gcp-sa-dataplex.iam.gserviceaccount.com) il ruolo Visualizzatore oggetti Storage (roles/storage.objectViewer) nel bucket Cloud Storage gestione temporanea di output (--storage-uri) in modo che il job di importazione possa leggere il file di metadati gestione temporanea:

gcloud storage buckets add-iam-policy-binding gs://STAGING_BUCKET \
    --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-dataplex.iam.gserviceaccount.com" \
    --role="roles/storage.objectViewer"

Sostituisci quanto segue:

• PROJECT_ID: il tuo ID progetto Google Cloud .
• USER_EMAIL: l'indirizzo email dell'account utente.
• SERVICE_ACCOUNT_EMAIL: l'indirizzo email del account di servizio.
• STAGING_BUCKET: il nome del bucket Cloud Storage di staging dell'output (--storage-uri).
• PROJECT_NUMBER: il tuo Google Cloud numero di progetto.

Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso.

Abilita API

Abilita l'API Knowledge Catalog.

Abilitare l'API

Prerequisiti dbt

Per importare l'intero set di metadati dbt, ti consigliamo di produrre tutti e quattro i file di artefatti JSON dbt. È richiesto solo manifest.json; gli altri arricchiscono l'importazione e la trasformazione viene eseguita senza problemi anche senza:

• manifest.json (obbligatorio): struttura principale del progetto e grafico di esecuzione. Contiene anche i modelli semantici, le metriche e le query salvate di MetricFlow.
• catalog.json: Nomi delle colonne e tipi di dati. Senza catalog.json, l'aspetto dello schema viene importato con colonne non tipizzate.
• run_results.json: Risultati del test e metadati di esecuzione.
• sources.json: Aggiornamento della fonte.

Nel terminale locale, in Cloud Shell o nell'ambiente CI/CD automatizzato in cui è installato dbt, vai alla directory principale del progetto dbt ed esegui i seguenti comandi dbt in ordine rispetto a un singolo profilo e target per generare l'insieme completo di file JSON degli artefatti dei metadati dbt:

• Per dbt Core 2.x e dbt Fusion:

  1. dbt source freshness
  2. dbt build

  3. dbt parse --write-catalog

• Per dbt Core 1.x (dove dbt parse non scrive un catalogo):

  1. dbt source freshness
  2. dbt build

  3. dbt docs generate --no-compile

Comprendere i ruoli di Cloud Storage

L'importazione dei metadati dbt prevede due diverse posizioni Cloud Storage che hanno scopi diversi e non devono essere confuse:

• Input (artefatti di origine dbt): dove si trovano i file JSON dbt generati. Può essere un percorso di directory locale sulla tua macchina o sul runner CI (ad esempio ./target/ o .) o un prefisso URI del bucket Cloud Storage di input (ad esempio gs://my-dbt-artifacts-bucket/target/). Fornisci questo percorso utilizzando il flag --artifacts-path. Il comando gcloud legge questi file di input durante la preparazione del job. Il chiamante che esegue il comando gcloud deve disporre dell'accesso in lettura (roles/storage.objectViewer o roles/storage.objectAdmin) se utilizza Cloud Storage. Il service agent Knowledge Catalog non ha bisogno dell'accesso al bucket degli artefatti di input.
• Output (bucket gestione temporanea di importazione di Knowledge Catalog): un prefisso URI del bucket Cloud Storage (ad esempio gs://my-staging-bucket/dbt-imports/) in cui il comando gcloud carica il file di importazione dei metadati trasformato (dbt_metadata.jsonl) e da cui il job di importazione di Knowledge Catalog legge durante l'importazione. Fornisci questo URI utilizzando il flag --storage-uri. Il chiamante che esegue il comando gcloud deve disporre dell'accesso in scrittura (roles/storage.objectCreator o roles/storage.objectAdmin) per caricare il file, mentre l'agente di servizio Knowledge Catalog deve disporre dell'accesso in lettura (roles/storage.objectViewer) per importarlo.

Importare i metadati dalle esecuzioni di dbt Cloud

Knowledge Catalog non si connette direttamente a dbt Cloud. Poiché un job dbt Cloud genera gli stessi file di artefatti di dbt Core, puoi importare i metadati da dbt Cloud recuperando questi file di artefatti in una directory locale o in un bucket Cloud Storage di input ed eseguendo il comando gcloud.

Prima di recuperare gli artefatti, configura il job dbt Cloud per generare il set completo di artefatti. Puoi quindi recuperare i file degli artefatti dall'esecuzione di un job dbt Cloud utilizzando uno dei seguenti metodi:

• Scarica gli artefatti dalla console dbt Cloud: scarica manualmente i file degli artefatti dalla pagina dei dettagli di esecuzione del job nell'interfaccia utente dbt Cloud per importazioni una tantum o test iniziali.
• Scarica gli artefatti utilizzando la CLI della piattaforma dbt: esegui i comandi dbt su dbt Cloud dal tuo terminale locale per salvare automaticamente gli artefatti generati nella directory del progetto locale durante lo sviluppo.
• Scarica gli artefatti utilizzando l'API amministrativa dbt: recupera gli artefatti in modo programmatico dalle esecuzioni completate tramite HTTP per pipeline automatizzate e pianificate.

Configura il job dbt Cloud

Nella console dbt Google Cloud , configura le impostazioni del job per generare l'insieme completo di artefatti di metadati:

1. Nella sezione Impostazioni di esecuzione, seleziona Esegui aggiornamento origine. dbt Cloud esegue dbt source freshness prima dei comandi del job per generare sources.json.
2. Nella sezione Comandi, aggiungi dbt build.
3. Aggiungi un comando per generare catalog.json in base al canale di rilascio:
   • Per le tracce di rilascio di dbt Core 2.x e dbt Fusion: aggiungi dbt parse --write-catalog come comando del job.
   • Per le tracce di rilascio di dbt Core 1.x: aggiungi dbt docs generate --no-compile come comando del job anziché selezionare l'opzione Genera documentazione durante l'esecuzione. La casella di controllo Genera documenti durante l'esecuzione esegue dbt docs generate senza --no-compile, il che sovrascrive i risultati del test di dbt build, come descritto in Prerequisiti di dbt. Tieni presente che se un passaggio di comando non va a buon fine, anche il job non va a buon fine, mentre il passaggio della casella di controllo non causa l'errore del job.

Se dbt build non va a buon fine, ad esempio perché un test non riesce, dbt Cloud ignora i comandi successivi e l'esecuzione non ha catalog.json. Per produrne sempre uno, aggiungi il comando del catalogo prima di dbt build. Il catalogo descrive quindi le tabelle come erano prima della build.

Per ulteriori informazioni, consulta Comandi dei job e Tracce di rilascio nella documentazione di dbt.

Scaricare gli artefatti dalla console dbt Google Cloud

Per scaricare manualmente gli artefatti da un'esecuzione completata nella console dbt Google Cloud :

1. Nella console dbt Google Cloud , apri l'esecuzione del job completata.
2. Vai alla scheda Artefatti per visualizzare i file degli artefatti generati.
3. Scarica manifest.json, catalog.json, run_results.json e sources.json in una directory locale.
4. Nel terminale locale o in Cloud Shell, esegui il comando di importazione gcloud descritto in Configura la connettività dbt e imposta --artifacts-path sulla directory contenente i file scaricati.

Per saperne di più, consulta Esecuzione della visibilità nella documentazione di dbt.

Scaricare gli artefatti utilizzando la CLI della piattaforma dbt

La CLI della piattaforma dbt (in precedenza la CLI dbt Cloud) esegue i comandi dbt sulla piattaforma dbt Cloud dal terminale locale e scarica automaticamente gli artefatti generati nella directory target/ del progetto dbt locale.

1. Nel terminale locale, vai alla directory principale del progetto dbt ed esegui i tre comandi elencati nei prerequisiti di dbt.
2. Esegui il comando di importazione gcloud descritto in Configura la connettività dbt e imposta --artifacts-path sulla radice del progetto o sulla directory target/.

La CLI viene eseguita all'interno del tuo ambiente di sviluppo utilizzando le credenziali del tuo data warehouse personale, quindi i metadati generati riflettono lo schema di sviluppo anziché le tabelle di produzione create da un job programmato. Utilizza l'interfaccia a riga di comando per i flussi di lavoro di test o sviluppo e un job di deployment per le importazioni di produzione pianificate.

Per saperne di più, consulta Installare la CLI della piattaforma dbt nella documentazione di dbt.

Scaricare gli artefatti utilizzando l'API amministrativa dbt

Puoi utilizzare l'API dbt Administrative per recuperare in modo programmatico gli artefatti da qualsiasi esecuzione del job completata. L'endpoint List Run Artifacts restituisce i percorsi dei file generati da un'esecuzione, mentre l'endpoint Retrieve Run Artifact scarica un file di artefatto specifico dal seguente URL:

https://ACCESS_URL/api/v2/accounts/ACCOUNT_ID/runs/RUN_ID/artifacts/FILE

ACCESS_URL dipende dalla regione che ospita il tuo account dbt Cloud. Autentica le richieste utilizzando un token di servizio dbt Cloud. Per saperne di più, consulta le seguenti pagine della documentazione di dbt:

• API amministrativa dbt
• Accesso, regioni e indirizzi IP
• Token del service account

Dal terminale locale, da Cloud Shell o dall'ambiente di flusso di lavoro automatizzato, scarica manifest.json, catalog.json, run_results.json e sources.json in una directory locale o in un bucket Cloud Storage, quindi esegui il comando gcloud descritto in Configura la connettività dbt rispetto a questo percorso.

Per impostazione predefinita, l'endpoint dell'artefatto restituisce gli artefatti dell'ultimo passaggio dell'esecuzione, a meno che tu non specifichi il parametro di query step. Quando configuri il job come descritto in Configura il job dbt Cloud, il passaggio finale è dbt parse --write-catalog o dbt docs generate --no-compile, che scrive solo catalog.json e lascia intatti gli altri tre artefatti nel passaggio predefinito.

Recuperare l'ID esecuzione

Per scaricare gli artefatti di un'esecuzione specifica, devi disporre del relativo ID esecuzione. Puoi copiare l'ID esecuzione dall'URL di esecuzione nella console dbt Google Cloud oppure eseguire una query sull'API dal terminale o dallo script del workflow per l'ultima esecuzione riuscita di un job:

GET https://ACCESS_URL/api/v2/accounts/ACCOUNT_ID/runs/?job_definition_id=JOB_ID&status=10&order_by=-finished_at&limit=1

Nei parametri di ricerca, status=10 filtra le esecuzioni completate con stato Success. Puoi eseguire il polling di questo endpoint in base a una pianificazione per identificare l'ultima esecuzione riuscita, scaricare i relativi artefatti ed eseguire il comando di importazione gcloud.

Attivare l'importazione utilizzando un webhook

Anziché eseguire il polling dell'API, puoi configurare un webhook dbt Cloud per attivare un'importazione automatica dei metadati al termine di un'esecuzione del job. Il webhook invia un payload a un endpoint HTTP che fornisci:

1. Nella console dbt Google Cloud , vai a Impostazioni account > Webhook e fai clic su Crea webhook (o Crea nuovo webhook). Configura l'abbonamento al webhook:
   • Eventi: seleziona Esecuzione completata (job.run.completed), che viene attivato solo al termine dell'esecuzione e quando gli artefatti sono disponibili per il download.
   • Job: seleziona i job di deployment dbt Cloud che vuoi monitorare.
   • Endpoint: inserisci l'URL HTTPS di un servizio che esegui (ad esempio un servizio Cloud Run o una funzione Cloud Run).
2. Salva il token secret webhook visualizzato da dbt Cloud. Il tuo servizio utilizza questo secret per verificare l'intestazione Authorization, che contiene una firma HMAC-SHA256 del corpo della richiesta.
3. Nel tuo servizio, leggi data.runId dal payload JSON, scarica gli artefatti dell'esecuzione utilizzando l'API amministrativa come descritto in precedenza ed esegui il comando gcloud alpha dataplex dbt metadata-jobs create.

Quando implementi il gestore webhook, considera quanto segue:

• dbt Cloud attende al massimo 10 secondi per una risposta. Poiché l'importazione dei metadati richiede diversi minuti, restituisci prima una risposta HTTP ed esegui l'importazione in background (ad esempio, come job Cloud Run o con il flag --async).
• job.run.completed viene attivato anche per le esecuzioni non riuscite, quindi le esecuzioni con test non riusciti vengono comunque importate. Non abbonarti a job.run.errored, perché può essere attivato prima che gli artefatti dell'esecuzione siano disponibili.

Per ulteriori informazioni sui payload webhook e sulla verifica della firma, consulta Webhook per i job nella documentazione di dbt.

Configura la connettività dbt

Per stabilire la connettività dbt, devi prima eseguire i comandi dbt appropriati per generare gli artefatti dei metadati. Una volta archiviati e accessibili i file JSON, il processo di importazione esegue le seguenti azioni:

1. Leggi artefatti di input: leggi gli artefatti JSON generati da dbt Core e MetricFlow dalla posizione di input (directory locale o URI Cloud Storage specificato in --artifacts-path).
2. Trasforma metadati: trasforma i contenuti nel formato di importazione dei metadati di Knowledge Catalog (dbt_metadata.jsonl).
3. Carica in staging: carica il file di importazione dei metadati trasformati nella posizione di staging di output di Cloud Storage specificata in --storage-uri.
4. Attiva job di importazione: attiva un job di importazione dei metadati di Knowledge Catalog che indica al service agent di Knowledge Catalog di leggere e importare i metadati in staging da --storage-uri nelle risorse Knowledge Catalog.

Dopo aver creato il job, Knowledge Catalog pianifica la prima esecuzione in base alla configurazione oppure puoi avviarla manualmente.

Esegui di nuovo l'importazione

Dopo la prima importazione, la maggior parte delle esecuzioni deve solo aggiornare i metadati delle risorse già esistenti. Utilizza --aspects-only per queste corse. Aggiorna solo ciò che l'esecuzione di dbt ha osservato e lascia tutto il resto nel gruppo di voci invariato, quindi è sicuro eseguire ripetutamente, in qualsiasi pianificazione e da più di un job.

Esegui un'importazione completa (ometti --aspects-only) quando il set di voci cambia:

• La prima importazione in un gruppo di voci.
• Una risorsa dbt viene aggiunta, rinominata o eliminata.
• Modifica del nome visualizzato, della descrizione o delle etichette di una voce.
• La gerarchia delle voci cambia.
• Le dipendenze dbt cambiano, ad esempio quando viene aggiunta o rimossa una chiamata di ref(), source(), test o macro. --aspects-only non creano né aggiornano i link delle voci.
• Modifichi --include-entry-links o --skip-bigquery-link.

Un'esecuzione completa riscrive gli aspetti richiesti di ogni voce dagli artefatti sul disco, quindi eseguila da un insieme di artefatti il più completo possibile che la pipeline possa produrre.

Esegui --aspects-only per aggiornamenti di routine:

• Dopo l'esecuzione del comando dbt della pipeline: dbt build, dbt test, dbt source freshness o una ricompilazione limitata a --select.
• Una colonna viene aggiunta, rimossa, ridigitata o viene modificata la descrizione.
• L'SQL del modello è stato modificato e l'esecuzione ha scritto anche catalog.json.
• Nuovi risultati del test o aggiornamento della fonte.

--aspects-only può aggiungere e aggiornare i metadati, ma non può rimuoverli.

Cercare e visualizzare i metadati di dbt

Per elencare i link alle voci di una voce dbt, chiama il metodo projects.locations:lookupEntryLinks. Ad esempio, per recuperare la tabella BigQuery in cui viene materializzato un modello dbt:

curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://dataplex.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION:lookupEntryLinks?entry=ENTRY_NAME&entryMode=SOURCE&entryLinkTypes=projects/dataplex-types/locations/global/entryLinkTypes/reference"

ENTRY_NAME è il nome completo della risorsa della voce dbt. I risultati sono paginati, con un massimo di 10 link per pagina.

Per scoprire di più sulla ricerca delle risorse, consulta Cercare risorse in Knowledge Catalog. Per saperne di più su espressioni di query e filtri, consulta Sintassi di ricerca per Knowledge Catalog.

Passaggi successivi

• Scopri come gestire i job del connettore.