Skip to content
ABRQ DATADocs Abrq DIP · latest
Product page Request a trial
On this page

Streams (API ingestion)

What streams do#

A stream pulls data out of an HTTP API on a schedule. Abrq DIP issues an outbound GET request to an endpoint you configure, reads the JSON response, maps selected fields to columns, and inserts rows into a destination table.

Note. Streams are outbound pull, not inbound webhooks. Abrq DIP calls your API; your API never calls Abrq DIP. There is no inbound webhook URL and no ingest token to hand out. If you need a push-style feed, land it in object storage or an FTP share and use a file ingestion instead.

In v1 only the GET method is supported. Everything else about the call — query parameters, headers, pagination — is configuration on the stream.

Use streams for public and partner REST APIs, internal service endpoints that expose a JSON list, and any paginated "give me the next page" API that returns a cursor or accepts offset/limit.

Before you begin#

  • Licensed feature: streams. Without it, the Streams nav item is hidden and opening the route directly shows Streams (API ingestion) isn't in your license. Talk to Abrq about adding the entitlement — see Entitlements.
  • Role. A viewer can open streams, read run history, and read logs. An editor is required to create, edit, run, or delete a stream, and to manage its destinations.
  • A destination connector. The primary destination of a stream must be a Postgres or MySQL connector. Create it first — see Data connectors.
  • The endpoint details: the full URL, any authentication headers, and (if the records are wrapped in an envelope) the path to the list of records.

Create a stream#

The wizard has three steps: Connect & probe → Map data → Destination & schedule.

Step 1 — Connect & probe#

Field Notes
Name Identifies the stream in lists, run history, and logs.
Endpoint URL The full URL Abrq DIP will GET, for example https://api.example.com/v1/orders.
Records path (optional) Dotted path to the list when the records are wrapped, for example data.children. Leave blank when the response is the array.
Auth headers Header name/value pairs sent with every request, for example Authorization: Bearer <API_TOKEN>.

Auth header values are encrypted at rest and redacted from the audit trail — see Secrets and encryption.

Tip. Set the auth headers before you press Probe endpoint. The probe makes a real call; without the headers a protected API answers 401 and there is nothing to map in step 2.

Press Probe endpoint. Abrq DIP fetches one response, applies the records path, and shows the fields it found. Fix the URL, the path, or the headers and re-probe until the sample looks right.

Step 2 — Map data#

Choose which probed fields become columns, name each column, and give it a SQL type: TEXT, BIGINT, DOUBLE PRECISION, BOOLEAN, or JSONB.

Each column has a kind:

Kind What it stores
scalar One scalar field from each record. The default.
raw The whole record, unmodified, in one column. Pair with JSONB.
envelope A value taken from outside the records list — for example a page-level fetched_at — stamped onto every row of that fetch.
explode_value Unnests one array field: one output row per array element, holding the element's value.
explode_index The companion position column for the exploded array. Adds a <field>_idx column so you can rebuild the original order.

Only one array can be exploded per stream. Objects and lists that you do not explode are skipped — if you need them, map them as a raw or JSONB column instead of losing them.

Step 3 — Destination & schedule#

Fetch mode

Mode When to use it Required settings
Static The same URL every run returns what you want. —
Token / Cursor / Last-ID The API returns a cursor you echo back on the next call. response_path (where the cursor sits in the response), param_name (what to send it as), and whether it is sent as a query parameter or a header.
Offset You track the position yourself. offset_param, limit_param, page_size.

Follow pages within a single run makes one run keep paging until the API runs out of pages. The max pages cap bounds it — default 50, maximum 10000. Leave the toggle off and each scheduled run fetches exactly one page, advancing the cursor for the next run.

Policies

  • Bad-data policy — skip and count (rows that fail type conversion are rejected and counted, the run continues) or fail the run.
  • Schema-drift policy — Rescue (the default), Evolve, or Reject. See Schema drift policies.

Destination and schedule

  • Destination connector — Postgres or MySQL only.
  • Destination FQN — a schema-qualified name such as analytics.api_orders. The table is created automatically on the first run.
  • Schedule — a preset (Manual only, Every minute, Every 5 minutes, Every 30 minutes, Every 6 hours, Every 12 hours, Every 24 hours) or a 5-field cron expression. Cron is evaluated in UTC.

Run a stream#

Run now on the stream starts a run immediately. It offers Truncate destination before insert — use it when you are reloading a full snapshot and want the table replaced rather than appended to.

The Logs dialog streams a live run over SSE. Each run shows:

  • the exact URL and query parameters that were sent,
  • the cursor value before and after the fetch,
  • rows fetched and rows inserted,
  • rejected rows, when the bad-data policy is skip and count.

