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.

The mental model: three objects, three stages

The platform is medallion-shaped. You author exactly three declarative objects, one per stage — you can't go below three, because the data must be pulled, shaped, then mapped:

Object (kind)	Produces	Engine	Your job
Source	external → bronze (Iceberg)	Temporal	declare where the data is + how to pull it
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 + map columns → properties

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
 │
 │   (you author one multi-doc YAML: Pipeline + BackingDataset)
 │
 ├─ 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 } }

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 (543 rows)
  ✓ ontology synced (snapshot 7779148679244111000)
  ✓ all stages green — pipeline live to the ontology.

--wait drives the whole chain: activate → materialize silver → create the BackingDataset (its silver target derived from the pipeline) → sync to the ontology.

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.
• [ ] Dedup + coalesce on the PK in the transform (GROUP BY … MAX(…)), so a NULL/sentinel duplicate never overwrites a real value.
• [ ] 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 — re-materializes only on pipelines apply --wait	eager — re-materializes whenever its bronze input updates
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 — 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 each new silver snapshot (default)
kind: BackingDataset
spec:
  sync_mode: on_materialization
  pipeline: my_pipeline

Gotcha: sync_mode: on_materialization only fires when something produces a new silver snapshot. If the Pipeline is manual, no new snapshots are produced, so the BackingDataset never auto-syncs — the policy is set but dormant. Make the Pipeline eager (and the Source scheduled) for the BackingDataset's policy to actually do anything.