Singer, Meltano & the Tap/Target Spec: Building Open-Source ELT Connectors

Singer taps are the extract half of an open-source contract that decides whether moving a new SaaS API or a Postgres table into your warehouse is a two-line config change or a two-week bespoke integration — and it is the layer that most data engineers reach for the moment a managed connector doesn't exist, prices badly, or can't run inside their VPC. The whole Singer idea is deliberately small: a tap reads from a source and prints a stream of newline-delimited JSON messages to stdout, a target reads those messages from stdin and writes them to a destination, and the two processes are joined by nothing more exotic than a Unix pipe. Because the tap/target spec is just a message contract, any tap composes with any target — tap-github | target-snowflake, tap-postgres | target-jsonl, tap-stripe | target-bigquery — and you never rewrite the loader when you add a source.

This guide is the senior-data-engineering walkthrough for building and reasoning about open-source ELT connectors the way an interviewer probes them: what the message protocol actually is (SCHEMA, RECORD, STATE, ACTIVATE_VERSION), how a tap introspects a source into a Singer catalog during discovery, how Singer state bookmarks a replication key so the next run resumes instead of re-reading the whole table, and how Meltano — the batteries-included runner — turns a pile of taps and targets into a declarative meltano.yml pipeline with managed state, plugin installs, stream maps, and an SCD Type 2 load in the target. Each section pairs a teaching block with a Solution-Tail interview answer: code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse the reshaping reps on the data-transformation practice library →, and sharpen the message-parsing muscle on the JSON practice library →.

---

On this page

• Why the Singer spec decides your open-source ELT strategy
• The tap/target message protocol
• Singer catalog and discovery
• Singer state and incremental replication
• Meltano orchestration, targets, and SCD
• Cheat sheet — Singer & Meltano connector recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why the Singer spec decides your open-source ELT strategy

One tiny message contract, and every tap composes with every target — the choice binds your whole ingestion layer

The one-sentence invariant: Singer is a specification for a tap (a source-reader that prints newline-delimited JSON messages to stdout) and a target (a destination-writer that reads those messages from stdin), joined by a Unix pipe, so any conformant tap works with any conformant target without either side knowing the other exists — and the moment you adopt that contract, adding a new source is "write or install a tap," not "rewrite the loader." The reason this matters at the architecture level is that it decouples the O(sources) problem from the O(destinations) problem: instead of writing sources × destinations bespoke integrations, you write sources taps and destinations targets and let the pipe do the fan-in. The spec you adopt in month one becomes the interface every future connector must honour, which is exactly why interviewers open here.

The four axes interviewers actually probe.

• Spec conformance. Does the tap speak the protocol correctly — SCHEMA before any RECORD for a stream, STATE messages that are safe resume points, messages on stdout and logs on stderr? A tap that prints a log line to stdout corrupts the stream for the target. This is the first thing a reviewer checks, and the first thing that breaks in a hand-rolled tap.
• Discovery / catalog. Can the tap introspect its source (tap --config config.json --discover) and emit a catalog of streams, each with a JSON schema, key_properties, and metadata that the operator edits to select streams and choose a replication method? Discovery is what separates a real connector from a hard-coded script.
• State / incremental. Does the tap emit STATE messages carrying a bookmark (a replication-key high-watermark) so the next run resumes from where it stopped instead of re-reading the whole source? Interviewers probe this because incrementality is the difference between a 30-second nightly run and an eight-hour full refresh.
• Extraction mode. FULL_TABLE (re-read everything, version with ACTIVATE_VERSION), INCREMENTAL (replication-key bookmark), or LOG_BASED (tail the WAL/binlog). Each has a different delete-handling and source-load story — the same trade-offs a change-data-capture design forces, expressed inside the Singer catalog.

The 2026 reality — Singer is the open-source lingua franca, Meltano is the runner.

• Singer taps and targets are the portable, self-contained building blocks. The Meltano Hub and the older Singer.io index list hundreds of community taps (tap-postgres, tap-github, tap-stripe, tap-salesforce) and targets (target-jsonl, target-postgres, target-snowflake, target-bigquery). Each is a normal Python package with a CLI.
• Meltano is the batteries-included orchestrator: a meltano.yml declares extractors (taps) and loaders (targets), Meltano installs them into isolated virtualenvs, manages configuration and secrets, persists state in a system database, and runs the pipe for you with meltano run tap-x target-y.
• The SDK (the Meltano Singer SDK, formerly singer-sdk) is how you build a conformant tap or target in 2026 without hand-writing the message loop — it gives you discovery, state, and pagination scaffolding, and it is what most new taps are written against.
• The managed alternatives — Airbyte, Fivetran, Stitch (which originated Singer) — trade the do-it-yourself control of Singer/Meltano for a hosted control plane. The senior answer names when the open-source path wins: VPC-only sources, a connector that doesn't exist, cost at high row volume, or a need to version connectors in your own repo.

What interviewers listen for.

• Do you describe Singer as "a message contract, tap on stdout, target on stdin, joined by a pipe" rather than "an ETL tool"? — required framing.
• Do you name the five message types (SCHEMA, RECORD, STATE, ACTIVATE_VERSION, and BATCH) without prompting? — senior signal.
• Do you separate Singer (the spec) from Meltano (the runner) from the SDK (the build framework)? — senior signal.
• Do you tie STATE to resumability — "the last STATE line is the next run's --state" — instead of hand-waving "it's incremental"? — required answer.
• Do you pick open-source over managed for a concrete reason (VPC, missing connector, cost, versioning) rather than dogma? — senior signal.

Worked example — the four-axis connector evaluation grid

Detailed explanation. The single most useful artifact for a Singer interview is a grid that scores a candidate connector on the four axes before you commit to it. Every "should we build a tap or buy a connector?" discussion converges on this grid; having it in your head turns a vague answer into a decision. Walk through scoring a hypothetical need: replicate a Postgres orders table and a REST tickets API into a warehouse.

• The two sources. public.orders on Postgres 16 (has an updated_at column) and a paginated REST endpoint GET /tickets?updated_since= returning JSON.
• The destination. A columnar warehouse; a local target-jsonl for a first smoke test.
• The question per axis. Is there a conformant tap? Does it support discovery? Does it bookmark incrementally? Which extraction modes does it offer?

Question. Build the four-axis evaluation for both sources and decide build-vs-reuse for each.

Input.

Axis	tap-postgres (orders)	tap-tickets (REST API)
Spec conformance	mature community tap	must write with the SDK
Discovery / catalog	yes — introspects information_schema	yes — schema hard-coded or inferred
State / incremental	INCREMENTAL on updated_at; LOG_BASED via WAL	INCREMENTAL on updated_since cursor
Extraction mode	FULL_TABLE / INCREMENTAL / LOG_BASED	FULL_TABLE / INCREMENTAL

Code.

# Install a Singer tap + target the standard way (isolated virtualenvs)
python -m venv .venv && source .venv/bin/activate
pip install tap-postgres target-jsonl

# 1. Discover the source's streams into a catalog
tap-postgres --config tap_postgres_config.json --discover > catalog.json

# 2. Smoke-test the pipe: extract selected streams, load to local JSONL files
tap-postgres --config tap_postgres_config.json --catalog catalog.json \
  | target-jsonl --config target_jsonl_config.json \
  > state.json

# 3. The last line written to state.json is the next run's bookmark
tail -n 1 state.json
# {"bookmarks": {"public-orders": {"replication_key": "updated_at",
#                                  "replication_key_value": "2026-08-18T09:15:22+00:00"}}}

Step-by-step explanation.

1. tap-postgres and target-jsonl install as ordinary Python CLIs. The convention is one isolated environment per plugin so their dependency trees never collide — Meltano automates this later, but the primitives are just pip install + a CLI.
2. --discover makes the tap introspect the source (here, Postgres information_schema) and print a catalog describing every stream it can produce. You commit and edit that catalog to select streams and set replication methods — nothing runs yet.
3. The pipe tap ... | target ... is the whole runtime. The tap writes SCHEMA/RECORD/STATE messages to stdout; the target reads them from stdin and writes JSONL files. Redirecting the target's stdout to state.json captures the emitted STATE.
4. The final STATE line is the bookmark. Feeding it back as --state state.json on the next run makes the tap emit only rows with updated_at greater than the bookmark — the extraction is now incremental with no code change, just a state file.
5. For the REST tickets source there is no mature tap, so the decision is build with the SDK: subclass Stream, define the schema, implement pagination and an incremental replication_key. The grid tells you Postgres is reuse and tickets is build — before you write a line of production code.

Output.

Source	Verdict	Reason
orders (Postgres)	reuse tap-postgres	mature tap; INCREMENTAL + LOG_BASED both available
tickets (REST API)	build with the SDK	no existing tap; INCREMENTAL on the updated_since cursor
Both	one target	target-jsonl locally, warehouse target in prod — unchanged by source choice

Rule of thumb. Score every prospective connector on the four axes — conformance, discovery, state, extraction mode — before deciding build-vs-reuse. Reuse a mature tap when one exists; build with the SDK when it doesn't. The target is chosen once and never rewritten when you add a source.

Worked example — Singer vs Meltano vs the SDK (who does what)

Detailed explanation. The most common muddle in a Singer interview is conflating the spec, the runner, and the build framework into one word ("Singer"). Naming the three layers cleanly is a senior signal because it shows you know where each responsibility lives. Walk through the separation with a single concrete pipeline.

• Singer (the spec). Defines the message types and the CLI contract (--config, --discover, --catalog, --state). It owns nothing at runtime — it is a document plus a JSON schema.
• The tap/target processes. Concrete executables that implement the spec. They know how to read one source or write one destination.
• Meltano (the runner). Installs plugins, stores config and secrets, persists state, resolves environments, and runs the pipe. It owns orchestration.
• The Singer SDK. A Python framework for authoring conformant taps/targets so you don't hand-write the message loop, discovery, or state plumbing.

Question. Map each responsibility — discovery, config, the pipe, state persistence, scheduling — to the layer that owns it.

Input.

Responsibility	Singer spec	tap/target process	Meltano	SDK
Message format	defines	emits/consumes	passes through	implements
Discovery	defines --discover	runs it	invokes it	scaffolds it
State persistence	defines STATE msg	emits STATE	stores it (system DB)	manages bookmarks
Plugin install / venv	—	—	owns	—
Scheduling / env	—	—	owns	—

Code.

# meltano.yml — the runner ties the three layers together declaratively
version: 1
default_environment: dev
project_id: singer-demo

plugins:
  extractors:
    - name: tap-postgres          # a Singer tap (implements the spec, via the SDK)
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config:
        host: db-primary.internal
        database: production
      select:
        - public-orders.*         # stream selection lives here, not in code
  loaders:
    - name: target-jsonl          # a Singer target
      variant: andyh1203
      pip_url: target-jsonl

# Meltano (the runner) invokes discovery, builds the pipe, persists state
meltano install                        # create isolated venvs for each plugin
meltano invoke tap-postgres --discover # runner calls the spec's --discover
meltano run tap-postgres target-jsonl  # runner builds tap | target AND stores STATE

Step-by-step explanation.

1. The spec contributes only definitions: what a SCHEMA message looks like, what --discover must print, what --state accepts. Nothing in the spec runs; it is the interface every other layer honours.
2. The tap/target processes are the concrete implementations. tap-postgres knows Postgres; target-jsonl knows how to write JSONL. Neither knows the other — they only agree on the message format from the spec.
3. Meltano owns everything operational: meltano install creates a virtualenv per plugin (so tap-postgres and target-snowflake never fight over dependency versions), meltano run constructs the tap | target pipe, and — critically — it captures the emitted STATE and stores it in its system database, so you never manually shuttle state.json around.
4. The SDK is the authoring layer. When you build tap-tickets, you subclass the SDK's Stream and Tap classes; the SDK generates the discovery catalog, emits correctly-ordered messages, and handles the bookmark arithmetic — you write the source-specific parts only.
5. The payoff of the separation: you can swap the runner (run the raw pipe by hand for a smoke test, or Meltano in prod) without touching the tap, and you can swap the tap without touching the target. Each layer has one job.

Output.

