Ingestion

Domain: Datasets
Depends on: Data Model, Data Schema, Resource Framework
RFCs: RFC-009, RFC-011, RFC-014

Abstract

This spec defines how external data is brought into . It defines the Source resource that describes where data comes from and how it is read, the phases of the ingestion process, the that turn data of various nature into an append-only history of events, how is determined, and how the state of a source is preserved to allow incremental ingestion.

Purpose

When interacting with data on the web we cannot make any assumptions about the guarantees the external source provides:

Just by looking at the data we cannot tell if it is mutable or preserves the history of changes
We often can’t tell where it originates from and who can be held accountable for its validity
We can’t be sure that it’s sufficiently replicated - the domain that hosts it may not exist the next day

is about getting the external data into the system as efficiently as possible, into the strict ledgers of , where properties like clear ownership, reproducibility, accountability, and complete historical account can be guaranteed and made explicit. This spec does not aim to standardize all possible data ingestion techniques. ODF is not meant to replace data workflow and data extraction tools - the ingestion mechanisms described here simplify the very first step in the data’s journey through the stream processing pipelines, and implementations may support additional ingress types, formats, and steps.

Sources

A source of data is described by a Source :

$schema: https://opendatafabric.org/schemas/sources/v1alpha1/Source
headers:
  name: ca.bankofcanada
spec:
  # Makes variables and secrets available for templating under `config`
  config:
    apiKey: SecretSet:ca.bankofcanada#apiKey
  # Where the data comes from
  ingress:
    kind: Url
    url: https://www.bankofcanada.ca/valet/observations/group/FX_RATES_DAILY/csv
    headers:
      - name: Authorization
        value: "Bearer ${{ config.apiKey }}"
  # How to read the data into structured form
  read:
    kind: Csv
    header: true
    schema:
      fields:
        - name: date
          type: Timestamp
        - name: FXUSDCAD
          type:
            kind: Decimal
            precision: 19
            scale: 6
  # OPTIONAL: Query that shapes the data
  preprocess:
    kind: Sql
    engine: datafusion
    query: |
      select
        date as event_time,
        FXUSDCAD as fxusdcad
      from input
  # How to combine new data with the history of previously seen data
  merge:
    kind: Ledger
    primaryKey:
      - event_time

A source is decoupled from the datasets it writes to. Ingestion is performed by an Ingest task that references the Source and the target root dataset, and that can be scheduled by a flow (see Tasks and Flows):

$schema: https://opendatafabric.org/schemas/flows/v1alpha1/Flow
headers:
  name: ca.bankofcanada
spec:
  target: Dataset:ca.bankofcanada
  triggers:
    - kind: Schedule
      cron: "*/30 * * * *"
  tasks:
    - kind: Ingest
      source: Source:ca.bankofcanada
      targetRecordsPerSlice: 10000

All data ingested into a dataset must conform to the dataset’s schema, as captured by the event.

Push and Polling Sources

Depending on the type of ingress, sources fall into two categories:

Polling sources - the node periodically scans a remote data source and ingests the latest updates. This is suitable for periodic database dumps, data published as files on the web, bulk data APIs, and blockchains.
Push sources - an external actor actively sends data into the node. This is suitable for IoT devices, business processes that report events directly, and streaming data APIs and event queues.

Although we aspire to reach a state where all authoritative data publishers push new events as soon as those occur, we realize that this level of integration will take time to achieve, and for a long time the vast majority of data will be ingested via the polling model.

Ingestion Phases

The ingestion process consists of the following phases, directly reflected in the structure of the :

Phase	Field	Description
Ingress	`ingress`	Obtains the data from an external source in its original raw form (e.g. via HTTP), or accepts the data pushed by an external actor
Prepare (optional)	`prepare`	Prepares raw binary data for reading, e.g. by extracting an archive or converting between formats
Read	`read`	Reads the data in some format into a structured form
Preprocess (optional)	`preprocess`	Reshapes the structured data with a query, e.g. renaming columns, type conversions, nesting
Merge	`merge`	Combines the new data with the history of previously seen data (see Merge Strategies)

Additionally:

vocab - specifies the names of the system columns (see ) in case the data uses names different from the defaults
config - brings variables and secrets into the config context, so they can be used for templating values of other fields (e.g. ${{ config.apiKey }})

Ingress

The ingress field () determines where data comes from:

