External Open-Data Sources

Public reference datasets loaded into the platform graph by Datta Extract: CVM (public companies, investment funds and sanctioning proceedings) and the public negative database — Portal da Transparência (CEIS/CNEP/CEPIM/PEP), the Central Bank and the UN Security Council consolidated list (see Public Negative Database). They feed the KYC/KYB vertical: ultimate-beneficial-owner drill-down stops, fund resolution and the negative base.

How it works

Each source is a Pipeline Designer data flow — exactly like any integration drawn by an analyst on the platform. There is no dedicated service, connector or Java class: the whole flow (download, cleanup, graph join) is composed of designer components and stored as a pipeline (origin=DESIGNER).

The way to reach it is Extract › Pacotes — the gallery that inventories every extraction pipeline in the installation. Each flow shows up as a card (with origin, schedule and last run) and clicking it opens the flow in the editor where it is editable. For the external-source flows that is Avançado mode, because their graph is not the linear chain the Assistente builds — the gallery states that reason on the card. There is no direct shortcut from the menu: without the gallery, the editor opens blank.

Execution is the platform's standard pipeline path: the pipeline-designer-service schedules by the flow's cron and runs it on the platform's own engine — the same one that runs any other pipeline designed in the UI. Previously the load was submitted to Spark; not any more.

The three CVM flows (files under scripts/pipelines/cvm/):

Common anatomy of the flows:

1. URL source (UrlSource) — downloads the CSV from CVM's open-data portal with anti-SSRF validation on every redirect and empty-file rejection.
2. Designer transforms — Formula (digits-only CNPJ, base CNPJ, load version), Filter (drops rows without a key), Rename, SelectColumns and Deduplicate. No data handling in code.
3. Graph target (KnowledgeGraphTarget) — key-based MERGE into the FEBRABAN context database plus the edges to the Federal Revenue graph. The Revenue endpoint uses Match mode: a record with no counterpart gets no edge — legitimate, never fabricated.

When the source is an API, not a file

A source that answers over an API comes in through the REST API node, with no dedicated code. The node covers the four pagination modes (by page number, by offset, by URL path and by a cursor returned in the response itself), JSON or XML responses, an API key in an encrypted field (instead of hand-written into the header) and the two-step read — the case where the query returns a link to the file rather than the records.

The graph target, in turn, covers a load fed by several sources describing the same entity: a cascade of candidate keys (the row merges by the first one with a value), an empty field that preserves the value already stored, a property that accumulates provenance in a list without duplicating, and a property written only on creation.

Every field, with an example, is in Advanced connector fields.

Flows whose target changes on every run

A flow can declare execution parameters — target context, search term, reference period, record ceiling —, each with a default value. That is what avoids having one package per combination: the same design serves every target, and one fix covers them all at once.

How the values reach the trigger, today:

In other words: to run with a value other than the default without editing the package, use the gallery's Executar agora (or the API). The form applies to that run only: the default stored in the package does not change, so what the scheduled run uses stays as it is in the package — to change that, edit the flow in Extract › Designer.

The values used are recorded in the pipeline's audit trail, so two runs of the same flow with different parameters are never confused.

Registering the flows (seed)

Automatically, by the installer

setup-datta.sh registers the flows in Phase 7B — External-source extraction packages, with no extra command. The phase talks to the api-gateway through the ingress (the in-cluster DNS name does not resolve from the installer host) and discovers the reference context database by reading GET /api/contextos — whatever the user already registered, never a guessed value (Project Guidelines §9).

If the reference context does not exist yet, the phase registers nothing, warns and moves on — installing the platform never depends on a business context existing. The same applies when the API is down or the Neo4j password is not available (a run with --skip-deploy and no .env): it becomes a warning in the final summary, never an installation failure. To skip the phase on purpose: ./setup-datta.sh --skip-seed-extract.

Manually

Run it after registering the context, in either of two ways:

# 1) Through the ingress, with an admin JWT (from outside the cluster — same as
#    the installer). The PER-ORIGIN databases are discovered from the context's
#    NEO4J dataSources (role=cvm | base-negativa | bcb); pass the variables only
#    to force different ones.
DATTA_API_URL=http://<gateway-ip>:7070 \
DATTA_JWT=<accessToken> \
NEO4J_PASSWORD=<password> \
./scripts/seed-extract-pipelines.sh

# 2) From inside the cluster, with the S2S token.
CVM_NEO4J_DB=<role cvm database> \
BASE_NEGATIVA_NEO4J_DB=<role base-negativa database> \
BCB_NEO4J_DB=<role bcb database> \
NEO4J_PASSWORD=<password> \
DATTA_INTERNAL_TOKEN=<S2S token> \
./scripts/seed-extract-pipelines.sh

The script is idempotent by name: an already-registered flow is not recreated, so later edits made in the designer are never overwritten — and running the installer twice does not duplicate a package. In S2S mode the three per-origin databases are still required with no default, because from inside the cluster there is no JWT to query the contexts with.

When the repository package changes

Silently skipping what already exists protects designer edits, but it cost the other side: a package fix in the repository never reached an already-installed platform. That is how the fix to the CVM sanctioning package stayed stuck — the seed reported "already registered" while the flow kept running the old definition, failing every night.

The seed now compares the repository design against the registered one and reports divergence without applying it:

! DIVERGENTE (não aplicado): CVM — Processos Sancionadores (dados abertos)

To apply it, when the repository version is the correct one (a source that changed format, a corrected key, a renamed column):

