Multi-Source DataHub Catalog Sync with ForEach

One database in the catalog is a demo; a catalog earns trust when it covers everything. The naive way to get there is one flow per source, which means every improvement gets copy-pasted N times. This blueprint keeps a single flow: io.kestra.plugin.core.flow.ForEach iterates over a list of JSON source definitions, and each iteration renders one io.kestra.plugin.datahub.Ingestion recipe from its JSON with fromJson, a Postgres CRM database and a MySQL warehouse out of the box. Adding a third source is adding one line to the values list. Discord gets a single summary when the whole sync completes.

How it works

1. sync_each_source (io.kestra.plugin.core.flow.ForEach) iterates over the values list. Each value is a JSON string describing one source, its DataHub source type, connection details, and the name of the Kestra secret holding its password.
2. Inside the loop, ingest_source builds the recipe dynamically. {{ fromJson(taskrun.value).type }} selects the source type and the config fields fill in from the same JSON, which works because postgres and mysql sources share the same core config keys.
3. Passwords never appear in the values list. Each definition carries only a secret name, and {{ secret(fromJson(taskrun.value).password_secret) }} resolves it at runtime, so the flow stays safe to commit with real hostnames in it.
4. Every iteration pushes to the same datahub-rest sink, resolved from the shared GMS secrets, so all sources land in one catalog.
5. notify sends one Discord summary after the fan-out completes, the errors block sends one alert naming the execution when any source fails, and a disabled-by-default Schedule trigger runs the whole sync nightly at 04:30.

What you get

• One flow that catalogs any number of databases, with per-source additions as one-line changes.
• Dynamic recipes rendered per iteration with fromJson, no recipe files to maintain per source.
• Per-source secret indirection, so each database keeps its own credential and rotation schedule.
• One summary message per sync instead of channel noise per database.

Who it's for

• Data platform teams cataloging a fleet of operational databases, not a single warehouse.
• Teams consolidating per-source cron jobs for datahub ingest into one observable pipeline.
• Governance owners who want catalog coverage to grow by editing a list, not by writing flows.

Why orchestrate this with Kestra

Fan-out is where orchestration pays for itself. Kestra runs the iterations as separate task runs with separate logs, so one unreachable database fails visibly without hiding the others, and the execution view shows per-source status at a glance. The recipe stays the standard DataHub YAML; only the values that differ per source are parameterized. When the API team hands you the next database, coverage is a pull request that adds one JSON line.

Prerequisites

• A running DataHub instance with its GMS endpoint reachable from Kestra.
• A read-only metadata user on each source database.
• Docker available to the Kestra worker for the acryldata/datahub-ingestion container image.
• A Discord incoming webhook.

Secrets

• DATAHUB_GMS_URL: URL of the DataHub GMS server.
• DATAHUB_TOKEN: DataHub personal access token with permission to ingest metadata.
• CRM_DB_PASSWORD: password for the Postgres CRM reader user.
• WAREHOUSE_DB_PASSWORD: password for the MySQL warehouse reader user.
• DISCORD_WEBHOOK_URL: Discord incoming webhook URL.

Quick start

1. Add the five secrets to your Kestra namespace.
2. Edit the two JSON definitions in values to point at real databases and execute the flow.
3. Confirm both databases appear in DataHub, then set disabled: false on the nightly trigger.

How to extend

• Add sources by appending JSON lines to values; any DataHub source sharing these config keys works as-is.
• Set concurrencyLimit on the ForEach to sync large fleets a few sources at a time.
• Add schema_pattern fields to the JSON definitions and the recipe to scope noisy databases.
• Move the definitions into a JSON flow input or a KV store entry when the list should change without editing the flow.

Links

• DataHub plugin documentation
• DataHub metadata ingestion documentation