Skip to main content

Connectors: connect to data sources and sinks

A Feldera pipeline can process data from multiple heterogeneous sources and produce outputs to multiple heterogeneous destinations. To this end it relies on a growing library of input and output connectors.

Basics​

Users configure connectors using a JSON object, that describes an external source or sink such as a database table or a Kafka topic. A SQL table can have multiple source connectors attached to it, specified as a list of JSON connector objects under the 'connectors' attribute in the WITH clause of the table. Similarly, a views can have multiple sink connectors attached to it.

Here is an example, where the VENDOR table has one connector, configured to fetch and insert JSON data from an HTTP URL, and a VENDOR_VIEW which sends the changes of the view to a Kafka topic in a format that can be consumed by Debezium:

CREATE TABLE vendor (
    id BIGINT NOT NULL PRIMARY KEY,
    name VARCHAR,
    address VARCHAR
) WITH ('connectors' = '[{
    "transport": {
        "name": "url_input", "config": {"path": "https://feldera-basics-tutorial.s3.amazonaws.com/vendor.json"}
    },
    "format": { "name": "json" }
}]');

CREATE VIEW vendor_view
WITH (
    'connectors' = '[{
        "max_queued_bytes": 1000000,
        "format": {
            "name": "json",
            "config": {
                "update_format": "debezium"
            }
        },
        "transport": {
            "name": "kafka_output",
            "config": {
                "bootstrap.servers": "redpanda:9092",
                "topic": "test_view"
            }
        }
    }]'
)
AS SELECT * FROM vendor;
info

The WITH clause for tables needs to be put at the end, after the column definitions, whereas for views it has to appear before the AS clause to resolve any parsing ambiguities.

A connector specification consists of three parts:

note

Some transports, e.g., Delta Lake and datagen, use fixed predefined data formats and do not require the format section in the connector specification.

This architecture allows the user to combine different transports and data formats.

These basics apply to all connectors except the HTTP input and output connectors which are not managed by the user, as they directly feed/fetch data into/from a pipeline via dedicated pipeline endpoints and therefore do not need to be configured in the WITH clauses of tables and views.

Generic attributes​

The following attributes are common to all connectors:

  • name - The name that is given to the connector, which must be unique among the connectors of the table or view. This is particularly useful to define when wanting to refer to it, for example to start or pause it at runtime. By default, it will be named unnamed-{index}, with the zero-based index within the list of connectors of its table/view.

  • paused - If set to to true the connector will not fetch or push data to the pipeline when started unless explicitly enabled through the API. By default this is set to false.

  • labels - An optional list of text labels associated with the connector. This property is used in conjunction with the start_after property to implement automatic connector orchestration.

  • start_after - Specifies one or more labels. When this property is set, the connector is created in the Paused state and is automatically activated once all connectors tagged with at least one of the specified labels have finished ingesting data. This property is used in conjunction with the labels property to implement automatic connector orchestration.

  • max_queued_records and max_queued_bytes - The approximate maximum number of records or bytes to keep in memory.

    For an input connector, these are maximum amounts that the endpoint will read into memory, before the endpoint pauses further reading until the pipeline has consumed some of the backlog. We recommend setting max_queued_bytes rather than max_queued_records to limit memory consumption, since records can be any size.

    For an output connector, max_queued_records is the maximum number that the endpoint will hold in memory waiting for the output endpoint to send them, before the circuit pauses execution until the backlog subsides. Output connectors do not yet honor max_queued_bytes.

    These values are approximate: the pipeline pauses input or execution soon after they are exceeded, but not always instantly.

    By default, max_queued_records is 1,000,000. If max_queued_bytes is unspecified, then it defaults to 1000 * max_queued_records.

  • max_batch_size - Maximum number of records from this connector to process in a single batch.

    When set, this caps how many records are taken from the connector’s input buffer and pushed through the circuit at once.

    This is typically configured lower than max_queued_records to allow the connector time to restart and refill its buffer while a batch is being processed.

    Not all input adapters honor this limit.

    If this is not set, the batch size is derived from max_worker_batch_size.

  • max_worker_batch_size - Maximum number of records processed per batch, per worker thread.

    When max_batch_size is not set, this setting is used to cap the number of records that can be taken from the connector’s input buffer and pushed through the circuit at once. The effective batch size is computed as: max_worker_batch_size × workers.

    This provides an alternative to max_batch_size that automatically adjusts batch size as the number of worker threads changes to maintain constant amount of work per worker per batch.

    Defaults to 10,000 records per worker.

  • index – (Output connectors only) The name of an index created by a SQL CREATE INDEX statement that defines the unique key for the view. This allows the connector to combine related insert and delete events into a single atomic update. See Uniqueness Constraints.

  • send_snapshot – (Output connectors only) When true, the connector emits a full snapshot of the materialized view the first time it runs, before streaming incremental updates. The view must be materialized (declared with CREATE MATERIALIZED VIEW). The default is false.

    The snapshot is sent exactly once: resuming from a checkpoint does not re-send it. Modifying the connector configuration causes a fresh snapshot to be replayed on the next start.

  • soft_delete – (Input connectors only) When true, the connector ingests deletions as insertions and reports the original polarity of each record in the is_delete metadata attribute. The default is false. See Soft deletes.

Soft deletes​

An input connector normally applies the changes it reads to the table it is attached to: an insertion adds a record, a deletion removes it. The table therefore holds the current contents of the input stream, and the records the stream deleted are gone.