The cursor before/after pair is the fastest way to diagnose a stream that keeps re-reading page one: if after equals before, the response_path is not finding the cursor in the response.

Fan out to more destinations#

Open a stream and go to Destinations (/streams/:id/destinations). It has three tabs: Destinations, Topology, and Dead letters.

Every stream has a primary destination — the table you configured in the wizard. Fan-out adds more sinks that receive the same rows.

Destination type Target Notes
Database Postgres or MySQL Schema-qualified table FQN, same shape as the primary.
Kafka Topic name Optional partition-key field chosen from your mapped columns.
Object storage S3, MinIO, Azure, GCS Format jsonl or parquet.
FTP / FTPS Remote path Format jsonl, json, or csv.

Warning. The object-storage/file card is labelled FTP / SFTP, but SFTP is not implemented. Only plain FTP and explicit FTPS work. Point SFTP requirements at an object-storage destination instead.

Two rules apply when you add a destination:

  1. The connection test must pass before the destination can be saved.
  2. It is created disabled — the form says "Left disabled by default — enable after you've reviewed." Nothing is delivered until you enable it.

Delivery health#

The Delivery state column shows one of:

State Meaning
Healthy Deliveries are succeeding.
Recovering The circuit breaker is half-open: one trial delivery is being attempted.
Halted The breaker is open — deliveries are suspended.

The circuit breaker opens after 5 consecutive failures, waits a 300-second cooldown, then allows one half-open trial. A successful trial closes the breaker; a failed trial re-opens it and the cooldown restarts.

Reset breaker is an operator override: it forces the breaker closed without waiting for the cooldown. Use it after you have fixed the underlying problem, not as a retry loop.

Delivery is at-least-once with idempotency keys, so a downstream retry can produce a duplicate attempt but the key lets an idempotent sink drop it. A failing destination head-of-line blocks only itself — the other destinations, including the primary table, keep flowing.

Dead letters#

Records that could not be delivered land on the Dead letters tab. Two actions are available:

  • Replay — re-delivers the record, targeted at that one destination only. The confirmation warns: "Duplicates are possible if they partially landed before."
  • Discard — drops the record permanently. There is no undo.

Primary-destination rules#

The primary destination cannot be deleted, and its target table is managed by the stream itself — change it in the stream wizard, not on the destinations page.

Field reference#

Field Where Values / default
Name Step 1 Free text, required.
Endpoint URL Step 1 GET only in v1.
Records path Step 1 Dotted path, optional (e.g. data.children).
Auth headers Step 1 Name/value pairs; encrypted at rest, redacted from audit.
Column type Step 2 TEXT, BIGINT, DOUBLE PRECISION, BOOLEAN, JSONB.
Column kind Step 2 scalar, raw, envelope, explode_value, explode_index.
Fetch mode Step 3 Static, Token/Cursor/Last-ID, Offset.
response_path, param_name Step 3 Cursor modes; param sent as query or header.
offset_param, limit_param, page_size Step 3 Offset mode only.
Follow pages / max pages Step 3 Default 50, maximum 10000.
Bad-data policy Step 3 Skip and count (default) or fail the run.
Schema-drift policy Step 3 Rescue (default), Evolve, Reject.
Destination connector Step 3 Postgres or MySQL only.
Destination FQN Step 3 Schema-qualified; table auto-created.
Schedule Step 3 Presets from Manual only to 24 h, or 5-field UTC cron.

Failure modes#

What you see Cause Fix
Streams (API ingestion) isn't in your license The streams entitlement is absent. Contact Abrq to add it, then reinstall the license.
explode_field is not supported for Kafka destinations (no table schema to unnest into) — use a Postgres/MySQL destination An exploded column was mapped on a stream whose destination is a Kafka topic. Send the un-exploded records to Kafka and explode into a database destination.
kafka destination_fqn must be a topic name, not a schema-qualified identifier A value like analytics.orders was typed into a Kafka destination's target. Use the bare topic name.
cannot delete the only destination on a stream You tried to delete the primary destination. Delete the stream instead, or add another destination first.
Probe returns no fields The records path does not match the response shape. Re-check the dotted path; leave it blank if the response is already an array.
Runs succeed but insert 0 rows every time The cursor is not advancing — before and after are equal in the logs. Verify response_path points at the cursor value the API actually returns.
Destination shows Halted The breaker opened after 5 consecutive failures. Fix the sink, then Reset breaker — or wait out the 300 s cooldown for the half-open trial.