Kind	Type	Description
`Url`	Polling	Fetches data by its URL using one of the supported protocols (e.g. HTTP, FTP), optionally passing request headers
`FilesGlob`	Polling	Matches a growing set of files on the local file system using a glob pattern
`Container`	Polling	Runs an OCI container to fetch data from an arbitrary source (see Container Interface)
`EvmLogs`	Polling	Streams transaction logs from an EVM-compatible blockchain node (see EVM Logs)
`Mqtt`	Push	Subscribes to topics of an MQTT broker
`RestEndpoint`	Push	Exposes an HTTP endpoint that accepts pushed records

FilesGlob is used for directories containing a growing set of files, e.g. periodic database dumps or batches of new data. Files should never change once added - the source remembers the last ingested file and only considers files that come after it in the order specified by order: lexicographically by name (ByName) or by event time (ByEventTime). Every file is processed individually and advances the of the dataset. Push ingresses buffer the received records until they are ingested by a task. The RestEndpoint ingress specifies the buffer () with its size and the overflowPolicy applied when the buffer is full: rejecting new records (Reject) or dropping the oldest ones (DropOldest).

Event Time Sources

Ideally, data records contain within them as a column, but many data sources don’t. Polling ingresses can specify how event time is determined via the eventTime field ():

Kind	Description
`FromMetadata`	Extracts event time from the metadata of the source, e.g. modification time of files or the `Last-Modified` header of HTTP resources
`FromPath`	Extracts event time from the path of the source using a regular expression whose first group contains the timestamp, and a timestamp format
`FromSystemTime`	Uses the of ingestion as the event time

Example of extracting event time from file names:

ingress:
  kind: FilesGlob
  path: /data/db-table-dump-*.csv
  eventTime:
    kind: FromPath
    pattern: 'db-table-dump-(\d+-\d+-\d+)\.csv'
    timestampFormat: '%Y-%m-%d'

Event time is especially important for the Snapshot merge strategy, as snapshot-style exports represent the state of a system at a certain time that is usually captured only in some form of metadata (e.g. in the name of the file, in the URL, or in HTTP caching headers).

Caching

Implementations should avoid redundant work and not ingest data if the source was not updated since the last poll. The exact mechanism depends on the ingress type and the protocol, e.g. for HTTP it relies on standard caching headers like ETag and Last-Modified, and the state needed for this is preserved as source state. Caching behavior can be controlled via the cache field (). The Forever kind specifies that after the source was processed once it will never be ingested again.

Container Interface

The Container ingress allows using a general purpose programming language to deal with particularly complex APIs, or for web scraping. The specified OCI image is expected to conform to the following interface:

Produce data to stdout
Write warnings and errors to stderr
Use the following environment variables:

Variable	Description
`ODF_LAST_MODIFIED`	Last modified time of data from the previous ingest run, if any (in RFC 3339 format)
`ODF_ETAG`	Caching tag of data from the previous ingest run, if any
`ODF_BATCH_SIZE`	Recommended number of records, for scripts that provide a continuous stream of data and can resume from previous state
`ODF_NEW_LAST_MODIFIED_PATH`	Path to a text file where the script may write the new `Last-Modified` timestamp
`ODF_NEW_ETAG_PATH`	Path to a text file where the script may write the new `ETag`
`ODF_NEW_HAS_MORE_DATA_PATH`	Path to a file which the script can create to indicate that more data is available for the next batch. When the file is created, the new `ETag` or `Last-Modified` must also be written

EVM Logs

The EvmLogs ingress reads logs of EVM-compatible blockchains directly from a node. The filter field is an SQL WHERE clause used to create the most efficient RPC request, and the optional signature specifies a Solidity event signature used to decode logs. The output structure corresponds to the output of the eth_getLogs RPC method:

Column	Type	Nullable	Description
`block_number`	`UInt64`
`block_hash`	`Binary`
`block_timestamp`	`Timestamp`	✅	Not returned by all node providers
`transaction_index`	`UInt64`
`transaction_hash`	`Binary`
`log_index`	`UInt64`
`address`	`Binary`
`topic0` … `topic3`	`Binary`	✅
`data`	`Binary`
`event`	`Struct`		Decoded event, only present when `signature` is specified

Prepare Steps

The prepare field specifies a list of steps () applied to the raw data before reading:

Kind	Description
`Decompress`	Extracts data compressed with `Gzip` or `Zip`. For multi-file archives `subPath` (can contain glob patterns) specifies which file to extract
`Pipe`	Executes an external command that receives data via `stdin` and outputs the result to `stdout`, e.g. to transcode formats that are not natively supported

Read Formats

The read field () specifies how data is read into the structured form:

Kind	Description
`Csv`	Comma-separated values, with options for separators, quoting, escaping, headers, null values, and date / timestamp formats
`Json`	JSON document containing an array of objects, optionally located at `subPath` within the document
`NdJson`	Newline-delimited JSON objects with the same schema
`GeoJson`	GeoJSON document with a `FeatureCollection` object in the root, creating a record per each `Feature` with properties extracted into individual columns and the geometry left in its own column
`NdGeoJson`	Similar to `GeoJson`, but with every `Feature` object on its own line
`EsriShapefile`	ESRI Shapefile, optionally located at `subPath` within an archive
`Parquet`	Apache Parquet

The read step can specify the schema of the data in the Data Schema format, used to coerce values into appropriate data types. Formats like JSON can’t be split and read efficiently, so newline-delimited variants should be preferred where possible.

Preprocessing

The optional preprocess field specifies an engine-specific query () that shapes the data after reading, e.g. to parse values, convert types, or rename columns to match the dataset schema. The query refers to the read data as input:

preprocess:
  kind: Sql
  engine: datafusion
  query: |
    select
      date,
      city,
      cast(replace(population, ',', '') as bigint) as population
    from input

Preprocessing queries are executed by using the Raw Query operation.

Merge Strategies

By design, ODF stores data only in the append-only event log format to preserve the entire history. Unfortunately, a lot of data in the world is not stored or exposed this way - some publishers expose data as periodic database dumps, while others provide a log of changes between the current and the previous export. () bridge the gap between the wide variety of data in the external world and the strict event-based world of ODF, by defining how the newly-ingested data is combined with the existing history. All strategies except Append rely on a user-specified primaryKey - the names of columns that uniquely identify a record throughout its lifetime.

Append

Under this strategy the new data is appended to the dataset in its entirety, without any modifications or deduplication. This strategy is rarely used for polling sources, but can be useful for push sources (e.g. IoT devices).

Ledger

This strategy should be used for data sources containing append-only ledgers of events. New data exports can have new rows added, but once data made it into one export it should never change or disappear. The primary key is used to identify which events were already seen, so that the source data can contain partially overlapping sets of records and only those records that were not previously seen are appended. For example, a source with historical populations of cities:

Year,Country,City,Population
2019,CA,Vancouver,2581000
2019,US,Seattle,3433000

Ingested with primary key (year, country, city) results in:

op	system_time	event_time	country	city	population
+A	s1	2019	CA	Vancouver	2581000
+A	s1	2019	US	Seattle	3433000

When a new row for 2020 appears in the source, only that row is appended, while rows for 2019 are recognized as already seen:

op	system_time	event_time	country	city	population
+A	s1	2019	CA	Vancouver	2581000
+A	s1	2019	US	Seattle	3433000
+A	s2	2020	CA	Vancouver	2606000

Having a time column doesn’t always mean that the data is a ledger - to be a true ledger all previous records must be immutable.

Snapshot

This strategy can be used for data exports that are taken periodically and contain only the latest state snapshot of the observed entity or system. Over time such exports can have new rows added, and old rows either removed or modified. Publishers that use such formats routinely lose information - when a record is updated or one export replaces another, the history of previous values is lost. This strategy transforms snapshot data into an append-only event stream by performing change data capture - diffing the current state of data against the reconstructed previous state. The primary key is used to correlate rows between the two snapshots, and the column of the output will contain:

Append (+A) - when a row with a certain key appears for the first time
Retraction (-R) - when a row with a certain key disappears from the snapshot
Correction (-C, +C) - when the data associated with a certain key has changed, with -C carrying the old values and +C carrying the new values (see )

To identify whether a row has changed, this strategy compares all columns one by one. If the data contains a column that is guaranteed to change whenever any other column changes (e.g. a modification timestamp, a version, or a data hash), it can be specified in compareColumns to speed up the detection of modified rows. For example, a source with current populations of cities, ingested with primary key (country, city) and event time taken from the source metadata:

Country,City,Population
CA,Vancouver,2581000
US,Seattle,3433000

op	system_time	event_time	country	city	population
+A	s1	e1	CA	Vancouver	2581000
+A	s1	e1	US	Seattle	3433000

After the population of Vancouver changes in the source, ingestion produces a correction:

op	system_time	event_time	country	city	population
+A	s1	e1	CA	Vancouver	2581000
+A	s1	e1	US	Seattle	3433000
-C	s2	e1	CA	Vancouver	2581000
+C	s2	e2	CA	Vancouver	2606000

Changelog Stream

This strategy is used for sources that already provide data in the native ODF changelog format that describes appends, retractions, and corrections (see Data Model). No processing except for format validation is performed.

Upsert Stream

This strategy is used for sources containing ledgers of insert-or-update and delete events. Unlike in the changelog stream, insert-or-update events only carry the new values, so this strategy uses the primary key to look up the previous values and re-classify events into appends or correction pairs.