Setting soft_delete to true converts the table into a log of every operation the connector reads. The connector pushes every record to the table as an insertion, including the records the stream deletes, and attaches the is_delete metadata attribute to the deleted ones. The table then represents the unbounded stream of updates rather than the current contents of the stream, and a query can select the part of it that it needs.

SQL programs can access the polarity of a record with the CONNECTOR_METADATA() function:

Metadata attributeSQL typeCONNECTOR_METADATA() field
Record polarityBOOLEANis_delete

The attribute is true for a deleted record and absent for an inserted one, so the column is NULL for insertions.

A SQL query can recover the current contents of the stream from such a log by ranking the changes of each key by time, keeping the most recent one, and returning it only when it is an insertion.

Such a query needs a column that orders the changes of a key in time. Any column of the table can serve. When the record itself carries no suitable timestamp, a connector metadata attribute is often the better choice, since several connectors report the time at which they received each change. Feldera does not add such a column on its own, so declare one that suits the source. The example below uses the timestamp the Kafka connector reports for each message as the kafka_timestamp metadata attribute:

CREATE TABLE changes (
    id BIGINT,
    s VARCHAR,
    -- Both columns come from connector metadata: the timestamp of the change,
    -- and whether it was a deletion.
    ts TIMESTAMP DEFAULT CAST(CONNECTOR_METADATA()['kafka_timestamp'] AS TIMESTAMP),
    is_delete BOOLEAN DEFAULT CAST(CONNECTOR_METADATA()['is_delete'] AS BOOLEAN)
) WITH (
    'connectors' = '[{
        "soft_delete": true,
        "transport": {
            "name": "kafka_input",
            "config": {
                "topic": "changes",
                "start_from": "earliest",
                "bootstrap.servers": "example.com:9092",
                "include_timestamp": true
            }
        },
        "format": {
            "name": "json",
            "config": { "update_format": "insert_delete" }
        }
    }]'
);

-- The current contents of the stream: the latest change of each key, kept only
-- when that change was an insertion.
CREATE MATERIALIZED VIEW live AS
SELECT id, s, ts
FROM (
    SELECT *, ROW_NUMBER() OVER (
        PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST
    ) AS rn
    FROM changes
)
WHERE rn = 1 AND is_delete IS NOT TRUE;

Ranking one key at a time and keeping a single row is the TopK pattern, which Feldera evaluates efficiently.

The is_delete term in the ORDER BY clause breaks the tie between changes that carry the same timestamp. A CDC stream reports an update as a single message that deletes the old value of a record and inserts the new one, so those two changes carry one timestamp; ranking the insertion first keeps the updated record, where ranking the deletion first would drop it.

As written, live needs unbounded state. A change carrying any timestamp can arrive at any moment and displace the record that is currently the latest one for its key, so Feldera has to keep every insertion and deletion the stream ever reported. An application that only needs values from a bounded time frame can filter the changes with a temporal filter before ranking them, which bounds the state of the query; see Soft deletes with temporal filters.

Notes and restrictions:

  • The table must not have a primary key: a deletion in such a table identifies a key rather than a record, so there is nothing to insert in its place. The compiler rejects a soft_delete connector attached to a table that has one.

  • Soft deletes are a property of the connector, not of the data format, so they apply to every format that expresses deletions, e.g., the JSON and Avro change event formats and Debezium streams, as well as to integrated connectors such as Delta Lake in cdc mode. A format that only expresses insertions, e.g., CSV or Parquet, is unaffected.

  • Like any column default, the is_delete column takes its value from the metadata only when the record itself does not contain the column. A record that carries an is_delete value overrides the polarity reported by the connector. Use a column name that doesn't occur in the table to avoid such conflicts.

  • Deletions that a soft-delete connector reads do not cancel out earlier insertions of the same record, so the table can receive several copies of a record that the input stream inserted and deleted repeatedly.

Configuring the output buffer​

By default a Feldera pipeline sends a batch of changes to the output transport for each batch of input updates it processes. This can result in a stream of small updates, which is normal and even preferable for output transports like Kafka; however it can cause performance problems for other connectors, such as the Delta Lake connector by creating a large number of small files.

The output buffer mechanism is designed to solve this problem by decoupling the rate at which the pipeline pushes changes to the output transport from the rate of input changes. It works by accumulating updates inside the pipeline for up to a user-defined period of time or until accumulating a user-defined number of updates and writing them as a single batch to the output transport.

The output buffer can be setup for each individual output connector as part of connector configuration. The following parameters are used to configure the output buffer:

  • enable_output_buffer - Enable output buffer.

  • max_output_buffer_time_millis - Maximum time in milliseconds data is kept in the output buffer.

    When not specified, data is kept in the buffer indefinitely until one of the other trigger conditions is satisfied. When this option is set the buffer will be flushed at most every max_output_buffer_time_millis milliseconds.

    This configuration option requires the enable_output_buffer flag to be set.

  • max_output_buffer_size_records - Maximum number of updates to be kept in the output buffer.

    This parameter bounds the maximal size of the buffer. Note that the size of the buffer is not always equal to the total number of updates output by the pipeline. Updates to the same record can overwrite or cancel previous updates.

    When not specified, defaults to 10,000,000.

    This configuration option requires the enable_output_buffer flag to be set.

See Delta Lake output connector documentation for an example of configuring the output buffer.

Additional resources​

For more information, see: