From External Source to Ontology — the authoring process

This guide is the end-to-end recipe for taking raw data from an external source all the way to typed entities in the ontology, entirely through frankctl — no database access, no dashboard required.

Every command and output below is real, taken from driving the ARTE IPMA surface-observations feed to a FIWARE WeatherObserved entity type.

Start from a curated template when one exists

Before authoring a vertical from scratch, check the packaged catalog:

frankctl templates list
frankctl templates show cira
frankctl templates render cira --component cityfy-events -o cira-events.yaml
frankctl pipelines apply -f cira-events.yaml --dry-run-server --live-ontology
frankctl pipelines apply -f cira-events.yaml --wait

A template pack is a versioned set of the same Source, IdentityPolicy, Pipeline, and BackingDataset declarations described below. Rendering is local and the output always goes through the existing apply, preflight, and execution path. A pack records which components are live-proven, which are only declared, and which stable Source names must already exist. It does not create a parallel runtime or grant Frank ownership of an externally managed ontology schema.

Use pipelines scaffold for a newly discovered feed without a curated pack. Use templates render when the reviewed mappings already exist and should be repeated without re-authoring them.

The mental model: three data stages, up to four resources

The platform is medallion-shaped: data is pulled, shaped, then mapped. The three stage resources are Source, Pipeline, and BackingDataset. Add an IdentityPolicy when a field-mapping Pipeline needs a declared, versioned recipe for stable identifiers. The resulting apply dependency is Source -> IdentityPolicy -> Pipeline -> BackingDataset, regardless of input file order.

Object (kind)	Produces or controls	Engine	Your job
Source	external → bronze (Iceberg)	Temporal	declare where the data is and how to pull it
IdentityPolicy	exact identity recipe pinned by Pipeline	Pipeline activation	declare how source fields become a stable ID
Pipeline	bronze → silver (Iceberg)	Dagster	shape the raw data into the entity's clean shape (1+ steps)
BackingDataset	silver → ontology	Temporal	bind the silver table to an entity type and map columns to properties

When the feed owns extraction configuration, include the Source and its spec.streams in the same manifest. Reapply converges declared streams by name in place; omitted existing streams are preserved as unmanaged drift, not deleted or silently disabled. For a paginated REST dataset, declare one connector-paginated logical stream and point the Pipeline at its one canonical Bronze table—never enumerate page tables in SQL. If the Source is replacing a known page-per-stream configuration, include those legacy stream names with is_enabled: false; this preserves their runtime records while preventing scheduled or direct Source runs from executing them.

A Pipeline holds as many steps as the cleaning needs — minimum one, and that one must do real work. Each step is a Transform of a kind: custom_sql, field_mapping, catalog_pattern, or custom_python. The terminal step's output is the silver table the BackingDataset reads.

The BackingDataset binds to the Pipeline by name. You do not hand-type the silver table name — it is derived from the Pipeline's terminal step.

The loop

 ┌─ datasets preview <bronze>     # 1. SEE the raw data
 │
 │   (one YAML: Source when owned + optional IdentityPolicy + Pipeline + BD)
 │
 ├─ pipelines apply -f feed.yaml  # 2. CREATE + activate (no write yet)
 ├─ pipelines validate <id>       # 3. SANDBOX-preview the cleaned output
 ├─ pipelines apply -f feed.yaml --wait   # 4. EXECUTE: silver + ontology sync
 └─ bd get / datasets preview / bd entities   # 5. VERIFY (status, silver, entities)

Step 1 — see the raw data

frankctl datasets preview \
  bronze.tenant_00000000_arte_ipma_obs_surface.open_data_observation_meteorology_stations_obs_surface_geojson \
  --limit 4