SEED_ATUALIZAR=true ./scripts/seed-extract-pipelines.sh

This overwrites the registered design — designer edits on those packages are lost. The pipeline keeps the previous version in its history (Extract › Designer › Versions), so you can compare beforehand and roll back afterwards.

The comparison ignores what the server administers (id, version, timestamps) and what it adds on write (config: {} on every edge). It also ignores the password, which the API returns masked: a credential change in the repository does not show up as divergence, and is applied by running with SEED_ATUALIZAR=true.

setup-datta.sh does call this script, in PHASE 7B — External-source extraction packages (see the call to scripts/seed-extract-pipelines.sh). On a fresh install the flows are registered with no extra command. The phase skips with a warning — without breaking the installation — when the reference context does not exist, the API is down, the Neo4j password is unavailable (a run with --skip-deploy and no .env) or the script is missing. To skip on purpose: ./setup-datta.sh --skip-seed-extract.

On a pre-existing cluster the seed stays manual: the phase only runs during setup-datta.sh, and an earlier installation never executed it. datta-install.sh and seed-demo-data.sh do not call the seed.

Reference context name: the seed looks the context up by CONTEXTO_REFERENCIA (default FEBRABAN), which must match datta.kyb.contexto-referencia — if they diverge, the packages end up registered against databases the KYB cross-reference never queries. The context declares one NEO4J dataSource per origin (role=cvm, base-negativa, bcb, cnpj) and the seed reads the databases from there — including CNPJ_NEO4J_DB, used again by the CNPJ packages (the Federal Revenue base stays in the cnpj database and the join with the other origins is by the cnpjBasico key, never by a cross-database edge). If the registered name differs from the configured one, the seed lists the existing contexts in the message instead of just saying it found nothing; if a role is missing, it lists what to register.

Operating through the UI

The starting point is Extract › Pacotes; search by the flow name (e.g. "CVM") or filter by origin/integration. From the card:

• Abrir (open) — loads the flow into the ETL Designer with the original nodes, positions and configuration; there you adjust transforms, the source URL or the cron, and save (version history records every change). Saving resends the complete definition, so description, cron, execution policy and metadata are not lost.
• Executar agora (run now) — triggers a one-off run straight from the gallery; per-node progress shows up in the run monitor (Ver histórico goes to the monitor already filtered by the flow). A flow declaring execution parameters first opens a form with them, filled with the package defaults: adjust what applies to this run and confirm. A flow with no parameters triggers straight away, with no form. A blank required field is stopped right there, with the explanation on the field itself; a refusal from the engine (invalid option, value outside the type) shows inside the form, without losing what you already filled in.
• Pausar / Ativar — controls scheduling without deleting the flow.
• Duplicar — creates a copy (useful for the air-gapped variant below). Caveat: the copy comes with masked credentials and they must be re-entered in the editor before running it.
• Arquivar / Excluir — takes the flow out of operation or removes it permanently (with double confirmation).
• Detalhes — side panel with origin, status, version, author, schedule and the flow drawn on a read-only canvas, to inspect the anatomy without opening the editor.
• Test — inside the designer, test mode runs on a sample and validates the flow without writing to the graph.

Air-gapped installations

Download the CSV from the source on a connected machine and stage it on the platform storage. Then, in Extract › Pacotes, duplicate the flow, open the copy in the designer and swap the source node from URL to File (FileSource) pointing at the CSV — keeping the ; delimiter and ISO-8859-1 encoding. The rest of the flow (transforms and target) is identical. Pause the original URL-based flow so it does not accumulate scheduling failures.

Scheduling

The cron is part of the flow itself (editable in the designer): companies at 04:00, funds at 04:10 and sanctioning proceedings at 04:20, America/Sao_Paulo timezone. Single retry with a 5-minute delay; failures notify through the channel configured in the pipeline's execution policy.

Permissions

The same as pipelines in general — there is no per-source permission:

Mind this when granting: today the Analista role has no PIPELINE_* permission at all — it still sees the gallery, because GET /api/pipelines/summary accepts PIPELINE_VIEW or DATTAX_EXECUTE — and Usuário Avançado lacks PIPELINE_DELETE. See RBAC.

CVM data in the graph

Destination: the Neo4j database of the FEBRABAN context (same graph as the Federal Revenue base — the join axis is the CNPJ). Labels: CompanhiaAberta, FundoInvestimento, AdministradorFiduciario, GestorRecursos, ProcessoSancionador, all carrying versaoCarga (execution date).

Points of attention:

• The sanctioning data never links to :Empresa, and that comes from the source. Since CVM's schema change (the old pas.csv went offline in 08/2026), the dataset publishes only the defendant's name — there is no CPF and no CNPJ. With no document there is no deterministic join key, so the flow loads the defendants as nodes and name matching (with score and human review) belongs to the KYB cross-check, exactly as it already works for the UN list. Creating a sanction edge by name would turn homonymy into an accusation.
• Two CSVs in the same zip. The flow has two sources — one per zip entry — and links them by NUP, the proceeding's key inside the dataset itself. That avoids numeric coincidence.
• Reload is a key-based MERGE. CVM registries are complete snapshots (not deltas); each run updates existing nodes and creates new ones. A record that leaves the source registry remains in the graph with an old versaoCarga — triage criteria can use the date to flag staleness.

Troubleshooting

Observability

Execution inherits the standard pipeline telemetry: runs and per-node status in the Extract monitor, pipeline-designer-service spans in the platform's OTel indices, and create/edit/run audit in the pipeline's own trail (designer audit tab).