Tutorial: Log Data to SQL

Build a pipeline that reads tags on a schedule, computes a summary, and writes it to a SQL database — with store-and-forward so an unavailable database does not lose data.

You will need: tags with live data, and a SQL database you can write to. Time: ~30 minutes.

---

Step 1 — Create the connection

Connections → New → SQL database.

Field	Value
Name	ReportingDB
Driver	postgresql
Host / Port / Database	Your server
Username / Password	A dedicated account with rights only on the target table
Max connections	10

Test, and save.

→ SQL databases

---

Step 2 — Create the target table

CREATE TABLE production_log (
  id          BIGSERIAL PRIMARY KEY,
  ts          TIMESTAMPTZ NOT NULL,
  line        TEXT        NOT NULL,
  avg_speed   DOUBLE PRECISION,
  avg_current DOUBLE PRECISION,
  run_minutes DOUBLE PRECISION,
  UNIQUE (ts, line)
);

CREATE INDEX ON production_log (line, ts DESC);

The unique constraint on (ts, line) is deliberate: store-and-forward may replay a batch that was written but not acknowledged. With the constraint, a replay is rejected as a duplicate instead of double-counting your production figures.

---

Step 3 — Write the insert

The SQL node runs the statement you type into it. Values come from the incoming message through {{…}}, and each one is handed to the driver as a bound parameter rather than pasted into the text — so this is parameterised, not string-building.

INSERT INTO production_log (ts, line, avg_speed, avg_current, run_minutes)
VALUES ({{payload.ts}}, {{payload.line}}, {{payload.avg_speed}},
        {{payload.avg_current}}, {{payload.run_minutes}})
ON CONFLICT (ts, line) DO NOTHING

Keep the statement to hand — it goes into the node in step 6.

ON CONFLICT DO NOTHING makes the write idempotent, which is what you want under replay.

Not the same as a named query

Named queries use :name and belong to screen bindings in the Designer. The pipeline SQL node is a different thing: it takes a raw statement and fills {{…}} from the message.

---

Step 4 — Build the pipeline

Inject (cron 0 * * * *) → Industrial I/O (read) → Python (aggregate) → SQL Query (insert)
                                                        ↓
                                                     Debug

Inject

Setting	Value
Repeat	Cron 0 * * * * — top of every hour

Industrial I/O

Setting	Value
Operation	read
Tags	Site1/Line1/Filler/Motor1/Speed, …/Current, …/Running

---

Step 5 — Aggregate in Python

An instantaneous read is a snapshot, not an hour's average. Ask the historian instead:

import datetime

def transform(payload, message):
    now   = datetime.datetime.now(datetime.timezone.utc).replace(
                minute=0, second=0, microsecond=0)
    start = now - datetime.timedelta(hours=1)

    trends = system.hist.getTrends(
        ["Site1/Line1/Filler/Motor1/Speed",
         "Site1/Line1/Filler/Motor1/Current"],
        start.isoformat(), now.isoformat(),
        maxPoints=60, agg="avg",
    )

    running = system.tag.read("Site1/Line1/Filler/Motor1/Running")
    if running["quality"] != "Good":
        return None                       # do not log from untrustworthy data

    return {
        "ts":          start.isoformat(),
        "line":        "L1",
        "avg_speed":   _mean(trends, 0),
        "avg_current": _mean(trends, 1),
        "run_minutes": 60.0 if running["value"] else 0.0,
    }

Two habits again: check quality before acting, and return None to end the branch rather than writing a row you do not trust.

---

Step 6 — Write with the SQL node

Setting	Value
Connection	ReportingDB
Query	The INSERT from step 3, with its {{payload.…}} placeholders

The Python node upstream returns exactly the keys the placeholders name — ts, line, avg_speed, avg_current, run_minutes — which is what makes the two halves line up. Rename a key in the Python and the placeholder resolves to nothing.

Use the SQL node rather than system.db in Python. The node's writes are covered by store-and-forward: if the database is unreachable the write is buffered and replayed in order; if it fails permanently it is quarantined for you to inspect. A script write simply fails.

That is the whole reason this is a pipeline and not a script.

---

Step 7 — Test

1. Attach a Debug node to the Python output.
2. Execute once with the Run button and read the Debug panel.
3. Confirm the row landed:

SELECT * FROM production_log ORDER BY ts DESC LIMIT 5;

4. Start the pipeline and let it run for a couple of hours.

---

Step 8 — Verify the resilience

This is the step people skip, and it is the one worth doing.

1. Stop the database.
2. Let the pipeline fire once or twice.
3. Check the Status panel — the buffer is holding the writes.
4. Start the database.
5. Confirm the buffered rows drain, in order, with no duplicates.

If duplicates appear, the unique constraint from step 2 is missing.

---

Step 9 — Operate it

Watch	Where
Executions and failures	The execution log panel
Buffer depth and pressure	The Status panel
Quarantined writes	The Status panel's quarantine list

A quarantined batch usually means a schema mismatch, a constraint violation or a credential change. Fix the cause before retrying, or it will re-quarantine.

Remove the Debug node before this goes to production. A Debug node on a busy branch is a real cost.

---

Variations

Incremental extract — track a watermark in a variable so a restart resumes rather than reprocessing:

last = system.vars.get("prod_watermark", "1970-01-01T00:00:00Z")
# ... query changes since `last` ...
system.vars.set("prod_watermark", newest, "persistent")

Event-driven instead of scheduled — replace Inject with an Industrial I/O subscribe node so a batch-complete signal triggers the write.

High-volume time-series — write to QuestDB with the QuestDB node in ingress mode instead of SQL.

Transactional multi-table write — move it to a Gateway script so all tables commit or none do.

Next

→ Publish a REST endpoint · Pipelines