geometry                                               type     properties
{"coordinates": [-31.1301, 39.4582], "type": "Point"}  Feature  {"descDirVento": "---", "humidade": 87.0, ...

This is GeoJSON: nested geometry.coordinates (a [lon, lat] array) and a properties ROW with the measurements. Note the sentinels — IPMA encodes a missing measurement as -99 and a missing wind direction as "---". A naive SELECT * would ship -99 to the ontology as a real temperature.

Step 2 — author the multi-doc

This is your judgment, not a generated artifact. The Pipeline step:

• flattens properties.* / geometry.coordinates[…] (Trino arrays are 1-indexed: coordinates[1] = lon, coordinates[2] = lat),
• nulls out the -99 sentinels with NULLIF,
• builds a composite primary key (station + time) so observations don't collide,
• and dedups + coalesces on the PK with GROUP BY … MAX(…) — MAX ignores NULLs, so a sentinel-blanked duplicate can't clobber a real value.

apiVersion: frank.platform/v1
kind: Pipeline
metadata:
  name: arte_ipma_weather_observed
spec:
  name: arte_ipma_weather_observed
  source_ids: [arte_ipma_obs_surface]
  schedule_config: { type: manual }
  steps:
    - name: weather_observed
      kind: custom_sql
      emits_to: backing_dataset
      sources:
        - iceberg.tenant_00000000_arte_ipma_obs_surface.open_data_observation_meteorology_stations_obs_surface_geojson
      config: { output_layer: silver }
      params:
        sql: >
          SELECT
            CONCAT(CAST(CAST(properties.idEstacao AS BIGINT) AS VARCHAR), '_', properties.time) AS obs_id,
            CAST(CAST(properties.idEstacao AS BIGINT) AS VARCHAR) AS station_code,
            MAX(properties.localEstacao) AS station_name,
            MAX(CAST(NULLIF(properties.temperatura, -99.0) AS DOUBLE)) AS temperature,
            MAX(CAST(NULLIF(properties.humidade, -99.0) AS DOUBLE)) AS relative_humidity,
            MAX(CAST(NULLIF(properties.pressao, -99.0) AS DOUBLE)) AS atmospheric_pressure,
            MAX(CAST(NULLIF(properties.intensidadeVento, -99.0) AS DOUBLE)) AS wind_speed,
            MAX(CAST(NULLIF(properties.precAcumulada, -99.0) AS DOUBLE)) AS precipitation,
            MAX(properties.time) AS date_observed,
            MAX(CAST(geometry.coordinates[2] AS DOUBLE)) AS latitude,
            MAX(CAST(geometry.coordinates[1] AS DOUBLE)) AS longitude
          FROM tenant_00000000_arte_ipma_obs_surface.open_data_observation_meteorology_stations_obs_surface_geojson
          GROUP BY CAST(CAST(properties.idEstacao AS BIGINT) AS VARCHAR), properties.time
---
apiVersion: frank.platform/v1
kind: BackingDataset
metadata:
  name: weather_observed_arte
spec:
  entity_type_id: weather_observed_arte
  entity_type_name: Weather Observed (ARTE IPMA)
  pipeline: arte_ipma_weather_observed     # name ref → resolved to pipeline_id; silver target DERIVED
  primary_key_column: obs_id
  title_key_column: station_name
  property_mappings:
    - { column: obs_id, property: external_id, is_primary_key: true }
    - { column: station_code, property: station_code }
    - { column: station_name, property: station_name }
    - { column: temperature, property: temperature }
    - { column: relative_humidity, property: relative_humidity }
    - { column: atmospheric_pressure, property: atmospheric_pressure }
    - { column: wind_speed, property: wind_speed }
    - { column: precipitation, property: precipitation }
    - { column: date_observed, property: date_observed }
    - { column: latitude, property: latitude }
    - { column: longitude, property: longitude }
  ensure_schema:                            # Frank creates/evolves the entity type
    display_name: Weather Observed (ARTE IPMA)
    tenant_scoped: false                    # global type — no X-Tenant-Id needed
    fields:
      - { field_key: external_id, field_type: { type: string } }
      - { field_key: station_code, field_type: { type: string } }
      - { field_key: station_name, field_type: { type: string } }
      - { field_key: temperature, field_type: { type: number } }
      - { field_key: relative_humidity, field_type: { type: number } }
      - { field_key: atmospheric_pressure, field_type: { type: number } }
      - { field_key: wind_speed, field_type: { type: number } }
      - { field_key: precipitation, field_type: { type: number } }
      - { field_key: date_observed, field_type: { type: string } }
      - { field_key: latitude, field_type: { type: number } }
      - { field_key: longitude, field_type: { type: number } }

Declaring a stable identity recipe

When the stable ID must be derived by a reusable field-mapping recipe, declare that recipe beside the Pipeline instead of embedding a policy UUID:

apiVersion: frank.platform/v1
kind: IdentityPolicy
metadata: { name: stop-time-external-id }
spec:
  name: stop-time-external-id
  strategy: passthrough
  source_fields: [_record_id]
  normalizers: []
  emit_format: "regional:stop_time:{resolved_key}"
  collision_policy: error
---
apiVersion: frank.platform/v1
kind: Pipeline
metadata: { name: regional_stop_times }
spec:
  name: regional_stop_times
  source_ids: [regional-schedules]
  steps:
    - name: stop_time_rows
      kind: field_mapping
      params:
        field_mappings:
          - target_column: external_id
            mapping_kind: identity
            identity_policy: { name: stop-time-external-id }
            field_order: 0

The name-only reference is valid because the policy is declared in the same file. A pre-existing policy must instead be pinned as {name, tier, version}. At apply time Frank resolves either form to an exact stored ID/version. The first successful binding freezes that recipe; a later recipe change creates a new policy version instead of changing the behavior of an existing transform.

Keep those references symbolic in tracked YAML. frankctl rejects authored runtime UUID/version pins and internally adds the complete recipe hash returned by policy reconciliation. Pipeline activation locks and rechecks the selected recipes before changing mappings. If a concurrent writer changed an unfrozen recipe between the policy and Pipeline requests, apply stops with 409 identity_policy_recipe_changed. Re-run the whole file so policy and Pipeline converge together; a Pipeline-only retry is intentionally unsafe.

Server dry-run reports current state but does not reserve it. A clean preflight can still be followed by that 409 if state changes before apply, which is why the real apply always performs the same locked check again.

This changes the generated external_id values, not the ontology schema. For an externally owned target, continue to omit ensure_schema.

Declaring relationship execution order

If a mapping uses is_relationship: true, declare its target as execution metadata on the BackingDataset:

metadata:
  name: air_quality_observed
  dependsOn:
    - entityTypeId: air_quality_station
      ontologyTenantId: cira
spec:
  ontology_tenant_id: cira
  sync_mode: manual
  property_mappings:
    - column: station_key
      property: ref_station
      is_relationship: true
      target_type: air_quality_station
      target_key: name

apply --wait synchronizes a same-file dependency first and requires an external dependency to be already synced. Dependency metadata is not yet persisted/enforced by the server, so every BackingDataset with dependsOn must explicitly use sync_mode: manual; missing mode and on_materialization are rejected before any API request. The CLI directly triggers those manual BackingDatasets in topological order and rejects cycles and self-references instead of relying on filename order or comments.

Mapping an externally owned ontology type

When another team owns the target schema, omit ensure_schema, declare the target tenant explicitly, and run the GET-only compatibility preflight before the first write:

frankctl pipelines apply -f feed.yaml --dry-run-server --live-ontology

Map only fields with the same meaning—not merely compatible carrier types. For example, a source category array is not automatically a target thematic keyword list, and a source age rating is not automatically a target audience. Unsupported values remain available in Bronze until the target model is extended through its own review process.

Geometry and JSON projected as Trino varchar need an explicit structural transform on the mapping:

- column: location_geojson
  property: location
  type: varchar
  sql_transform_id: json_parse

If the target already contains rows, keep the established stable lookup property as the BackingDataset primary key while adopting any new unique ID in a separate mapped column. Changing the lookup property before existing rows carry the new ID can insert duplicates instead of updating in place.

Need help choosing the entity type / mappings?

The AI primitives propose them (they call the platform's Martha workflows):

# Suggest an ontology entity type from a source schema
frankctl ai suggest target-schema -f schema.json
#   → fiware:Weather/WeatherObserved @ 0.95

# Suggest column → property mappings (with transforms) between two schemas
frankctl ai suggest field-mappings -f schemas.json
#   → temperatura → temperature; id_estacao → stationCode  CAST(id_estacao AS VARCHAR)

target-schema expects {source_schema: [...], source_table: "..."}; field-mappings expects {source_schema, target_schema}.

Step 3 — sandbox-preview before writing

apply (without --wait) creates and activates the pipeline; then validate runs it in a sandbox (sampled, no real table write) so you can confirm the cleaning is correct:

frankctl pipelines apply -f feed.yaml
frankctl pipelines validate <pipeline-id> --sample-limit 200

[ok] weather_observed [custom_sql]  (200 rows, 556ms)
  obs_id=1210746_2026-06-17T09:00:00  temperature=17.4  atmospheric_pressure=1018.9  latitude=39.1259 …
  obs_id=1210713_2026-06-17T09:00:00  temperature=21.1  atmospheric_pressure=null    latitude=40.1398 …

The -99 became null, coordinates resolved, the PK is unique. Good to ship.

Step 4 — execute

frankctl pipelines apply -f feed.yaml --wait

  ✓ silver materialized step=weather_rows dagster_run_id=... transform_run_id=... snapshot=7779148679244111000 rows=543
  ✓ ontology synced workflow_id=... ontology_sync_run_id=... snapshot=7779148679244111000 rows=543
  ✓ all declared stages reached correlated terminal success; ontology entity readback remains separate proof.

--wait drives the whole chain: activate → materialize silver → create the BackingDataset (its silver target derived from the pipeline) → sync to the ontology. It follows the exact Dagster, TransformRun, and ontology SyncRun IDs; it does not accept an older "latest" run as evidence. On an unchanged rerun, ontology already converged; no replay is a successful no-op, not proof of a new write. With --json, each BackingDataset result also contains the exact persisted REST bulk job IDs and submitted/completed counts when the bulk path was used; payloads and credentials are never part of that evidence.

Step 5 — verify, all in the CLI

frankctl bd get <bd-id>                                  # status: synced
frankctl datasets preview silver.tenant_00000000_transforms.arte_ipma_weather_observed_weather_observed
frankctl bd entities <bd-id> --limit 5                   # read the ontology back

entity_type weather_observed_arte — 5 entities
external_id                  station_name             temperature  atmospheric_pressure  latitude  longitude
1200522_2026-06-17T11:00:00  Funchal                  22.5         1022.2                32.6479   -16.888
1210984_2026-06-17T11:00:00  Madeira, Quinta Grande   18.3                               32.663    -17.015

Real values, sentinels nulled, coordinates correct — confirmed against the ontology itself, never leaving frankctl.

Authoring checklist

The mechanism is seamless; the judgment is yours. Before you apply --wait:

• [ ] Primary key is unique. Composite it (station + time) if one column isn't. A non-unique PK means the ontology upsert collapses rows.
• [ ] Every relationship has metadata.dependsOn. Match the target entity type and ontology tenant; self-references are unsupported.
• [ ] Duplicate policy is explicit. Prefer a uniqueness failure for ordinary whole-row records. If deduplication is intentional, select one complete row deterministically. Use fieldwise GROUP BY … MAX(…) only when the source contract explicitly defines duplicate rows as complementary fragments that may be coalesced.
• [ ] Sentinels nulled (NULLIF(col, -99) etc.) — don't ship magic numbers as real measurements.
• [ ] Types cast to match the ensure_schema field types (number vs string).
• [ ] Envelope columns dropped (_engine, _extracted_at, …) — they're lineage metadata, not entity properties.
• [ ] Sandbox-validated (pipelines validate) before the real run.
• [ ] Entities verified (bd entities) after.

Keeping a feed live: scheduling

Everything above runs the feed once, on demand. Whether a feed keeps flowing on its own is controlled independently at each of the three stages — and a feed is only truly automatic when all three are set. By default each stage is manual, so a freshly authored feed is a one-shot until you schedule it.

Stage	Field	Manual (default)	Automatic
Source → bronze	schedule_type	manual — pulls only on sources sync	cron or interval (+ schedule_value) — keeps pulling
Pipeline → silver	schedule_config.type	manual — runs only through an explicit trigger	eager — reacts to upstream materialization; cron or interval — runs on an independent wall clock
BackingDataset → ontology	sync_mode	—	on_materialization (default) — pushes on every new silver snapshot

The chain is event-driven once wired: a scheduled Source lands fresh bronze → an eager Pipeline re-materializes silver → an on_materialization BackingDataset pushes the delta to the ontology. No manual step in the loop.

Source and Pipeline clocks are deliberately independent. A Pipeline cron or interval does not wait for a Source sync and may materialize unchanged Bronze data. Prefer eager when the intended contract is "run after fresh input"; use a Pipeline clock only when the transform itself has a time-based reason to run.

This is an upsert chain, not a full mirror: replacing Bronze/Silver does not delete ontology entities that disappear upstream, and null mapped values do not clear properties already present on the target. Treat deletion reconciliation and explicit null-clearing as separate product capabilities.

This automatic mode is only safe for BackingDatasets without declarative relationship prerequisites. A BackingDataset with metadata.dependsOn must use sync_mode: manual and be run through frankctl pipelines apply --wait until dependency graphs are persisted and enforced server-side.

# Source — pull every 30 minutes
kind: Source
spec:
  schedule_type: cron
  schedule_value: "*/30 * * * *"
---
# Pipeline — re-materialize whenever bronze updates
kind: Pipeline
spec:
  schedule_config: { type: eager }
  steps: [ ... ]
---
# BackingDataset — auto-push new data or an effective mapping change (default)
kind: BackingDataset
spec:
  sync_mode: on_materialization
  pipeline: my_pipeline

Gotcha: Scheduled data refresh still requires something to produce a new silver snapshot. A mapping or other effective-SyncSpec change can replay an existing snapshot, but it does not create fresh source data. Make the Pipeline eager (and the Source scheduled) when the BackingDataset must continuously publish new data.

Operate the Pipeline schedule without editing its generated Transforms:

frankctl pipelines schedule get <pipeline-id>
frankctl pipelines schedule set <pipeline-id> --type eager
frankctl pipelines pause <pipeline-id>
frankctl pipelines resume <pipeline-id>
frankctl pipelines trigger <pipeline-id>

The get/set/pause/resume commands succeed only after the effective Dagster root configuration is read back as matching the durable Pipeline intent.

During a rolling sync-contract upgrade, existing V1 Temporal histories remain on ontology-sync-task-queue; new V2 starts use ontology-sync-v2-task-queue. Operators must establish the V2 poller before cutting the API over, while keeping the V1 worker until legacy histories and retries drain.