Layer	Owns	You touch it when
Singer spec	the message + CLI contract	never (it's a document)
tap / target	one source / one destination	adding a source or destination
Meltano	install, config, state, scheduling	wiring a pipeline / operating it
Singer SDK	authoring scaffolding	building a new tap or target

Rule of thumb. Say "Singer is the spec, the tap/target are the processes, Meltano is the runner, the SDK is how you build one" — four layers, four jobs. Conflating them is the tell of someone who has read about Singer but never shipped a connector.

Worked example — when open-source ELT beats a managed connector

Detailed explanation. The senior version of "should we use Singer/Meltano or Fivetran?" is not a religious answer — it is a decision driven by four concrete constraints. Codifying them makes your interview answer reproducible and defensible. Walk through the four constraints with the two sources from earlier.

• Existence. Does a managed connector for this source exist at all? Long-tail internal APIs usually have none.
• Network. Can data leave your VPC? Some sources are only reachable inside a private network where a hosted SaaS connector cannot run.
• Cost. Managed connectors price on rows or MAR (monthly active rows). At high volume the open-source path can be an order of magnitude cheaper.
• Control / versioning. Do you need the connector pinned in your own repo, patchable, and reviewable? Open-source taps live in your codebase; managed connectors are a vendor black box.

Question. Decide open-source vs managed for the orders and tickets sources given a VPC-only Postgres and a niche ticketing API.

Input.

Constraint	orders (VPC Postgres)	tickets (niche API)
Managed connector exists?	yes (generic Postgres)	no
Reachable outside VPC?	no	yes
Volume / cost pressure	high (100M+ rows)	low
Must pin/patch in repo?	yes (compliance)	yes

Code.

# A tiny decision helper (illustrative)
def choose_ingestion(has_managed: bool,
                     reachable_outside_vpc: bool,
                     high_volume: bool,
                     needs_repo_control: bool) -> str:
    """Return 'open-source (Singer/Meltano)' or 'managed'."""
    if not has_managed:
        return "open-source (Singer/Meltano)"      # nothing to buy
    if not reachable_outside_vpc:
        return "open-source (Singer/Meltano)"      # SaaS connector can't reach it
    if high_volume or needs_repo_control:
        return "open-source (Singer/Meltano)"      # cost / versioning
    return "managed"                               # otherwise buy convenience


print(choose_ingestion(True,  False, True,  True))   # orders
# → open-source (Singer/Meltano)
print(choose_ingestion(False, True,  False, True))   # tickets
# → open-source (Singer/Meltano)

Step-by-step explanation.

1. The orders source has a managed Postgres connector, but it lives inside a VPC the SaaS control plane cannot reach — the network constraint alone forces the open-source path, where the tap runs inside the VPC next to the database.
2. Even setting network aside, orders is high-volume (100M+ rows) and under a compliance rule that the connector be pinned and reviewable — either constraint independently points to open-source.
3. The tickets source has no managed connector at all, so existence decides it: you build a tap with the SDK. There is nothing to buy.
4. The helper encodes the precedence: no managed connector or an unreachable source is an immediate open-source verdict; otherwise high volume or a versioning requirement tips it; only a low-volume, reachable, uncontrolled source justifies paying for managed convenience.
5. The honest senior caveat: open-source ELT trades money for engineering time. You now own the tap's bugs, its schema drift, and its on-call. The decision is real, not free — but for VPC-locked, high-volume, or nonexistent-connector cases, it is the right one.

Output.

Source	Verdict	Deciding constraint
orders (VPC Postgres)	open-source	network (VPC) + volume + control
tickets (niche API)	open-source	no managed connector exists
(hypothetical) low-volume public SaaS	managed	none of the four constraints bind

Rule of thumb. Reach for Singer/Meltano when the connector doesn't exist, the source is VPC-locked, the volume makes managed pricing hurt, or compliance needs the connector in your repo. Otherwise, buying managed convenience is a legitimate choice — name the constraint, don't preach.

Senior interview question on open-source ELT strategy

A senior interviewer often opens with: "You're standing up ingestion for a new warehouse. You have a VPC-locked Postgres, three niche internal REST APIs with no managed connectors, and a mandate to keep connectors reviewable in your own repo. Walk me through why you'd choose Singer/Meltano, how you'd structure the project, which pieces you'd reuse versus build, and how you'd keep a single source's failure from taking down the others."

Solution Using a Meltano project with reused taps, SDK-built taps, and isolated runs

# meltano.yml — one project, many extractors, isolated loaders
version: 1
default_environment: prod
environments:
  - name: prod
  - name: dev

plugins:
  extractors:
    - name: tap-postgres              # REUSE — mature community tap
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config:
        host: db-primary.internal      # runs INSIDE the VPC, next to Postgres
        database: production
      metadata:
        public-orders:
          replication-method: INCREMENTAL
          replication-key: updated_at
      select:
        - public-orders.*
        - public-customers.*

    - name: tap-tickets               # BUILD — SDK tap, pinned to our git repo
      pip_url: git+https://git.internal/data/tap-tickets.git@v0.4.1
      config:
        api_base: https://tickets.internal/api
      metadata:
        tickets:
          replication-method: INCREMENTAL
          replication-key: updated_at

  loaders:
    - name: target-snowflake
      variant: meltanolabs
      pip_url: meltanolabs-target-snowflake

# Run each source as an isolated invocation so one failure is contained
set -euo pipefail
for tap in tap-postgres tap-tickets tap-billing; do
  meltano run "$tap" target-snowflake || echo "FAILED: $tap" >> run_failures.log
done
# Each `meltano run` is its own tap|target pipe with its own state entry.

Step-by-step trace.

Concern	Choice	Reasoning
VPC-locked Postgres	tap runs inside the VPC	no data leaves the private network
Niche APIs (no connector)	SDK-built taps, git-pinned	reviewable, patchable, in our repo
Dependency conflicts	one venv per plugin (Meltano)	tap-postgres and target never clash
State	Meltano system DB per (tap, target)	each source resumes independently
Blast radius	one meltano run per source	one tap's failure doesn't stop the rest
Reuse vs build	reuse tap-postgres, build tap-tickets	build only what doesn't exist

After the project is wired, each source is an independent meltano run tap-X target-snowflake invocation with its own state row; a schema-drift crash in tap-tickets logs a failure and the loop moves on to tap-billing, so one flaky API never blocks the Postgres feed. Reused taps carry zero maintenance; the two SDK-built taps are versioned in the internal git repo and reviewed like any other code.

Output:

Metric	Managed-everything	Singer / Meltano
VPC-locked source support	not possible	native (tap runs in-VPC)
Connectors that don't exist	blocked	build with the SDK
Connector reviewability	vendor black box	pinned in git, code-reviewed
Per-source isolation	vendor-controlled	one meltano run each
Ongoing cost	per-row/MAR pricing	compute + engineering time

Why this works — concept by concept:

• Tap/target contract — because a tap only agrees with a target on the message format, every source is independent. Reused taps and SDK-built taps emit the same SCHEMA/RECORD/STATE stream, so target-snowflake loads all of them unchanged.
• Meltano-managed virtualenvs — one isolated environment per plugin means tap-postgres's dependencies never collide with the target's. This is the operational reason Singer connectors compose at all in one project.
• Per-source state rows — Meltano stores a separate bookmark per (extractor, loader) pair in its system database, so each source resumes from its own high-watermark and a reset on one doesn't touch the others.
• Isolated runs for blast-radius control — running each source as its own meltano run (rather than one mega-pipeline) means a schema-drift crash in one tap is caught, logged, and stepped over — the Postgres feed still lands.
• Cost — you trade vendor per-row pricing for compute plus the engineering time to own two SDK taps. At VPC-locked, high-volume, no-connector-exists scale that trade is strongly favourable; the honest cost is the on-call for the taps you now maintain.

ETL
Topic — etl
ETL problems on open-source ingestion pipelines

Practice →

Design Topic — design Design problems on connector architecture

Practice →

---

2. The tap/target message protocol

SCHEMA, RECORD, STATE, ACTIVATE_VERSION — newline-delimited JSON on stdout, logs on stderr

The mental model in one line: the Singer tap/target spec is a stream of newline-delimited JSON objects, each with a type field — a tap prints SCHEMA (the shape of a stream), RECORD (one row), STATE (a resumable bookmark), and optionally ACTIVATE_VERSION (a table-version marker) to stdout, a target reads them from stdin, and the single hardest-and-most-important rule is that only these messages go to stdout while every log line, warning, and metric goes to stderr. Break that one rule — print a log to stdout — and the target tries to parse your log line as a message and the whole pipe fails. Every hand-written tap gets this wrong once.

The five message types.

• SCHEMA. Declares a stream and its JSON schema: {"type": "SCHEMA", "stream": "users", "schema": {...}, "key_properties": ["id"], "bookmark_properties": ["updated_at"]}. A stream's SCHEMA must be emitted before any RECORD for that stream. Targets use it to create/evolve the destination table.
• RECORD. One row of data: {"type": "RECORD", "stream": "users", "record": {...}, "time_extracted": "..."}. The record must validate against the most recent SCHEMA for that stream.
• STATE. A resumable bookmark: {"type": "STATE", "value": {...}}. The tap emits it periodically; the last STATE the target durably persisted is the next run's --state. STATE is opaque to the target except that it echoes it downstream once records are safely written.
• ACTIVATE_VERSION. A table-version marker used by FULL_TABLE syncs to implement atomic "swap in the new snapshot" semantics: {"type": "ACTIVATE_VERSION", "stream": "users", "version": 1692300000000}.
• BATCH. An optional bulk-transfer message (paths to serialized record files) for high-throughput fast-sync; most connectors never emit it, but naming it completes the set.

The ordering contract — what conformance actually requires.

• SCHEMA first. For each stream, exactly one SCHEMA precedes its RECORDs. Re-emitting SCHEMA mid-stream is legal (schema evolution) and re-declares the shape from that point on.
• STATE is a checkpoint, not a per-record event. Emit it after a meaningful chunk of records, and only for progress you are willing to lose-and-resume. A STATE says "everything before this bookmark is safely upstream of me."
• The target echoes STATE. A well-behaved target writes STATE to its stdout only after the records preceding it are durably written to the destination. That is what makes the bookmark safe — it advances only behind flushed data.
• stdout is sacred. Messages only. print()-debugging into stdout is the classic corruption bug; use the logging framework, which writes to stderr.

The tap CLI contract.

• --config config.json. Connection + tuning parameters (host, token, start_date, page size).
• --discover. Print the catalog and exit; do not extract.
• --catalog catalog.json (a.k.a. --properties in older taps). The edited catalog telling the tap which streams to run and how.
• --state state.json. The bookmark from the previous run; the tap resumes from it.

Common interview probes on the protocol.

• "Where do logs go?" — required answer: stderr; stdout is messages only.
• "What must precede the first RECORD?" — the stream's SCHEMA.
• "When is it safe to advance STATE?" — only after the preceding records are durably written by the target.
• "How do a tap and target communicate?" — a Unix pipe: tap | target; JSON lines, no shared library.

Worked example — a minimal hand-written tap

Detailed explanation. To internalise the protocol, write a tap by hand — no SDK — that emits one stream. It prints a SCHEMA, three RECORDs, and a closing STATE, all to stdout, and logs to stderr. This is the smallest thing that is genuinely a Singer tap. Walk through it line by line.

• Stream. users with id, name, updated_at.
• Messages. one SCHEMA → three RECORD → one STATE.
• Discipline. JSON to stdout via a single writer; logs to stderr.

Question. Write a standalone tap that emits a conformant users stream and a resumable bookmark on updated_at.

Input.

Field	Type	Role
id	integer	key_properties
name	string	data
updated_at	string (date-time)	replication key / bookmark

Code.

#!/usr/bin/env python3
"""A minimal hand-written Singer tap for a `users` stream."""
import json
import sys
import logging

# Logs go to STDERR — never stdout.
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("tap-users")

def write(msg: dict) -> None:
    """Every Singer message is one JSON object + newline on STDOUT."""
    sys.stdout.write(json.dumps(msg) + "\n")
    sys.stdout.flush()

USERS = [
    {"id": 1, "name": "Ada",   "updated_at": "2026-08-18T09:00:00+00:00"},
    {"id": 2, "name": "Linus", "updated_at": "2026-08-18T09:05:00+00:00"},
    {"id": 3, "name": "Grace", "updated_at": "2026-08-18T09:10:00+00:00"},
]

def main() -> None:
    # 1. SCHEMA — must come before any RECORD for this stream
    write({
        "type": "SCHEMA",
        "stream": "users",
        "schema": {
            "type": "object",
            "properties": {
                "id":         {"type": "integer"},
                "name":       {"type": "string"},
                "updated_at": {"type": "string", "format": "date-time"},
            },
        },
        "key_properties": ["id"],
        "bookmark_properties": ["updated_at"],
    })

    # 2. RECORDs — each validates against the SCHEMA above
    max_bookmark = None
    for row in USERS:
        write({"type": "RECORD", "stream": "users", "record": row})
        max_bookmark = row["updated_at"]   # rows arrive sorted by updated_at
        log.info("emitted user id=%s", row["id"])   # → stderr, safe

    # 3. STATE — the resumable bookmark, emitted after the records
    write({
        "type": "STATE",
        "value": {"bookmarks": {"users": {
            "replication_key": "updated_at",
            "replication_key_value": max_bookmark,
        }}},
    })

if __name__ == "__main__":
    main()

Step-by-step explanation.

1. The write() helper is the only thing that touches stdout, and it does exactly one job: serialize a dict to a single JSON line and flush. Centralising stdout access in one function is how you guarantee nothing else leaks into the message stream.
2. logging.basicConfig(stream=sys.stderr, ...) sends every log to stderr. This is the non-negotiable discipline — if log.info had written to stdout, the downstream target would try to parse "emitted user id=1" as a message and abort.
3. The SCHEMA message is emitted first, declaring the stream shape, key_properties (the primary key the target uses for upserts), and bookmark_properties (which field the incremental logic tracks). No RECORD may precede it.
4. Each RECORD carries one row under record, and it must conform to the schema — the target validates and uses the schema to create or evolve the destination table. As rows stream by (sorted by updated_at), we track the maximum as the bookmark-to-be.
5. The closing STATE publishes replication_key_value = max(updated_at seen). Feeding this state back on the next run is what makes the tap incremental — the resume point is data, not code.

Output.

{"type": "SCHEMA", "stream": "users", "schema": {...}, "key_properties": ["id"], "bookmark_properties": ["updated_at"]}
{"type": "RECORD", "stream": "users", "record": {"id": 1, "name": "Ada",   "updated_at": "2026-08-18T09:00:00+00:00"}}
{"type": "RECORD", "stream": "users", "record": {"id": 2, "name": "Linus", "updated_at": "2026-08-18T09:05:00+00:00"}}
{"type": "RECORD", "stream": "users", "record": {"id": 3, "name": "Grace", "updated_at": "2026-08-18T09:10:00+00:00"}}
{"type": "STATE", "value": {"bookmarks": {"users": {"replication_key": "updated_at", "replication_key_value": "2026-08-18T09:10:00+00:00"}}}}

Rule of thumb. Funnel all stdout writes through one write() helper, send every log to stderr, emit SCHEMA before any RECORD, and close with a STATE whose bookmark is the max replication-key you actually emitted. That five-line discipline is 90% of protocol conformance.

Worked example — a minimal target that consumes the stream

Detailed explanation. The other half of the pipe: a target reads messages from stdin, creates a file-per-stream on the first SCHEMA, appends RECORDs, and — crucially — only echoes a STATE to its own stdout after the preceding records are flushed to disk. This is the target-jsonl shape in miniature. Walk through it.

• Input. messages on stdin.
• Behaviour. open a <stream>.jsonl on SCHEMA; append record on RECORD; flush + echo STATE.
• Safety. advance STATE only behind flushed data.

Question. Write a target that writes each stream to a JSONL file and echoes STATE only after a durable flush.

Input.

Message in	Target action
SCHEMA	open/reset <stream>.jsonl; remember key_properties
RECORD	append record as one JSON line
STATE	fsync open files, then echo STATE to stdout
ACTIVATE_VERSION	(jsonl target) no-op / rotate file

Code.

#!/usr/bin/env python3
"""A minimal Singer target: newline-JSON in, one JSONL file per stream out."""
import json
import sys
import logging

logging.basicConfig(stream=sys.stderr, level=logging.INFO)
log = logging.getLogger("target-jsonl-min")

def main() -> None:
    files: dict[str, "TextIO"] = {}      # stream -> open file handle
    last_state: str | None = None

    for line in sys.stdin:               # read the pipe line by line
        line = line.strip()
        if not line:
            continue
        msg = json.loads(line)
        mtype = msg["type"]

        if mtype == "SCHEMA":
            stream = msg["stream"]
            files[stream] = open(f"{stream}.jsonl", "w")   # create/reset
            log.info("opened %s.jsonl", stream)

        elif mtype == "RECORD":
            stream = msg["stream"]
            files[stream].write(json.dumps(msg["record"]) + "\n")

        elif mtype == "STATE":
            # Flush every open file BEFORE acknowledging the bookmark,
            # so STATE never advances ahead of durably-written data.
            for f in files.values():
                f.flush()
            last_state = json.dumps(msg["value"])
            sys.stdout.write(last_state + "\n")   # echo STATE downstream
            sys.stdout.flush()

        elif mtype == "ACTIVATE_VERSION":
            pass   # jsonl target: nothing to activate

    for f in files.values():
        f.close()

if __name__ == "__main__":
    main()

Step-by-step explanation.

1. The target's entire runtime is a loop over stdin lines, each parsed as one JSON message. This mirrors the tap: the two processes never share code, only the newline-JSON contract.
2. On SCHEMA, the target opens (and resets) a <stream>.jsonl file and remembers the stream exists. A real warehouse target would CREATE TABLE or evolve the schema here instead of opening a file.
3. On RECORD, it appends the inner record object as one JSON line. Nothing fancy — the schema was already validated by convention upstream, and target-jsonl is deliberately a thin sink.
4. On STATE, it flushes every open file first, then echoes the state to its own stdout. This ordering is the correctness heart of the whole spec: STATE is only acknowledged after the data it covers is durable, so a crash-and-resume never skips unwritten rows.
5. ACTIVATE_VERSION is a no-op for JSONL (there is no atomic swap for a plain file), but a warehouse target would use it to promote a freshly-loaded table version. The echoed STATE is what a runner like Meltano captures and stores.

Output.

File	Content after run
users.jsonl	3 lines, one JSON object per user
target stdout	the echoed STATE line (captured by the runner)
target stderr	opened users.jsonl log line

Rule of thumb. A target must flush its destination writes before echoing a STATE. Advance the bookmark only behind durably-written data, and the "at-least-once, resume-safe" guarantee falls out for free. STATE ahead of unflushed data is how you silently lose rows on a crash.

Worked example — the three protocol bugs that corrupt a pipe

Detailed explanation. Three mistakes account for almost every "my hand-rolled tap doesn't work" ticket. Each violates a specific clause of the contract, and each produces a distinctive failure. Walk through all three with the fix.

• Bug 1 — logging to stdout. A stray print() writes a non-JSON line into the message stream.
• Bug 2 — RECORD before SCHEMA. The target receives a row for a stream it has no schema for.
• Bug 3 — STATE emitted before records are safe. The bookmark advances ahead of data; a crash loses rows.

Question. Diagnose each symptom and state the one-line fix.

Input.

Symptom	Root cause	Clause violated
json.decoder.JSONDecodeError in target	print() to stdout	stdout = messages only
target errors "no schema for stream X"	RECORD before SCHEMA	SCHEMA-before-RECORD
rows missing after a mid-run crash	STATE ahead of flush	STATE only behind durable data

Code.

# WRONG — three classic bugs
print("starting sync")                       # BUG 1: pollutes stdout
write({"type": "RECORD", "stream": "u", ...})# BUG 2: no SCHEMA emitted yet
write({"type": "STATE",  "value": bookmark}) # BUG 3: emitted before RECORDs
for row in rows:
    write({"type": "RECORD", "stream": "u", "record": row})

# RIGHT — conformant ordering, logs on stderr
log.info("starting sync")                    # FIX 1: stderr via logging
write({"type": "SCHEMA", "stream": "u", ...})# FIX 2: SCHEMA first
for row in rows:
    write({"type": "RECORD", "stream": "u", "record": row})
    max_bm = row["updated_at"]
write({"type": "STATE", "value": {"bookmarks":
       {"u": {"replication_key_value": max_bm}}}})  # FIX 3: STATE after records

Step-by-step explanation.

1. Bug 1 surfaces as a JSONDecodeError in the target, not the tap — the tap runs fine, but its stray print("starting sync") is a non-JSON line the target chokes on. The fix is to route the message through logging (which is configured to stderr).
2. Bug 2 makes the target reject a RECORD for a stream it has never seen a SCHEMA for; well-behaved targets raise "no schema for stream u." Emitting the SCHEMA first — exactly once before the records — resolves it.
3. Bug 3 is the silent, dangerous one: emitting STATE before the records means the bookmark now claims progress the target hasn't received. If the process crashes, the next run resumes past rows that were never written. The fix is ordering: records first, STATE last, and (on the target side) flush before echoing.
4. The corrected block reads top-to-bottom as the contract itself: log to stderr, SCHEMA, then RECORDs while tracking the max bookmark, then a single STATE reflecting only what was emitted.
5. These three are worth memorising because they are exactly what an interviewer asks you to spot in a code sample. Naming the clause each one violates — not just "it's broken" — is the senior signal.

Output.

Bug	Failure mode	Fix
log on stdout	target JSONDecodeError	log to stderr
RECORD before SCHEMA	"no schema for stream"	emit SCHEMA first
STATE before flush	rows lost on crash	STATE last; flush before echo

Rule of thumb. Memorise the three: logs to stderr, SCHEMA before RECORD, STATE only behind flushed data. Every protocol bug you will ever debug (or be asked to spot in an interview) is one of these three clauses being violated.

Senior interview question on the Singer message protocol

A senior interviewer might ask: "I'll hand you a hand-written tap that a junior shipped. It occasionally corrupts the target and sometimes loses rows after a restart. Walk me through exactly what the protocol requires — message types, ordering, the stdout/stderr split, and the STATE-safety rule — and show me the corrected message loop with a proper incremental bookmark."

Solution Using a disciplined message loop with stderr logging and safe STATE

#!/usr/bin/env python3
"""Conformant Singer tap message loop with a safe incremental bookmark."""
import json, sys, logging
from datetime import datetime, timezone

logging.basicConfig(stream=sys.stderr, level=logging.INFO)   # logs -> stderr
log = logging.getLogger("tap-orders")

def write(msg: dict) -> None:
    sys.stdout.write(json.dumps(msg, default=str) + "\n")
    sys.stdout.flush()

def sync_orders(config: dict, state: dict) -> None:
    stream = "orders"
    # Resume point: the bookmark from the previous run (or the config start_date)
    bookmark = (state.get("bookmarks", {})
                     .get(stream, {})
                     .get("replication_key_value", config["start_date"]))

    # 1. SCHEMA before any RECORD
    write({"type": "SCHEMA", "stream": stream,
           "schema": ORDERS_SCHEMA,
           "key_properties": ["id"],
           "bookmark_properties": ["updated_at"]})

    # 2. Pull only rows newer than the bookmark, in replication-key order
    rows = fetch_orders_since(config, bookmark)   # ORDER BY updated_at ASC
    max_bm = bookmark
    emitted = 0
    for i, row in enumerate(rows, start=1):
        write({"type": "RECORD", "stream": stream, "record": row,
               "time_extracted": datetime.now(timezone.utc).isoformat()})
        max_bm = row["updated_at"]
        emitted += 1
        # 3. Periodic STATE — checkpoint progress every 10k rows
        if i % 10_000 == 0:
            write({"type": "STATE", "value": {"bookmarks":
                   {stream: {"replication_key": "updated_at",
                             "replication_key_value": max_bm}}}})
            log.info("checkpoint at %s (%d rows)", max_bm, i)

    # 4. Final STATE reflects the max replication-key actually emitted
    write({"type": "STATE", "value": {"bookmarks":
           {stream: {"replication_key": "updated_at",
                     "replication_key_value": max_bm}}}})
    log.info("sync complete: %d rows, bookmark -> %s", emitted, max_bm)

def main() -> None:
    config = json.load(open(_arg("--config")))
    state  = json.load(open(_arg("--state"))) if _has("--state") else {}
    sync_orders(config, state)

Step-by-step trace.

Requirement	Where it lives	Effect
logs on stderr	basicConfig(stream=sys.stderr)	stdout stays pure messages
SCHEMA first	step 1 before the loop	target can create/evolve the table
resume from bookmark	state.get(...).get("replication_key_value")	incremental, not full re-read
records in key order	fetch_orders_since ... ORDER BY updated_at	monotonic bookmark
periodic STATE	every 10k rows	crash resumes near the failure
final STATE	after the loop	bookmark = max emitted key

After deployment, the tap reads the prior bookmark, emits only newer rows in updated_at order, checkpoints STATE every 10k rows, and closes with a STATE equal to the highest key it actually emitted. A mid-run crash restarts from the last checkpoint — no corruption, no lost rows — and the target, echoing STATE only after flushing, keeps the bookmark honest end to end.

Output:

Behaviour	Before (buggy tap)	After (conformant loop)
stdout contents	messages + stray logs	messages only
First message per stream	sometimes RECORD	always SCHEMA
Restart after crash	loses rows	resumes at last checkpoint
Bookmark value	wall clock / guess	max emitted replication-key
Full vs incremental	re-reads table	reads only newer rows

Why this works — concept by concept:

• stdout/stderr split — messages on stdout, logs on stderr. Because logging is configured to stderr once at the top, no log line can ever corrupt the message stream the target parses.
• SCHEMA-before-RECORD ordering — emitting the stream's SCHEMA before any RECORD lets the target create or evolve the destination table before the first row lands, which is what makes the loader schema-driven rather than guess-driven.
• Bookmark from prior STATE — reading replication_key_value out of the incoming state makes the run incremental: the tap fetches only rows newer than the bookmark, so O(delta) work instead of O(table).
• Periodic + final STATE behind emitted data — checkpointing every 10k rows and closing with max_bm means the bookmark never claims more progress than was emitted; a crash resumes near the failure and a clean finish records exactly the high-watermark.
• Cost — O(delta) rows per run plus one STATE object per checkpoint (tiny). Compared to a full-table re-read every run (O(table)), the incremental loop is the difference between a seconds-long nightly delta and an hours-long re-scan. The only added cost is keeping records sorted by the replication key.

JSON
Topic — json
JSON parsing and message-stream problems

Practice →

ETL Topic — etl ETL problems on tap/target streaming

Practice →

---

3. Singer catalog and discovery

--discover introspects the source into a Singer catalog of streams, each with a schema and breadcrumb metadata

The mental model in one line: discovery is the phase where a tap, invoked with --discover, introspects its source and prints a catalog — a list of streams, each carrying a JSON schema, key_properties, and a metadata array of breadcrumb-scoped entries — and the operator then edits that catalog (or supplies metadata via the runner) to select which streams to sync and which replication method each uses, so the same tap can pull one table or fifty without a code change. The catalog is the contract between "what the source can offer" (discovery) and "what this pipeline wants" (selection). Getting comfortable reading and editing a catalog is what turns a tap from a black box into a controllable connector.

The catalog anatomy.

• streams. A list; one entry per table/endpoint the tap can produce. Each has a tap_stream_id (stable id), a stream (name), a schema, and metadata.
• schema. A JSON Schema describing the record shape. Targets use it to create and evolve the destination table; the tap uses it to validate records.
• key_properties. The primary key — the target upserts/dedupes on it.
• metadata. An array of {"breadcrumb": [...], "metadata": {...}} entries. The empty breadcrumb [] is stream-level metadata (selection, replication method, replication key); ["properties", "email"] is field-level metadata (inclusion, is-a-key).

The breadcrumb metadata model — the part everyone finds confusing.

• Stream-level (breadcrumb: []). Carries selected (sync this stream or not), replication-method (FULL_TABLE / INCREMENTAL / LOG_BASED), replication-key (which field to bookmark), and discovered facts like table-key-properties.
• Field-level (breadcrumb: ["properties", "<field>"]). Carries inclusion (available = selectable, automatic = always included such as a key, unsupported = the tap can't emit it) and selected for column-level selection.
• Who writes what. The tap discovers the structural facts (schema, keys, inclusion). The operator sets the intent (selected, replication-method, replication-key). Discovery output is a starting point you edit, not a final answer.

Discovery vs selection — two distinct steps.

• Discovery. tap --config config.json --discover > catalog.json. Read-only introspection; no data extracted.
• Selection. Edit catalog.json (or, under Meltano, use select: and metadata: in meltano.yml) to mark streams and fields selected and choose replication methods.
• Run. tap --config config.json --catalog catalog.json — the tap emits only the selected streams, using the selected replication method and key.

Common interview probes on discovery.

• "What does --discover produce?" — a catalog of streams with schema, keys, and metadata.
• "How do you choose which streams to sync?" — stream-level selected metadata (or Meltano select:).
• "Where does the replication method live?" — stream-level metadata (replication-method, replication-key).
• "What is inclusion: automatic?" — a field the tap always emits (typically a primary/replication key) regardless of selection.

Worked example — a discovery catalog with two streams

Detailed explanation. Run discovery against a source that has a users table (incremental-capable, has updated_at) and a regions lookup table (small, full-table). The catalog shows both streams with their schemas and stream-level metadata. Walk through the structure so you can read and edit any catalog.

• users. incremental on updated_at; id is the key.
• regions. full-table; code is the key.
• Metadata. stream-level selection + method; field-level inclusion.

Question. Produce the discovery catalog for these two streams with correct breadcrumb metadata.

Input.

Stream	key_properties	replication candidate	intended method
users	id	updated_at	INCREMENTAL
regions	code	(none)	FULL_TABLE

Code.

{
  "streams": [
    {
      "tap_stream_id": "users",
      "stream": "users",
      "schema": {
        "type": "object",
        "properties": {
          "id":         {"type": "integer"},
          "name":       {"type": "string"},
          "email":      {"type": ["string", "null"]},
          "updated_at": {"type": "string", "format": "date-time"}
        }
      },
      "key_properties": ["id"],
      "metadata": [
        {
          "breadcrumb": [],
          "metadata": {
            "selected": true,
            "replication-method": "INCREMENTAL",
            "replication-key": "updated_at",
            "table-key-properties": ["id"]
          }
        },
        {"breadcrumb": ["properties", "id"],
         "metadata": {"inclusion": "automatic"}},
        {"breadcrumb": ["properties", "updated_at"],
         "metadata": {"inclusion": "automatic"}},
        {"breadcrumb": ["properties", "email"],
         "metadata": {"inclusion": "available", "selected": true}}
      ]
    },
    {
      "tap_stream_id": "regions",
      "stream": "regions",
      "schema": {
        "type": "object",
        "properties": {
          "code": {"type": "string"},
          "name": {"type": "string"}
        }
      },
      "key_properties": ["code"],
      "metadata": [
        {
          "breadcrumb": [],
          "metadata": {
            "selected": true,
            "replication-method": "FULL_TABLE",
            "table-key-properties": ["code"]
          }
        }
      ]
    }
  ]
}

Step-by-step explanation.

1. Each stream is one object in streams, with a stable tap_stream_id, a schema, key_properties, and a metadata array. The tap discovered the schema and keys by introspecting the source; the operator edits the metadata to express intent.
2. The users stream-level metadata (breadcrumb: []) sets selected: true, replication-method: INCREMENTAL, and replication-key: updated_at. Those three fields are the entire "sync this table incrementally on updated_at" instruction.
3. Field-level metadata uses breadcrumbs like ["properties", "id"]. id and updated_at are inclusion: automatic — a key and a replication key are always emitted, whether or not you select them, because the pipeline can't function without them.
4. email is inclusion: available and selected: true, meaning the operator opted it in. Setting it to false would drop the column from the stream — column-level selection, useful for leaving PII behind at extraction time.
5. The regions stream is FULL_TABLE with no replication key — small lookup tables are cheap to re-read every run, and there is no incremental cursor to track. The catalog cleanly expresses "one table incremental, one table full" in the same file.

Output.

Stream	selected	method	key	replication-key
users	true	INCREMENTAL	id	updated_at
regions	true	FULL_TABLE	code	—
users.email	selected	—	—	column-level opt-in

Rule of thumb. Read a catalog top-down: stream-level metadata (breadcrumb: []) tells you whether and how a stream syncs; field-level metadata (breadcrumb: ["properties", ...]) tells you which columns and which are automatic. Edit intent (selected, replication-method, replication-key); never edit discovered structure (schema, keys) by hand.

Worked example — applying selection to run a subset

Detailed explanation. Discovery finds fifty streams, but this pipeline wants only three. Under raw Singer you edit the catalog's selected flags; under Meltano you declare select: and metadata: in meltano.yml and let the runner apply them to the discovered catalog. Walk through both so you can do it either way.

• Raw Singer. flip selected in catalog.json, pass --catalog.
• Meltano. select: globs + metadata: overrides, applied automatically.
• Result. only the chosen streams/fields are emitted.

Question. Select only users and orders (incremental) out of a fifty-stream source, and drop the email column.

Input.

Want	Raw Singer	Meltano
sync users, orders only	selected: true on those streams	select: globs
drop email column	selected: false on the field	!users.email exclusion glob
set replication method	stream metadata	metadata: block

Code.

# meltano.yml — selection + metadata applied to the discovered catalog
plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config:
        host: db-primary.internal
        database: production
      # Only these streams sync; everything else stays deselected.
      select:
        - public-users.*
        - public-orders.*
        - "!public-users.email"        # exclude the PII column at extraction
      # Replication intent, applied onto the discovered catalog:
      metadata:
        public-users:
          replication-method: INCREMENTAL
          replication-key: updated_at
        public-orders:
          replication-method: INCREMENTAL
          replication-key: updated_at

# See exactly what Meltano will sync after applying select + metadata
meltano select tap-postgres --list --all
#  [selected  ] public-users.id
#  [excluded  ] public-users.email
#  [selected  ] public-users.updated_at
#  [selected  ] public-orders.id
#  ...
meltano run tap-postgres target-jsonl   # emits only the selected streams

Step-by-step explanation.

1. Under Meltano you never hand-edit the catalog. The select: list is a set of glob rules over <stream>.<field>; public-users.* selects every field of that stream, and the leading ! on !public-users.email excludes one field — column-level security applied at extraction.
2. The metadata: block overlays replication intent onto whatever discovery found: it sets both streams to INCREMENTAL on updated_at. Meltano applies these overrides to the freshly-discovered catalog at run time, so schema drift in the source is picked up while your intent stays declarative.
3. meltano select tap-postgres --list --all is the dry-run: it prints the resolved selection so you can confirm email is excluded and the keys are selected before any data moves.
4. meltano run then discovers, applies selection + metadata, and pipes only the chosen streams to the target. The forty-seven unselected streams are never emitted — no wasted extraction.
5. The equivalent raw-Singer flow is: tap --discover > catalog.json, edit selected/replication-method by hand, then tap --catalog catalog.json | target. Meltano's declarative select:/metadata: is the same operation without the manual JSON surgery, and it survives re-discovery.

Output.

Stream / field	Resolved state
public-users (id, name, updated_at)	selected, INCREMENTAL
public-users.email	excluded
public-orders	selected, INCREMENTAL
47 other streams	deselected (not emitted)

Rule of thumb. Prefer declarative selection: select: globs plus a metadata: block in meltano.yml, verified with meltano select --list --all, beats hand-editing catalog JSON because it survives re-discovery and reads as intent. Use a leading ! glob to drop sensitive columns before they ever leave the source.

Senior interview question on the Singer catalog

A senior interviewer might ask: "A source exposes eighty tables but this warehouse feed needs eight of them — two incremental, six full-table — and one table has a raw SSN column that must never be extracted. Walk me through discovery, how the catalog's breadcrumb metadata expresses all of that, and how you'd keep the selection declarative and re-discovery-safe rather than a hand-edited JSON blob."

Solution Using discovery plus declarative selection and metadata overrides

# 1. Discover once to see everything the source can offer (read-only)
meltano invoke tap-postgres --discover > /tmp/catalog.json
jq '.streams[].tap_stream_id' /tmp/catalog.json | wc -l   # 80 streams

# 2. meltano.yml — express the full intent declaratively
plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config: {host: db-primary.internal, database: production}

      # Eight streams selected; the SSN column excluded everywhere.
      select:
        - public-orders.*
        - public-customers.*
        - "!public-customers.ssn"      # never extract the raw SSN
        - public-regions.*
        - public-products.*
        - public-suppliers.*
        - public-warehouses.*
        - public-carriers.*
        - public-tax_rates.*

      metadata:
        # two incremental streams
        public-orders:
          replication-method: INCREMENTAL
          replication-key: updated_at
        public-customers:
          replication-method: INCREMENTAL
          replication-key: updated_at
        # six full-table lookups (small, no cursor)
        public-regions:    {replication-method: FULL_TABLE}
        public-products:   {replication-method: FULL_TABLE}
        public-suppliers:  {replication-method: FULL_TABLE}
        public-warehouses: {replication-method: FULL_TABLE}
        public-carriers:   {replication-method: FULL_TABLE}
        public-tax_rates:  {replication-method: FULL_TABLE}

# 3. Prove the resolved selection before moving any data
meltano select tap-postgres --list --all | grep -E "customers.ssn|orders.id"
#  [excluded  ] public-customers.ssn
#  [automatic ] public-orders.id
meltano run tap-postgres target-snowflake

Step-by-step trace.

Requirement	Mechanism	Result
see all 80 streams	--discover (read-only)	full catalog, nothing extracted
sync only 8	select: globs	72 streams stay deselected
2 incremental	metadata: replication-method INCREMENTAL	bookmarked on updated_at
6 full-table	metadata: replication-method FULL_TABLE	re-read each run
never extract SSN	!public-customers.ssn exclusion	column dropped at the tap
re-discovery-safe	declarative in meltano.yml	survives source schema changes

After wiring, discovery still sees all eighty tables, but the run emits exactly eight streams — two carrying an updated_at bookmark, six re-read in full — and the ssn column is excluded at extraction so it never touches the network or the warehouse. Because selection lives in meltano.yml, re-running discovery next quarter picks up new columns without disturbing the intent.

Output:

Concern	Hand-edited catalog	Declarative meltano.yml
Streams synced	8 (until someone re-discovers)	8, re-discovery-safe
SSN exclusion	manual JSON edit, easily lost	! glob, permanent
Replication methods	scattered in catalog JSON	one metadata: block
Auditability	diff a big JSON blob	read the yml intent
Drift handling	re-edit by hand	overrides re-applied automatically

Why this works — concept by concept:

• Discovery as read-only introspection — --discover enumerates every stream the source can offer without extracting anything, so you plan selection against ground truth instead of guessing table names.
• Stream-level metadata — replication-method and replication-key at breadcrumb: [] are the entire per-table sync instruction; two streams get INCREMENTAL, six get FULL_TABLE, expressed in one metadata: block.
• Field-level exclusion — the !public-customers.ssn glob sets that field deselected, so the tap never emits it; column-level security enforced at the extraction boundary, not downstream.
• Declarative, re-discovery-safe selection — because select:/metadata: live in meltano.yml and are applied onto each fresh discovery, source schema changes are absorbed without losing your intent or re-hand-editing catalog JSON.
• Cost — O(selected streams) extraction instead of O(all streams); six full-table lookups are cheap by construction, two incremental streams are O(delta). The exclusion glob costs nothing and removes an entire class of PII-leak risk. Auditability improves from "diff an 80-stream JSON" to "read a short yml block."

Data Validation
Topic — data-validation
Data-validation problems on schema and catalog checks

Practice →

ETL Topic — etl ETL problems on stream selection and discovery

Practice →

---

4. Singer state and incremental replication

Singer state bookmarks a replication key so the next run resumes — the difference between a delta and a full re-read

The mental model in one line: Singer state is a JSON object of bookmarks — one per stream — that records how far a tap has progressed (typically the maximum replication-key value emitted), and because a tap accepts the previous run's state via --state and emits an updated state at the end, the pattern turns every run into an incremental delta (WHERE replication_key > bookmark) instead of an O(table) full re-read, provided the extraction mode is INCREMENTAL. State is the single durable artifact between runs; get its semantics right and your pipeline is cheap and resumable, get them wrong and you either re-read everything nightly or silently skip rows.

The three replication methods and their state stories.

• INCREMENTAL. The tap bookmarks a monotonic replication-key (updated_at, an incrementing id, a cursor). Each run emits WHERE key > bookmark and advances the bookmark to the new max. Cheap; blind to physical deletes (same limitation as timestamp CDC).
• FULL_TABLE. The tap re-reads the entire source every run. State is minimal (or a within-run resumption bookmark for large tables). Captures deletes implicitly (the row is simply absent next time) but is O(table) every run. Often paired with ACTIVATE_VERSION for atomic swaps.
• LOG_BASED. The tap tails the database's WAL/binlog/oplog; state is the log position (LSN / binlog coordinates / resume token). Captures every DML including deletes, sub-second, but requires source-DB replication permission.

The bookmark shape.

• Location. state["bookmarks"][<stream>].
• Incremental bookmark. {"replication_key": "updated_at", "replication_key_value": "2026-08-18T09:10:00+00:00"}.
• Full-table version. {"version": 1692300000000} — the active table version for ACTIVATE_VERSION.
• Log-based position. {"lsn": 24591040} or {"log_file": "...", "log_pos": 4} or {"resume_token": "..."}.

Why STATE ordering is a correctness property.

• Emit STATE only behind flushed records. The tap should emit a bookmark only for progress the target has (or will have) durably persisted. A well-behaved runner captures a STATE only after the target echoes it post-flush.
• Advance to max(emitted), not wall clock. Bookmark to the highest replication-key you actually emitted, exactly like a CDC watermark. Advancing to "now" risks skipping rows whose key is older than now but hadn't been read yet.
• Overlap, don't gap. Incremental replication is usually at-least-once: use key >= bookmark (inclusive) and dedupe downstream on the primary key, so a crash re-sends the boundary row rather than skipping it.

Common interview probes on state.

• "What makes a Singer pipeline incremental?" — a replication-key bookmark in STATE plus INCREMENTAL method.
• "Where is the resume point stored?" — the last STATE (a file under raw Singer; the system DB under Meltano).
• "Does incremental catch deletes?" — no (same as timestamp CDC); use FULL_TABLE or LOG_BASED for deletes.
• "Inclusive or exclusive bound?" — inclusive (>=) + downstream dedupe, so you never gap the boundary row.

Worked example — an incremental tap that bookmarks updated_at

Detailed explanation. Take the hand-written tap from section 2 and make it genuinely incremental: it reads the incoming bookmark, queries WHERE updated_at >= bookmark, emits records in key order, and closes with a STATE at the new max. Run it twice to see the bookmark advance and the second run do far less work.

• Bookmark in. previous replication_key_value (or config start_date on first run).
• Query. updated_at >= bookmark ORDER BY updated_at.
• Bookmark out. max updated_at emitted.

Question. Implement the incremental sync and show the state before/after two runs.

Input.

Run	Incoming bookmark	Rows matching >= bookmark	New bookmark
1 (bootstrap)	1970-01-01	3 (all)	2026-08-18T09:10
2	2026-08-18T09:10	1 (the boundary + newer)	2026-08-18T09:25

Code.

def sync_incremental(config, state, db):
    stream = "users"
    start = config.get("start_date", "1970-01-01T00:00:00+00:00")
    bookmark = (state.setdefault("bookmarks", {})
                     .setdefault(stream, {})
                     .get("replication_key_value", start))

    write({"type": "SCHEMA", "stream": stream, "schema": USERS_SCHEMA,
           "key_properties": ["id"], "bookmark_properties": ["updated_at"]})

    # Inclusive lower bound + downstream dedupe => at-least-once, no gaps
    rows = db.query(
        "SELECT id, name, updated_at FROM users "
        "WHERE updated_at >= %s ORDER BY updated_at ASC", (bookmark,))

    max_bm = bookmark
    for row in rows:
        write({"type": "RECORD", "stream": stream, "record": row})
        max_bm = row["updated_at"]

    state["bookmarks"][stream] = {
        "replication_key": "updated_at",
        "replication_key_value": max_bm,
    }
    write({"type": "STATE", "value": state})
    return state

Step-by-step explanation.

1. The bookmark is read out of the incoming state, defaulting to config.start_date on the very first run. That default is the bootstrap: no prior bookmark means "read from the beginning."
2. The query uses updated_at >= bookmark — an inclusive lower bound. Combined with the target upserting on id, this makes the pipeline at-least-once: the boundary row may be re-sent, but it is deduped on the primary key, so no row is ever skipped at the seam.
3. Rows come back ORDER BY updated_at ASC so the bookmark advances monotonically; the tap tracks the max as it emits. If rows arrived out of key order, a mid-run crash could bookmark past an unemitted row.
4. After the loop, the state's bookmark is set to the max updated_at actually emitted — never to a wall-clock "now" — and a single STATE message publishes it. On run 2, that bookmark makes the query return only the boundary row plus anything newer.
5. Run 1 (bootstrap) emits all three rows and bookmarks 09:10. Run 2 starts at 09:10, re-sends the 09:10 row (deduped downstream) plus the new 09:25 row, and bookmarks 09:25 — O(delta) work, exactly the point of incremental replication.

Output.

Run	State in	Rows emitted	State out
1	{} → start_date 1970	id 1,2,3	updated_at = 09:10
2	updated_at = 09:10	id 3 (boundary) + new	updated_at = 09:25
3	updated_at = 09:25	(none new)	updated_at = 09:25

Rule of thumb. Use an inclusive >= bound on the replication key, order rows by that key, bookmark to max(emitted), and dedupe downstream on the primary key. That combination gives you at-least-once incremental replication with no gaps at the run boundary — the Singer analogue of a CDC safety window.

Worked example — FULL_TABLE with ACTIVATE_VERSION

Detailed explanation. Some sources have no reliable replication key (no updated_at, no monotonic id) — a small dimension re-read in full each run. FULL_TABLE plus ACTIVATE_VERSION gives atomic swap semantics: the tap stamps every record of a run with a version, and emits ACTIVATE_VERSION at the end so the target can atomically promote the new snapshot and drop the old one. Walk through it.

• Version. a per-run integer (usually epoch ms).
• Records. each carries the run's version.
• Swap. ACTIVATE_VERSION tells the target "this version is complete; make it live."

Question. Implement a FULL_TABLE sync with versioned records and an end-of-run activation.

Input.

Element	Value
Stream	regions (no replication key)
Version	epoch-ms at run start
Activation	after all records emitted

Code.

import time

def sync_full_table(config, db):
    stream = "regions"
    version = int(time.time() * 1000)     # this run's table version

    write({"type": "SCHEMA", "stream": stream, "schema": REGIONS_SCHEMA,
           "key_properties": ["code"]})

    # (optional) tell the target a new version is starting
    write({"type": "ACTIVATE_VERSION", "stream": stream, "version": version})

    for row in db.query("SELECT code, name FROM regions"):
        write({"type": "RECORD", "stream": stream, "record": row,
               "version": version})     # every record stamped with the version

    # Final ACTIVATE_VERSION => atomically promote this snapshot, drop older rows
    write({"type": "ACTIVATE_VERSION", "stream": stream, "version": version})
    write({"type": "STATE", "value": {"bookmarks":
           {stream: {"version": version}}}})

Step-by-step explanation.

1. version is a per-run monotonic integer (epoch milliseconds). It labels every record of this run so the target can distinguish "rows from the current snapshot" from "rows from the previous snapshot."
2. The optional leading ACTIVATE_VERSION signals a new version is beginning; some targets use it to open a staging table for the version.
3. Every RECORD carries version, so as the full table streams in, the target accumulates the new snapshot tagged with this run's version alongside the still-live old version.
4. The final ACTIVATE_VERSION is the atomic swap: it tells the target "the new version is complete — promote it and delete any row not carrying this version." That is how FULL_TABLE captures deletes: a row deleted at the source simply isn't in the new version, so activation drops it downstream.
5. The STATE carries only the active version, not a replication-key — there is no cursor for a full-table stream. On the next run a new version is minted and the cycle repeats, each run leaving exactly one consistent snapshot live.

Output.

Phase	Target state
RECORDs streaming	new version loading beside the live one
final ACTIVATE_VERSION	new version promoted; non-matching rows dropped
deleted source row	absent from new version → removed on activation
STATE	{"version": <this run>}

Rule of thumb. Use FULL_TABLE + ACTIVATE_VERSION when a source has no reliable replication key. Stamp every record with a per-run version and emit a closing ACTIVATE_VERSION so the target swaps atomically and drops deleted rows — the full-table way to get delete handling that incremental replication can't.

Worked example — the delete-handling gap in incremental

Detailed explanation. The most common incremental-replication bug in production is silent divergence: the warehouse row count creeps above the source because incremental replication never learns about physical deletes. This is the exact blind spot timestamp CDC has, and the fixes are the same. Walk through the diagnosis and the three mitigations.

• Symptom. warehouse count > source count, growing over time.
• Cause. DELETE at source leaves no updated_at bump; incremental never re-reads it.
• Fixes. soft-delete, periodic FULL_TABLE reconcile, or switch to LOG_BASED.

Question. Diagnose the divergence and pick a mitigation for a table where deletes are rare but real.

Input.

Mitigation	Mechanism	Cost
soft-delete	deleted_at UPDATE, not DELETE	app change; keeps it incremental
periodic FULL_TABLE	occasional full re-read + activate	O(table) on the reconcile run
LOG_BASED	tail the WAL	needs replication permission

Code.

-- Root cause: incremental sees UPDATE/INSERT, never a physical DELETE
-- (identical blind spot to timestamp-based CDC)

-- Fix A — soft delete: keep incremental viable
--   before:  DELETE FROM users WHERE id = 42;
--   after:   UPDATE users SET deleted_at = now(), updated_at = now()
--            WHERE id = 42;        -- bumps updated_at => next run ships it

# Fix B — periodic FULL_TABLE reconcile via a second Meltano job.
#   Nightly incremental for freshness; weekly full-table to catch deletes.
schedules:
  - name: users-incremental
    extractor: tap-postgres
    loader: target-snowflake
    interval: "@hourly"          # INCREMENTAL in meltano.yml metadata
  - name: users-full-reconcile
    extractor: tap-postgres-full # same tap, metadata: FULL_TABLE
    loader: target-snowflake
    interval: "@weekly"          # ACTIVATE_VERSION drops deleted rows

Step-by-step explanation.

1. The divergence is diagnosed by comparing counts: SELECT count(*) at source vs warehouse. A steadily growing gap with no missing inserts points squarely at unhandled deletes — the incremental blind spot.
2. Fix A (soft-delete) keeps the stream incremental: the application replaces DELETE with UPDATE ... SET deleted_at, updated_at. Bumping updated_at means the next incremental run ships the row as an update carrying deleted_at, and downstream filters it out. Cheapest to run, but requires an application change.
3. Fix B runs two schedules against the same table: a frequent INCREMENTAL job for freshness and an infrequent FULL_TABLE job whose ACTIVATE_VERSION swap physically drops rows deleted at the source. The reconcile pays O(table) but only weekly.
4. Fix C (not shown in code) is switching the stream to LOG_BASED, which captures the DELETE from the WAL natively — the cleanest correctness story, at the price of requiring wal_level=logical and a replication slot.
5. The senior framing: incremental replication trades delete-correctness for cheapness, exactly like timestamp CDC. Name the trade explicitly and pick the mitigation that fits — soft-delete for app-owned tables, periodic full-table for read-only sources, log-based when the DBA cooperates.

Output.

Table profile	Recommended fix
app-owned, you control writes	soft-delete (stay incremental)
read-only, deletes rare	periodic FULL_TABLE reconcile
high-value, deletes matter	LOG_BASED (WAL)
tiny lookup	FULL_TABLE every run

Rule of thumb. Incremental replication is blind to physical deletes — state your mitigation up front. Soft-delete keeps it incremental, a periodic FULL_TABLE reconcile catches deletes cheaply for read-only sources, and LOG_BASED captures them natively when you have replication permission.

Senior interview question on Singer state

A senior interviewer might ask: "You inherit a Singer pipeline where the warehouse row count keeps drifting above the source, and after a crash last week it skipped a few hundred rows. Walk me through how Singer state and replication keys work, why both bugs happen, and how you'd fix the incremental bookmarking and the delete handling — including where the state actually lives under Meltano."

Solution Using an inclusive bookmark, at-least-once dedupe, and a reconcile job

# 1. Incremental tap logic — inclusive bound, bookmark to max emitted
def sync(config, state, db):
    stream = "orders"
    bm = (state.get("bookmarks", {}).get(stream, {})
               .get("replication_key_value", config["start_date"]))

    write({"type": "SCHEMA", "stream": stream, "schema": ORDERS_SCHEMA,
           "key_properties": ["id"], "bookmark_properties": ["updated_at"]})

    rows = db.query(
        "SELECT * FROM orders WHERE updated_at >= %s "   # inclusive => no gap
        "ORDER BY updated_at ASC", (bm,))

    max_bm = bm
    for i, row in enumerate(rows, 1):
        write({"type": "RECORD", "stream": stream, "record": row})
        max_bm = row["updated_at"]
        if i % 5000 == 0:                                 # periodic checkpoint
            _emit_state(state, stream, max_bm)
    _emit_state(state, stream, max_bm)                    # final bookmark

def _emit_state(state, stream, value):
    state.setdefault("bookmarks", {})[stream] = {
        "replication_key": "updated_at", "replication_key_value": value}
    write({"type": "STATE", "value": state})

# 2. Meltano: state lives in the system DB; two schedules cover deletes
state_backend:
  uri: postgresql://meltano@meta-db/meltano   # durable, shared state store
schedules:
  - name: orders-incremental
    extractor: tap-postgres          # metadata: INCREMENTAL on updated_at
    loader: target-snowflake
    interval: "@hourly"
  - name: orders-full-reconcile
    extractor: tap-postgres-full     # metadata: FULL_TABLE + ACTIVATE_VERSION
    loader: target-snowflake
    interval: "@daily"

-- 3. Downstream dedupe (target upserts on the primary key => at-least-once safe)
MERGE INTO analytics.orders t
USING staging.orders_delta s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.status = s.status, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (id, status, updated_at)
                     VALUES (s.id, s.status, s.updated_at);

Step-by-step trace.

Bug	Root cause	Fix in the solution
skipped rows after crash	STATE ahead of data / exclusive bound	inclusive >= + periodic checkpoint
count drift upward	incremental blind to deletes	daily FULL_TABLE reconcile
bookmark lost on restart	state file not durable	Meltano state backend (Postgres)
double-counted rows	at-least-once redelivery	MERGE upsert on primary key
bookmark past unread row	rows not key-ordered	ORDER BY updated_at ASC

After the fix, the incremental job resumes from a durable bookmark in Meltano's state backend, re-sends the boundary row (harmlessly deduped by the MERGE), and never gaps at a crash seam; the daily full-table reconcile activates a fresh version and drops rows deleted at the source, so the warehouse count tracks the source instead of drifting.

Output:

Metric	Before	After
Rows skipped on crash	hundreds	0 (inclusive + checkpoint)
Count drift vs source	grows daily	flat (reconcile)
State durability	local file, lost on restart	Meltano Postgres backend
Duplicate handling	double counts	idempotent MERGE
Freshness	hourly	hourly + daily reconcile

Why this works — concept by concept:

• Inclusive replication-key bound — updated_at >= bookmark re-sends the boundary row instead of risking a gap at the seam; paired with a primary-key MERGE it is at-least-once with no skips.
• Periodic + final STATE — checkpointing every 5k rows and closing with max(emitted) means a crash resumes near the failure and the bookmark never claims unemitted progress.
• Meltano state backend — moving state from a local file to a shared Postgres store makes the bookmark survive worker restarts and lets parallel workers coordinate; this is where "state actually lives" under Meltano.
• FULL_TABLE reconcile for deletes — the daily full-table job with ACTIVATE_VERSION drops rows absent from the new snapshot, closing the incremental delete blind spot without giving up hourly freshness.
• Cost — hourly O(delta) incremental plus one daily O(table) reconcile, versus O(table) every hour if you gave up incrementality. The MERGE adds an indexed upsert per delta row. Net: freshness and delete-correctness at a fraction of full-refresh compute.

ETL
Topic — etl
ETL problems on incremental replication and bookmarks

Practice →

Data Processing Topic — data-processing Data-processing problems on state and checkpointing

Practice →

---

5. Meltano orchestration, targets, and SCD

Meltano turns taps and targets into a declarative pipeline — managed installs, config, state, stream maps, and an SCD load in the target

The mental model in one line: Meltano is the runner that makes Singer operable — a meltano.yml declares extractors (taps) and loaders (targets) with their pip_url, config, select:, and metadata:; meltano install builds an isolated virtualenv per plugin; meltano run tap-x target-y constructs the tap | target pipe and persists the emitted state in a durable backend; and stream maps and the target's SCD support let you transform, mask, and historise data on the way in — all without hand-shuttling catalog.json or state.json files. Meltano is to Singer what a package manager plus a scheduler is to a pile of executables: the same primitives, made reproducible and declarative.

What Meltano owns.

• Plugin management. pip_url + variant pin each tap/target; meltano install creates an isolated venv per plugin so dependency trees never collide.
• Config + secrets. Config lives in meltano.yml (non-secret) and environment/.env (secret, referenced as $VAR). Environments (dev, prod) override config per stage.
• State. A state backend (local, or a database/blob store) stores one bookmark per (extractor, loader). No manual --state shuffling.
• Selection + metadata. select: globs and metadata: overrides are applied to each fresh discovery, so selection is declarative and re-discovery-safe.
• Run + schedule. meltano run builds the pipe; schedules: (with the Airflow/Dagster utilities) run it on a cadence.

Stream maps — inline transformation and masking.

• What. A mapper plugin (meltano-map-transformer) rewrites messages between the tap and the target: rename/drop fields, hash PII, add computed columns, split streams.
• Why in the pipe. Masking a national ID or hashing an email before it reaches the target keeps sensitive raw values out of the warehouse entirely.
• Example ops. __else__: __NULL__ to drop unlisted fields; hash_email: md5(email) to pseudonymise; stream_maps keyed per stream.

SCD in the target — historising dimensions.

• Type 1 (overwrite). The target upserts on the primary key; history is lost. This is the default upsert behaviour of most warehouse targets.
• Type 2 (versioned history). Each change becomes a new row with valid_from / valid_to / is_current and a surrogate key. Some targets/loaders support this directly; more commonly the target loads raw and a downstream dbt model builds the SCD Type 2 dimension.
• The _sdc_* metadata columns. Singer targets stamp records with _sdc_extracted_at, _sdc_batched_at, _sdc_received_at, and _sdc_sequence — the timestamps a downstream SCD model uses to order changes and set valid_from.

Common interview probes on Meltano.

• "What does Meltano add over raw Singer?" — installs, config/secrets, state persistence, selection, scheduling, maps.
• "Where does state live?" — the configured state backend, one bookmark per (extractor, loader).
• "How do you mask PII in flight?" — a stream map (mapper) between tap and target.
• "How do you build SCD Type 2?" — loader upsert for Type 1; a dbt model over _sdc_* columns for Type 2 history.

Worked example — a meltano.yml wiring tap-postgres → target-jsonl

Detailed explanation. The canonical Meltano project: one extractor (tap-postgres) with selection and incremental metadata, one loader (target-jsonl), two environments, and a meltano run. This is the smallest thing that is a real, reproducible pipeline. Walk through every block.

• Extractor. tap-postgres, selected streams, INCREMENTAL metadata.
• Loader. target-jsonl writing to a local directory.
• Run. meltano install then meltano run.

Question. Author a complete meltano.yml for an incremental Postgres → JSONL pipeline and run it.

Input.

Block	Value
extractor	tap-postgres (meltanolabs)
streams	public-orders, public-customers
method	INCREMENTAL on updated_at
loader	target-jsonl → ./output

Code.

version: 1
default_environment: dev
project_id: singer-demo
environments:
  - name: dev
  - name: prod

plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config:
        host: db-primary.internal
        port: 5432
        database: production
        user: cdc_reader
        # password comes from the environment, never committed:
        password: $TAP_POSTGRES_PASSWORD
        start_date: "2026-01-01T00:00:00Z"
      select:
        - public-orders.*
        - public-customers.*
        - "!public-customers.ssn"
      metadata:
        public-orders:
          replication-method: INCREMENTAL
          replication-key: updated_at
        public-customers:
          replication-method: INCREMENTAL
          replication-key: updated_at

  loaders:
    - name: target-jsonl
      variant: andyh1203
      pip_url: target-jsonl
      config:
        destination_path: ./output

export TAP_POSTGRES_PASSWORD='strong-secret'
meltano install                          # isolated venv per plugin
meltano run tap-postgres target-jsonl    # discover -> pipe -> persist STATE
ls output/                               # public-orders.jsonl  public-customers.jsonl

Step-by-step explanation.

1. The plugins.extractors block pins tap-postgres by variant + pip_url, so meltano install reproducibly builds the exact tap in its own virtualenv — the reproducibility that raw pip install doesn't guarantee across machines.
2. Config is split: non-secret values (host, database, start_date) live in meltano.yml; the password is $TAP_POSTGRES_PASSWORD, resolved from the environment so no secret is committed.
3. select: opts in two streams and excludes the ssn column with a ! glob; metadata: sets both streams to INCREMENTAL on updated_at. Selection and replication intent are declarative and applied onto each fresh discovery.
4. The loader target-jsonl writes each stream to ./output/<stream>.jsonl. Swapping this block for target-snowflake is the only change needed to load a warehouse instead — the extractor is untouched, proving the tap/target decoupling.
5. meltano run tap-postgres target-jsonl discovers, applies selection, builds the pipe, and — the operational win over raw Singer — captures the emitted STATE into Meltano's state backend automatically. The next run resumes with no manual --state.

Output.

Artifact	Result
output/public-orders.jsonl	selected order rows, incremental
output/public-customers.jsonl	customer rows, ssn excluded
Meltano state backend	bookmark per stream, auto-persisted
swap to prod warehouse	change only the loader block

Rule of thumb. Keep secrets in the environment ($VAR), selection and replication intent in select:/metadata:, and let meltano run own state persistence. Swapping the loader block — and nothing else — is what lets one extractor feed a local smoke test and a production warehouse.

Worked example — masking PII with a stream map

Detailed explanation. A customers stream carries a raw email and a national_id. You want the email pseudonymised and the national ID dropped before the data reaches the warehouse — not fixed downstream. A Meltano stream map (a mapper plugin in the pipe) rewrites the messages in flight. Walk through it.

• Mapper. meltano-map-transformer sits between tap and target.
• Ops. hash email; drop national_id; keep everything else.
• Effect. the target never sees the raw values.

Question. Configure a stream map that hashes email and removes national_id from the customers stream.

Input.

Field	In	Out
email	[e-mail]	md5 hash
national_id	123-45-6789	removed
id, name	unchanged	unchanged

Code.

plugins:
  extractors:
    - name: tap-postgres
      # ... as before ...

  mappers:
    - name: pii-mask
      variant: meltano
      pip_url: meltano-map-transformer
      mappings:
        - name: mask-customers
          config:
            stream_maps:
              public-customers:
                email: md5(email)          # pseudonymise, join-stable
                national_id: __NULL__      # remove the field entirely
                # all other fields pass through unchanged

  loaders:
    - name: target-jsonl
      # ... as before ...

# Insert the mapper mapping into the run, between tap and target
meltano run tap-postgres mask-customers target-jsonl
# customers.jsonl now has hashed email and NO national_id field

Step-by-step explanation.

1. The mapper is declared as its own plugin with a named mapping (mask-customers). In the run command it sits between the extractor and loader — tap | mapper | target — so it sees every message the tap emits before the target does.
2. stream_maps.public-customers scopes the transformation to that one stream. email: md5(email) replaces the value with its MD5 hash — a stable pseudonym so downstream joins on email still work without ever exposing the raw address.
3. national_id: __NULL__ is the mapper's directive to drop the field entirely from both the SCHEMA and every RECORD, so the column never reaches the target's table at all.
4. Fields not mentioned pass through unchanged, so id and name flow normally. The mapper also rewrites the stream's SCHEMA message so the target creates a table without national_id and with email typed as a string hash.
5. Because the masking happens in the pipe, the raw email and national_id never land in the warehouse — the strongest form of column-level protection, enforced at ingestion rather than patched in a downstream view.

Output.

customers record	Before mapper	After mapper
email	[e-mail]	534b44a19bf... (md5)
national_id	123-45-6789	(field absent)
id / name	7 / Alice	7 / Alice

Rule of thumb. Mask and drop PII with a stream map in the pipe (tap | mapper | target), not in a downstream view. Hashing keeps joins working; __NULL__ removes a field from schema and records alike — sensitive raw values never touch the destination.

Worked example — SCD Type 2 in the loader

Detailed explanation. A dim_customer dimension must keep history: when a customer's tier changes, you want the old row closed and a new row opened, not an overwrite. Singer targets stamp _sdc_* metadata; the standard 2026 pattern loads the raw stream and builds SCD Type 2 in a downstream dbt model using those columns. Walk through the model.

• Raw load. target upserts/append with _sdc_extracted_at, _sdc_sequence.
• SCD model. window over changes per key; set valid_from/valid_to/is_current.
• Result. a full history dimension.

Question. Build the SCD Type 2 dimension for dim_customer from a Singer-loaded raw table.

Input.

Column	Source
customer_id	business key
tier	tracked attribute
_sdc_extracted_at	Singer metadata → valid_from
_sdc_sequence	Singer metadata → tie-breaker

Code.

-- Raw table as loaded by target-snowflake (Singer stamps the _sdc_* columns)
--   raw.customers(customer_id, tier, updated_at,
--                 _sdc_extracted_at, _sdc_sequence, _sdc_batched_at)

-- SCD Type 2 dimension: one row per (customer_id, change), with validity window
CREATE OR REPLACE TABLE analytics.dim_customer AS
WITH changes AS (
    SELECT
        customer_id,
        tier,
        _sdc_extracted_at AS valid_from,
        _sdc_sequence,
        -- next change's timestamp closes this version's window
        LEAD(_sdc_extracted_at) OVER (
            PARTITION BY customer_id
            ORDER BY _sdc_extracted_at, _sdc_sequence
        ) AS valid_to
    FROM (
        -- collapse consecutive identical tiers to real changes only
        SELECT *,
               tier <> LAG(tier) OVER (PARTITION BY customer_id
                                       ORDER BY _sdc_extracted_at, _sdc_sequence)
                    OR LAG(tier) OVER (PARTITION BY customer_id
                                       ORDER BY _sdc_extracted_at, _sdc_sequence) IS NULL
                    AS is_change
        FROM raw.customers
    )
    WHERE is_change
)
SELECT
    {{ dbt_utils.generate_surrogate_key(['customer_id','valid_from']) }} AS customer_sk,
    customer_id,
    tier,
    valid_from,
    COALESCE(valid_to, TIMESTAMP '9999-12-31') AS valid_to,
    valid_to IS NULL                            AS is_current
FROM changes;

Step-by-step explanation.

1. The Singer target loaded raw.customers and stamped each row with _sdc_extracted_at (when the tap read it) and _sdc_sequence (order within a batch). Those metadata columns are the reliable ordering signal the SCD model needs — more trustworthy than the source updated_at alone.
2. The inner query flags real changes: tier <> LAG(tier) marks a row where the tracked attribute actually changed (and the first-ever row via the IS NULL clause), so consecutive identical loads don't create spurious history rows.
3. valid_from is _sdc_extracted_at for the change; valid_to is the next change's _sdc_extracted_at via LEAD, so each version's window closes exactly when the next version opens.
4. The current row has a NULL LEAD (no next change), so is_current = valid_to IS NULL and its valid_to is set to the far-future sentinel 9999-12-31 — the standard open-ended SCD Type 2 convention.
5. A surrogate key over (customer_id, valid_from) uniquely identifies each version, so fact tables can join to the exact historical dimension row that was current at event time. The result is a full Type 2 history built from Singer's own metadata, no bespoke CDC required.

Output.

customer_sk	customer_id	tier	valid_from	valid_to	is_current
a1f…	7	silver	2026-01-10	2026-04-02	false
b2e…	7	gold	2026-04-02	2026-07-15	false
c3d…	7	platinum	2026-07-15	9999-12-31	true

Rule of thumb. Load raw with the Singer target (it stamps _sdc_extracted_at / _sdc_sequence), then build SCD Type 2 in a downstream dbt model: flag real changes with LAG, close windows with LEAD, mark is_current on the open-ended row. Order by _sdc_*, not the source timestamp, for correct history.

Senior interview question on Meltano and SCD

A senior interviewer might ask: "Design a production Meltano pipeline that ingests a Postgres customers table into Snowflake with hourly freshness, masks the email and drops the SSN before it lands, keeps state durably, and exposes a dim_customer with full SCD Type 2 history. Walk me through the meltano.yml, the stream map, the state backend, and the SCD model — and tell me what makes it idempotent."

Solution Using Meltano with a mapper, a durable state backend, and a dbt SCD model

# meltano.yml — extractor + mapper + loader, durable state, prod environment
version: 1
default_environment: prod
state_backend:
  uri: postgresql://meltano@meta-db/meltano   # durable, shared state store

plugins:
  extractors:
    - name: tap-postgres
      variant: meltanolabs
      pip_url: meltanolabs-tap-postgres
      config: {host: db-primary.internal, database: production,
               password: $TAP_PG_PASSWORD, start_date: "2026-01-01T00:00:00Z"}
      select: ["public-customers.*", "!public-customers.ssn"]
      metadata:
        public-customers: {replication-method: INCREMENTAL, replication-key: updated_at}

  mappers:
    - name: pii-mask
      variant: meltano
      pip_url: meltano-map-transformer
      mappings:
        - name: mask-customers
          config:
            stream_maps:
              public-customers:
                email: md5(email)      # pseudonymise
                ssn: __NULL__          # belt-and-braces (also excluded in select)

  loaders:
    - name: target-snowflake
      variant: meltanolabs
      pip_url: meltanolabs-target-snowflake
      config: {account: $SF_ACCOUNT, database: RAW, default_target_schema: singer}

schedules:
  - name: customers-hourly
    interval: "@hourly"
    extractor: tap-postgres
    loader: target-snowflake

export TAP_PG_PASSWORD=... SF_ACCOUNT=...
meltano install
meltano run tap-postgres mask-customers target-snowflake   # hourly via schedule

-- dbt model dim_customer.sql — SCD Type 2 from Singer's _sdc_* metadata
WITH changes AS (
  SELECT customer_id, tier, email,
         _sdc_extracted_at AS valid_from,
         LEAD(_sdc_extracted_at) OVER (PARTITION BY customer_id
             ORDER BY _sdc_extracted_at, _sdc_sequence) AS valid_to
  FROM {{ source('singer','customers') }}
  QUALIFY tier <> LAG(tier) OVER (PARTITION BY customer_id
              ORDER BY _sdc_extracted_at, _sdc_sequence)
       OR LAG(tier) OVER (PARTITION BY customer_id
              ORDER BY _sdc_extracted_at, _sdc_sequence) IS NULL
)
SELECT {{ dbt_utils.generate_surrogate_key(['customer_id','valid_from']) }} AS customer_sk,
       customer_id, tier, email, valid_from,
       COALESCE(valid_to, TIMESTAMP '9999-12-31') AS valid_to,
       valid_to IS NULL AS is_current
FROM changes;

Step-by-step trace.

Requirement	Mechanism	Result
hourly freshness	schedules: @hourly + INCREMENTAL	O(delta) each hour
mask email	stream map md5(email)	raw email never lands
drop SSN	!select + mapper __NULL__	column gone at ingestion
durable state	state_backend: postgresql://	resumes across restarts
SCD Type 2	dbt model over _sdc_*	full history dimension
idempotency	key upsert + _sdc_sequence order	reruns don't duplicate

After deployment, the hourly job extracts only changed customers, the mapper hashes the email and nulls the SSN in flight, target-snowflake upserts the raw stream into RAW.singer.customers with _sdc_* metadata, and the dbt dim_customer model turns that raw history into an SCD Type 2 dimension. State lives in Meltano's Postgres backend, so a worker restart resumes from the last bookmark instead of re-reading the table.

Output:

Metric	Naive (full refresh + view masking)	Meltano + mapper + dbt SCD
Freshness	nightly	hourly (incremental)
Raw PII in warehouse	yes, masked in view	no — masked in flight
History	overwrite (Type 1)	full SCD Type 2
State durability	ad-hoc file	Postgres state backend
Rerun safety	may double-count	idempotent upsert

Why this works — concept by concept:

• Declarative extractor + incremental metadata — select: and metadata: express which streams and which replication method, applied onto each discovery, so hourly runs pull only O(delta) rows.
• Stream map masking in the pipe — the mapper rewrites SCHEMA and RECORD messages between tap and target, so email is hashed and ssn removed before the warehouse ever sees them — column security at the ingestion boundary.
• Durable state backend — moving state into shared Postgres makes the bookmark survive restarts and lets the scheduler run reliably; this is Meltano's core operational value over raw Singer.
• _sdc_*-driven SCD Type 2 — the target stamps _sdc_extracted_at/_sdc_sequence, and the dbt model orders by them (not the source timestamp) to close/open validity windows correctly, yielding full history without bespoke CDC.
• Idempotency — the target upserts on the primary key and the SCD model orders deterministically by _sdc_sequence, so an at-least-once redelivery or a rerun produces the same dimension — no duplicate history rows.
• Cost — hourly O(delta) extraction, a lightweight in-pipe hash, one warehouse upsert per delta row, and an incremental dbt build. Compared to nightly full refresh with downstream view masking, this is fresher, safer with PII, and history-complete at a fraction of the scan cost.

ETL
Topic — etl
ETL problems on orchestrated ELT pipelines

Practice →

Data Transformation
Topic — data-transformation
Data-transformation problems on SCD and dimensions

Practice →

---

Cheat sheet — Singer & Meltano connector recipes

• The one-line spec. A tap prints newline-delimited JSON messages to stdout; a target reads them from stdin; join them with a Unix pipe (tap | target). Logs go to stderr. Any conformant tap composes with any conformant target — that decoupling is the entire value proposition.
• Message-type quick reference. SCHEMA (stream shape + key_properties + bookmark_properties, must precede its RECORDs), RECORD (one row under record), STATE (resumable bookmark, emitted behind flushed data), ACTIVATE_VERSION (atomic table-version swap for FULL_TABLE), and the optional BATCH (bulk file transfer). Memorise the ordering: SCHEMA → RECORD… → STATE.
• Tap CLI contract. tap --config config.json --discover > catalog.json to introspect; edit the catalog (or use Meltano select:/metadata:); tap --config config.json --catalog catalog.json --state state.json to run incrementally. --config is connection + tuning, --catalog is selection, --state is the resume bookmark.
• Catalog metadata template. Stream-level (breadcrumb: []): {selected, replication-method, replication-key, table-key-properties}. Field-level (breadcrumb: ["properties","<f>"]): {inclusion: available|automatic|unsupported, selected}. Edit intent (selected, method, key); never hand-edit discovered structure (schema, keys).
• Incremental bookmark template. state["bookmarks"][stream] = {"replication_key": "updated_at", "replication_key_value": <max emitted>}. Query WHERE updated_at >= bookmark ORDER BY updated_at ASC, dedupe downstream on the primary key (at-least-once, no gaps), and advance to max(emitted) — never wall clock.
• Three protocol bugs to avoid. (1) logging to stdout → target JSONDecodeError; fix: log to stderr. (2) RECORD before SCHEMA → "no schema for stream"; fix: SCHEMA first. (3) STATE ahead of flushed data → rows lost on crash; fix: STATE last, target flushes before echoing. Every protocol bug is one of these three.
• Replication-method decision matrix. INCREMENTAL (replication-key bookmark; cheap; blind to deletes) for tables with a monotonic updated_at/id. FULL_TABLE (+ACTIVATE_VERSION; O(table); captures deletes) for small lookups or keyless sources. LOG_BASED (tail the WAL; sub-second; captures deletes; needs replication permission) for high-value, delete-sensitive tables. Same trade space as CDC, expressed in the catalog.
• meltano.yml skeleton. plugins.extractors[] (name, variant, pip_url, config, select:, metadata:), plugins.loaders[], optional plugins.mappers[], environments:, state_backend:, schedules:. Then meltano install (venv per plugin) and meltano run tap-x target-y (pipe + auto-persisted state).
• Selection globs. select: ["public-orders.*", "public-customers.*", "!public-customers.ssn"] — .* selects a stream's fields, a leading ! excludes a field (column-level security at extraction). Verify with meltano select tap --list --all before running.
• Stream-map masking. mappers[].mappings[].config.stream_maps.<stream>: {email: md5(email), ssn: __NULL__} and run tap | mapper | target. Hash to keep joins working; __NULL__ to drop a field from schema and records alike — raw PII never reaches the destination.
• SCD Type 2 recipe. Load raw with the Singer target (it stamps _sdc_extracted_at, _sdc_sequence, _sdc_batched_at). In dbt: flag real changes with LAG(attr) <> attr, set valid_from = _sdc_extracted_at, valid_to = LEAD(_sdc_extracted_at), is_current = valid_to IS NULL, surrogate key over (business_key, valid_from). Order by _sdc_*, not the source timestamp.
• Build vs reuse. Reuse a Meltano Hub tap when a mature one exists; build with the Singer SDK (subclass Tap/Stream, define schema, implement pagination + replication_key) when it doesn't. Pin SDK-built taps in your own git repo for reviewability.
• When open-source ELT wins. No managed connector exists, the source is VPC-locked, volume makes managed per-row pricing hurt, or compliance needs the connector reviewable in your repo. Otherwise buying managed convenience is a legitimate call — name the constraint.

Frequently asked questions

What is a Singer tap in one sentence?

A Singer tap is a self-contained executable that reads from one data source and prints a stream of newline-delimited JSON messages — SCHEMA (the shape of a stream), RECORD (one row), and STATE (a resumable bookmark) — to stdout, so that any conformant Singer target can read those messages from stdin and load them into a destination without the two ever sharing code. The tap owns source-specific concerns (authentication, pagination, discovery of available streams, incremental bookmarking on a replication key), while the target owns destination-specific concerns (creating tables, upserting, type mapping). Because the only contract between them is the message format, one tap composes with every target and one target with every tap — which is exactly why open-source ELT built on Singer decouples the "how many sources" problem from the "how many destinations" problem.

Singer vs Airbyte vs Fivetran — when do I pick each?

Pick Singer/Meltano when you need do-it-yourself control: a connector that doesn't exist yet (build it with the Singer SDK), a source locked inside a VPC where a hosted control plane can't reach, high row volume where managed per-row pricing hurts, or a compliance requirement that connectors be pinned and reviewable in your own git repo. Pick Airbyte when you want an open-source platform with a UI, a large prebuilt connector catalog, and a scheduler — it can even run Singer taps — but you accept operating the platform. Pick Fivetran (or Stitch, which originated Singer) when you want fully-managed, zero-ops ingestion for common SaaS sources and are willing to pay per active row for that convenience. The honest senior framing: Singer/Meltano trades money for engineering time — you own the tap's bugs and on-call — so it wins precisely when a managed connector is missing, unreachable, too expensive, or insufficiently controllable, and a managed tool wins for common, reachable, moderate-volume sources.

What is the Singer catalog and how does discovery work?

The Singer catalog is the JSON document a tap produces when you run it with --discover: it lists every stream (table or endpoint) the tap can offer, and for each one a JSON schema, its key_properties, and a metadata array. Discovery is read-only introspection — tap --config config.json --discover > catalog.json — and it moves no data; it just enumerates what's available. You then edit the catalog (or, under Meltano, declare select: globs and a metadata: block) to express intent: which streams to sync (selected), which replication method each uses (replication-method), and which field to bookmark (replication-key). Metadata is breadcrumb-scoped — the empty breadcrumb [] carries stream-level settings, while ["properties","<field>"] carries field-level settings like inclusion: available|automatic|unsupported. Running tap --catalog catalog.json then emits only the selected streams using the selected methods, so the same tap serves one table or fifty with no code change.

How does Singer state make a pipeline incremental?

Singer state is a JSON object of per-stream bookmarks that records how far the tap has progressed — typically the maximum replication-key value it emitted, such as {"bookmarks": {"orders": {"replication_key": "updated_at", "replication_key_value": "2026-08-18T09:10:00Z"}}}. A tap accepts the previous run's state via --state state.json and, for an INCREMENTAL stream, queries only rows newer than the bookmark (WHERE updated_at >= bookmark), then emits an updated STATE at the end. That turns each run into an O(delta) read instead of an O(table) full re-read. The correctness rules: emit STATE only behind data the target has durably flushed, use an inclusive lower bound plus a downstream primary-key dedupe (so a crash re-sends the boundary row rather than skipping it), and advance the bookmark to the max value actually emitted — never to wall-clock "now." Under Meltano the state isn't a loose file; it lives in a durable state backend, one bookmark per (extractor, loader), so runs resume across worker restarts.

What does Meltano add on top of Singer?

Raw Singer gives you taps, targets, and a message contract; Meltano makes them operable. A single declarative meltano.yml pins each plugin by pip_url and variant, and meltano install builds an isolated virtualenv per plugin so their dependency trees never collide. Meltano manages configuration and secrets (non-secrets in the yml, secrets as $VAR from the environment), applies select: globs and metadata: overrides onto each fresh discovery, and — critically — persists the emitted STATE in a durable backend so you never hand-shuttle state.json. meltano run tap-x target-y builds the tap | target pipe for you; schedules: run it on a cadence; and mappers let you transform or mask records in flight (tap | mapper | target). In short, Meltano is to Singer what a package manager plus a scheduler plus a secrets manager is to a pile of standalone executables: the same primitives, made reproducible, declarative, and schedulable.

How do I load SCD Type 2 with a Singer target?

Most Singer targets do a Type 1 upsert (overwrite on the primary key), so the standard 2026 pattern for SCD Type 2 is to load raw with the target and build the history dimension downstream — usually in dbt — using the _sdc_* metadata columns the target stamps on every row (_sdc_extracted_at, _sdc_sequence, _sdc_batched_at). In the model you order changes by _sdc_extracted_at, _sdc_sequence (more reliable than the source timestamp), flag real changes with tier <> LAG(tier) so repeated identical loads don't create spurious versions, set valid_from = _sdc_extracted_at and valid_to = LEAD(_sdc_extracted_at) to close each version's window, mark is_current = valid_to IS NULL, and mint a surrogate key over (business_key, valid_from). The open row gets a far-future valid_to sentinel like 9999-12-31. For the extraction side, pair it with FULL_TABLE + ACTIVATE_VERSION or INCREMENTAL depending on whether you need deletes reflected in the history. The result is a full Type 2 dimension built from Singer's own metadata, no bespoke change-data-capture required.

Practice on PipeCode

• Drill the ETL practice library → for the tap/target, incremental-load, watermark, and open-source ingestion problems senior interviewers love.
• Rehearse on the data-transformation practice library → for the SCD Type 2, dimension-modelling, and reshaping patterns that sit downstream of a Singer load.
• Sharpen the message-parsing muscle with the JSON practice library → for the SCHEMA/RECORD/STATE parsing, schema-validation, and catalog-editing scenarios.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-axis connector decision matrix against real graded inputs.

Lock in Singer & Meltano muscle memory

Docs explain the spec. PipeCode drills explain the decision — when incremental replication is blind to deletes, when a stray log line corrupts the stream, when FULL_TABLE plus ACTIVATE_VERSION earns its keep, when Meltano's state backend saves a restart. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice data-transformation problems →