Committing Data

The result of every ingestion is committed into the of the dataset as an event, which carries:

prevOffset and newData - the last offset of the previous data slice and the new data slice, if any, allowing the next ingestion to be prepared without deep scanning of the chain
prevCheckpoint and newCheckpoint - checkpoint used to restore and persist the ingestion state, if any
newWatermark - the new of the dataset, if any
newSourceState - the state of the source, if any

An ingestion may produce no new data but still commit a block, e.g. to advance the watermark or to update the source state.

Source State

To avoid expensive recomputations, the state of the source the data was ingested from (e.g. an ETag of an HTTP resource, the name of the last file matched by a glob pattern, or the height of the last indexed block of a blockchain) is stored in the newSourceState field () of the AddData event:

newSourceState:
  sourceName: odf/polling
  kind: odf/etag
  value: W/"95c5fde3918cba7c33eaac7ff9d02b22"

Where:

sourceName - identifies the source the state belongs to
kind - identifies the type of the state, similarly to MIME types. Standard kinds are odf/etag (an identifier with semantics of the ETag HTTP header) and odf/last-modified (an RFC 3339 timestamp with semantics of the Last-Modified HTTP header)
value - an opaque value of the state

Both sourceName and kind are plain strings, allowing implementations to define their own extensions. It is safe to ignore source state that an implementation does not understand. If the state didn’t change but is still relevant for subsequent runs it should be carried over into the new event, as only the last state per source is considered when resuming.

Validity of Root Data

Unlike data of , data of cannot be reproduced - it captures observations of the outside world. Its validity therefore depends entirely on its publisher, who is in full control of the data they present, so measuring and processing errors or even malicious data can make their way into a root dataset. What can be validated mechanically is that root data was not altered after it was published (see Metadata Chain: Validation and Data & Checkpoint Hashing: Validation and Verification), and who published it. Trust in the data itself is established through:

Authority - the reputation of the publisher, e.g. organizations that directly operate the systems from which the data is collected
Cross-validation - comparing data against other similar sources to detect anomalies

Legacy Source Events

Deprecated: Sources used to be defined in the of root datasets using the following events. They are superseded by Source resources, as source definitions don’t contribute to provenance or validity of a dataset, but may reveal sensitive information about the infrastructure of a publisher (see RFC-018).

Event	Description
`SetPollingSource`	Defines the polling source of a dataset, with `fetch`, `prepare`, `read`, `preprocess`, and `merge` steps
`DisablePollingSource`	Disables the polling source
`AddPushSource`	Defines a named push source of a dataset, with `read`, `preprocess`, and `merge` steps
`DisablePushSource`	Disables a named push source

Under this model:

A dataset may have multiple named push sources active simultaneously (e.g. for devices from different manufacturers producing slightly different data), but push and polling sources are mutually exclusive
All sources must result in the same final data schema, as captured in the SetDataSchema event
The source state of the polling source uses the odf/polling source name

Schemas

Schema	Description
	Source resource
	Specification of a source
	Where data comes from
	Buffering of pushed records
	How event time is determined
	Caching of external data
	Ordering of input files
	Preparation of raw data
	Reading data into structured form
	Combining new data with history
	State of a source for incremental ingestion
	Metadata event committing the ingested data

Open Data Fabric

Kamu CLI

Kamu Node

Abstract

Purpose

Sources

Push and Polling Sources

Ingestion Phases

Ingress

Event Time Sources

Caching

Container Interface

EVM Logs

Prepare Steps

Read Formats

Preprocessing

Merge Strategies

Append

Ledger

Snapshot

Changelog Stream

Upsert Stream

Committing Data

Source State

Validity of Root Data

Legacy Source Events

Schemas

​Abstract

​Purpose

​Sources

​Push and Polling Sources

​Ingestion Phases

​Ingress

​Event Time Sources

​Caching

​Container Interface

​EVM Logs

​Prepare Steps

​Read Formats

​Preprocessing

​Merge Strategies

​Append

​Ledger

​Snapshot

​Changelog Stream

​Upsert Stream

​Committing Data

​Source State

​Validity of Root Data

​Legacy Source Events

​Schemas

Abstract

Purpose

Sources

Push and Polling Sources

Ingestion Phases

Ingress

Event Time Sources

Caching

Container Interface

EVM Logs

Prepare Steps

Read Formats

Preprocessing

Merge Strategies

Append

Ledger

Snapshot

Changelog Stream

Upsert Stream

Committing Data

Source State

Validity of Root Data

Legacy Source Events

Schemas