Managed Ingestion: Fivetran, Airbyte, and dlt

In plain terms

A data connector packages source-specific extraction and destination loading so that a platform team does not have to implement every API from scratch. The earlier ingestion articles showed what can fail: pagination, checkpoints, updates, schema drift, and retries. Buying a service or using a library changes who maintains those mechanisms. It does not remove the need to verify them for the source and destination you actually use.

This is a learning and verification article, not an account of operating all three products in production. Fivetran and Airbyte are compared using documentation. The lab uses a small SQLite model and a separate local dlt/DuckDB pipeline; it does not connect to a live SaaS account. Here “managed ingestion” includes a hosted service, a platform you operate, and a library whose execution you own.

What a connector does between the API and the warehouse

Evaluate a connector against these six responsibilities. Exact behavior depends on the source, destination, sync mode, product version, and configuration; the list is not a guarantee that every connector implements all of them.

• Extraction against the source’s rules. Pagination, rate limits, retry backoff, authentication renewal, and source-specific behavior. Verify supported endpoints and fields, deletion coverage, and the source versions the connector actually tests.
• Incremental state. A stream checkpoint must be coordinated with successful destination work. Check overlap, late updates, duplicate handling, deletion capture, and replay after failure; a timestamp cursor alone does not close every gap.
• Normalization of nested data. Some connectors flatten objects and create child tables; others retain JSON or use destination-specific layouts. Verify parent keys, array positions, and replacement of children when a parent changes.
• Schema inference and evolution. New fields and type conflicts may add columns, coerce values, create variants, quarantine data, or fail the load. The model and dlt example show a variant column; this is not a universal product behavior.
• Write dispositions. Append, replace, and merge serve different input contracts. Merge requires stable keys, deliberate version ordering, and correct child/deletion handling; a primary key alone does not make every rerun safe.
• A run report. Inspect status, extracted and loaded counts, checkpoint, errors, and freshness. A successful incremental run can legitimately have zero new rows. Alerts must be configured and missing deliveries checked separately.

What it does not do is model. The tables it lands are the source’s shape, flattened: raw-layer tables in the architecture article’s terms, with the source’s names, the source’s grain, and the source’s ideas about what a customer is. The core layer, the joins, the definitions, and the history are still the platform’s work, so connector output still needs explicit analytical models. Coverage is also uneven: a catalogue of five hundred connectors says nothing about the quality of the one you need, and an internal API may need a custom source or connector.

Three shapes of managed ingestion

The options overlap. One well-supported product may cover a platform’s needs; combining products is useful only when coverage, deployment boundaries, or economics justify the added operations. Custom connectors and connector builders also exist, so an internal API is not automatically a reason to reject a managed platform.

The bill decides the design

Separate rows extracted, distinct row identities observed, changed rows, and billable usage. The cost exercise uses a hypothetical rule that bills every distinct identity observed during the month. It is not a Fivetran invoice model: Fivetran MAR rules distinguish changes, initial loads, re-syncs, and connector-specific cases, including free unchanged rows in qualifying re-imports. Full scans still add source, network, and destination work even when unchanged rows are not billed. Airbyte plans also differ between volume and capacity pricing. Verify the actual plan and connector before estimating cost.

Self-hosted open source moves the money rather than removing it. A platform that runs connectors needs a host, upgrades, a person who notices when a connector’s new version breaks a stream, and the same person when the source’s API changes and the catalogue’s connector has not caught up. Those hours are the price, and they are easier to overlook than an invoice.

A decision rule, and where it flips

• Buy a hosted connector for a standard source the vendor maintains well, when the row volume prices below an engineer’s time and the data may leave the network.
• Run an open platform for a source only its catalogue covers, or when deployment requirements rule out the available managed or hybrid options, and budget the operations.
• Write with a library for internal systems, unusual authentication, partner APIs, when a full cost and reliability comparison supports owning the code.
• Reassess yearly, because catalogues grow, prices change, and the volume that flipped the decision moves with the business.

Under the exercise’s linear price and labor assumptions, the two annual cost lines intersect at 3.6 million observed identities per month. That is a sensitivity calculation, not a purchasing threshold. Include infrastructure, warehouse work, support, on-call effort, development lead time, security needs, and migration risk. Tiered or capacity pricing may have no single crossover point.

Operating managed ingestion

A bought connector still needs acceptance tests: initial backfill, no-change sync, edits at cursor boundaries, deletes, schema changes, expired credentials, interrupted loads, and recovery. Check freshness and completeness against the source’s expected deliveries rather than only the job’s success flag. Route schema changes and alerts to owners. Restrict credentials and raw data, budget source and destination load, and keep cost per source visible. A staging layer can translate vendor-specific tables into stable names and types, but replacing the connector also requires backfill, state migration, reconciliation, and cutover planning.

The subscription company’s sources, assigned

Anti-patterns

• Hand-writing the payment provider connector. Building before testing whether a supported connector meets the required fields, freshness, and recovery behavior.
• Buying the internal system connector. Expecting a packaged connector without checking custom connector support.
• Full-refresh syncs on row-based pricing. Unnecessary extraction and load work, with billable impact depending on the actual pricing rules.
• Core models on vendor table names. Every mart breaks the day the connector is replaced.
• “It’s managed” as monitoring. The sync failed on the vendor’s side for a week; the completeness check would have said so on day one.
• Self-hosting to save money without staffing it. The licence saved, the operations unpaid, the connector six versions behind.
• Ignoring the variant column. The source changed a type, the connector coped, and the dashboard summed the half that stayed numeric.

Lab

The SQLite model uses complete order objects, an inclusive date cursor, and in-memory source and state. Each sync commits parent and child writes together; on a caught normalization or load failure it restores the schema and leaves the cursor unchanged. Replace reads the full snapshot and clears all known tables even for empty input. Missing child lists mean empty lists in this model, not partial patches. Merge does not detect source deletions, and a timestamp cursor misses updates older than its lower bound. Pagination, backoff, concurrent workers, persistent checkpoints, and crash recovery between database commit and state persistence are outside this model. Run the setup anew before each independent exercise; exercise 6 is standalone.

import json, random, sqlite3, copy, re, math
from collections import Counter, defaultdict
from datetime import date, timedelta

rng = random.Random(0)
source = {"orders": {}, "token": "secret-token"}

def source_day(day):
    """The SaaS source's day: new orders, some edits, and nested customer and item data as its API returns it."""
    d = day.isoformat()
    for _ in range(40):
        n = len(source["orders"]) + 1
        source["orders"][n] = {"order_id": n, "updated_at": d, "amount_cents": rng.randint(500, 9900),
                               "customer": {"id": f"C{rng.randint(1, 60)}", "address": {"city": rng.choice(["Seoul", "Austin", "Berlin"])}},
                               "items": [{"product_id": f"P{rng.randint(1, 9)}", "qty": rng.randint(1, 3)} for _ in range(rng.randint(1, 3))]}
    for n in rng.sample(sorted(source["orders"]), min(6, len(source["orders"]))):
        source["orders"][n]["amount_cents"] += 100
        source["orders"][n]["updated_at"] = d

def api_fetch(token, updated_since):
    if token != source["token"]:
        raise PermissionError("401 invalid token")
    return [json.loads(json.dumps(r)) for r in source["orders"].values() if r["updated_at"] >= updated_since]

class Connector:
    """What a managed connector does between the API and the warehouse: state, normalization, schema evolution, load."""

    def __init__(self, write_disposition="merge"):
        if write_disposition not in {"append", "merge", "replace"}:
            raise ValueError("unsupported write disposition")
        self.wh = sqlite3.connect(":memory:", isolation_level=None)
        self.state, self.schema, self.disposition, self.runs = {"cursor": "0000-00-00"}, defaultdict(dict), write_disposition, []

    def normalize(self, record, table):
        """Nested objects and top-level lists of flat objects; reject other shapes."""
        flat, children = {}, []
        def walk(obj, prefix="", root=False):
            if not isinstance(obj, dict):
                raise ValueError("record must be an object")
            for key, value in obj.items():
                if not isinstance(key, str) or not re.fullmatch(r"[a-z][a-z0-9_]*", key) or "__" in key:
                    raise ValueError("unsupported or ambiguous field name")
                name = prefix + key
                if isinstance(value, dict):
                    walk(value, name + "__")
                elif isinstance(value, list):
                    if not root:
                        raise ValueError("nested lists are outside this model")
                    for position, item in enumerate(value):
                        if not isinstance(item, dict) or any(isinstance(v, (dict, list)) for v in item.values()):
                            raise ValueError("child items must be flat objects")
                        for child_key in item:
                            if not isinstance(child_key, str) or not re.fullmatch(r"[a-z][a-z0-9_]*", child_key) or "__" in child_key:
                                raise ValueError("unsupported child field name")
                        children.append((f"{table}__{key}", dict(item, _parent_id=record["order_id"], _position=position)))
                else:
                    flat[name] = value
        walk(record, root=True)
        return flat, children

    def column_type(self, value):
        if type(value) is int and not -(2 ** 63) <= value < 2 ** 63:
            raise ValueError("integer is outside SQLite signed 64-bit range")
        if type(value) is float and not math.isfinite(value):
            raise ValueError("non-finite number")
        try:
            return {int: "INTEGER", float: "REAL", str: "TEXT", bool: "INTEGER"}[type(value)]
        except KeyError:
            raise ValueError("unsupported scalar value") from None

    def evolve(self, table, row):
        """Add columns the schema has never seen; on a type conflict, add a variant column instead of failing."""
        for column, value in list(row.items()):
            if value is None:
                continue
            kind = self.column_type(value)
            if column not in self.schema[table]:
                self.schema[table][column] = kind
                if self.wh.execute("SELECT count(*) FROM sqlite_master WHERE name = ?", (table,)).fetchone()[0]:
                    self.wh.execute(f"ALTER TABLE {table} ADD COLUMN {column} {kind}")
            elif self.schema[table][column] != kind:
                variant = f"{column}__v_{kind.lower()}"
                row[variant] = row.pop(column)
                self.evolve(table, {variant: row[variant]})

    def load(self, table, rows, key=None):
        if not rows:
            return 0
        for row in rows:
            self.evolve(table, row)
        columns = list(self.schema[table])
        if not self.wh.execute("SELECT count(*) FROM sqlite_master WHERE name = ?", (table,)).fetchone()[0]:
            self.wh.execute(f"CREATE TABLE {table} ({', '.join(f'{c} {t}' for c, t in self.schema[table].items())}" + (f", PRIMARY KEY ({key})" if key and self.disposition == "merge" else "") + ")")
        if self.disposition == "replace":
            self.wh.execute(f"DELETE FROM {table}")
        verb = "INSERT OR REPLACE" if self.disposition == "merge" and key else "INSERT"
        self.wh.executemany(f"{verb} INTO {table} ({', '.join(columns)}) VALUES ({', '.join('?' * len(columns))})",
                            [[row.get(c) for c in columns] for row in rows])
        return len(rows)

    def sync(self, token="secret-token"):
        previous_schema = copy.deepcopy(self.schema)
        try:
            # Replace requires a complete snapshot, including an empty one.
            cursor = "0000-00-00" if self.disposition == "replace" else self.state["cursor"]
            records = api_fetch(token, cursor)
            parents, children, ids = [], defaultdict(list), set()
            for record in records:
                order_id = record.get("order_id")
                if type(order_id) is not int or order_id <= 0 or order_id in ids:
                    raise ValueError("order_id must be a unique positive integer in this snapshot")
                stamp = record.get("updated_at")
                if not isinstance(stamp, str) or date.fromisoformat(stamp).isoformat() != stamp:
                    raise ValueError("updated_at must be YYYY-MM-DD")
                ids.add(order_id)
                flat, kids = self.normalize(record, "orders")
                parents.append(flat)
                for child_table, child in kids:
                    children[child_table].append(child)
            self.wh.execute("BEGIN")
            if self.disposition == "replace":
                for table in self.schema:
                    self.wh.execute(f"DELETE FROM {table}")
            loaded = self.load("orders", parents, key="order_id")
            # Clear old children even when every incoming list is empty or absent.
            child_tables = set(children) | {t for t in self.schema if t.startswith("orders__")}
            for child_table in sorted(child_tables):
                if self.disposition == "merge" and child_table in self.schema:
                    self.wh.executemany(f"DELETE FROM {child_table} WHERE _parent_id = ?", [(n,) for n in ids])
                loaded += self.load(child_table, children.get(child_table, []))
            self.wh.execute("COMMIT")
            if records:
                self.state["cursor"] = max(r["updated_at"] for r in records)
            report = {"status": "succeeded", "rows": loaded, "cursor": self.state["cursor"]}
        except Exception as error:
            if self.wh.in_transaction:
                self.wh.execute("ROLLBACK")
            self.schema = previous_schema
            report = {"status": "failed", "error": str(error), "rows": 0}
        self.runs.append(report)
        return report

first_day = date(2026, 3, 1)
for k in range(3):
    source_day(first_day + timedelta(days=k))
connector = Connector()
print(connector.sync())
# {'status': 'succeeded', 'rows': 353, 'cursor': '2026-03-03'}
print({t: connector.wh.execute(f"SELECT count(*) FROM {t}").fetchone()[0] for t in connector.schema})
# {'orders': 120, 'orders__items': 233}
print(connector.schema["orders"])
# {'order_id': 'INTEGER', 'updated_at': 'TEXT', 'amount_cents': 'INTEGER', 'customer__id': 'TEXT', 'customer__address__city': 'TEXT'}

One sync, 120 orders and their 233 line items in two tables, and a schema the connector inferred: the customer object became two prefixed columns and the items list became a child table.

1. Incremental state and write dispositions. Sync again with nothing new at the source, then after another day, and watch the cursor and the row counts. Then run the same two syncs through a connector whose write disposition is append.

print("second sync with nothing new at the source:", connector.sync())
# second sync with nothing new at the source: {'status': 'succeeded', 'rows': 135, 'cursor': '2026-03-03'}
print("rows after it:", {t: connector.wh.execute(f"SELECT count(*) FROM {t}").fetchone()[0] for t in connector.schema})
# rows after it: {'orders': 120, 'orders__items': 233}
source_day(first_day + timedelta(days=3))
print("sync after another day at the source:     ", connector.sync())
# sync after another day at the source:      {'status': 'succeeded', 'rows': 272, 'cursor': '2026-03-04'}
print("rows after it:", {t: connector.wh.execute(f"SELECT count(*) FROM {t}").fetchone()[0] for t in connector.schema}, "source has", len(source["orders"]), "orders")
# rows after it: {'orders': 160, 'orders__items': 322} source has 160 orders

appender = Connector(write_disposition="append")
appender.sync()
appender.sync()
print("the same two syncs with write disposition append:", appender.wh.execute("SELECT count(*), count(DISTINCT order_id) FROM orders").fetchone())
# the same two syncs with write disposition append: (205, 160)

2. Schema drift. Have the source start sending a new field and the amount as a decimal string for three orders. Sync, and inspect the schema and the affected rows.

drifted = Connector()
drifted.sync()
for n in (1, 2, 3):
    source["orders"][n]["channel"] = "app"
    source["orders"][n]["amount_cents"] = f"{source['orders'][n]['amount_cents'] / 100:.2f}"
    source["orders"][n]["updated_at"] = "2026-03-09"
report = drifted.sync()
print("sync after the source changed its shape:", report["status"], "cursor", report["cursor"])
# sync after the source changed its shape: succeeded cursor 2026-03-09
print("orders columns now:", drifted.schema["orders"])
# orders columns now: {'order_id': 'INTEGER', 'updated_at': 'TEXT', 'amount_cents': 'INTEGER', 'customer__id': 'TEXT', 'customer__address__city': 'TEXT', 'amount_cents__v_text': 'TEXT', 'channel': 'TEXT'}
print(drifted.wh.execute("SELECT order_id, amount_cents, amount_cents__v_text, channel FROM orders WHERE order_id <= 4 ORDER BY order_id").fetchall())
# [(1, None, '69.11', 'app'), (2, None, '84.61', 'app'), (3, None, '27.81', 'app'), (4, 2907, None, None)]
print("rows where the amount moved to the variant column:", drifted.wh.execute("SELECT count(*) FROM orders WHERE amount_cents IS NULL").fetchone()[0])
# rows where the amount moved to the variant column: 3

3. Normalization. Sync a fresh connector and read back one order across the parent and child tables. Check that every order’s item count in the child table equals the source’s.

fresh = Connector()
fresh.sync()
print(fresh.wh.execute("SELECT order_id, customer__id, customer__address__city, updated_at FROM orders WHERE order_id IN (10, 11) ORDER BY order_id").fetchall())
# [(10, 'C7', 'Austin', '2026-03-01'), (11, 'C52', 'Berlin', '2026-03-01')]
print(fresh.wh.execute("SELECT _parent_id, _position, product_id, qty FROM orders__items WHERE _parent_id = 10 ORDER BY _position").fetchall())
# [(10, 0, 'P5', 3), (10, 1, 'P2', 3), (10, 2, 'P6', 3)]
rebuilt = {row[0]: row[1] for row in fresh.wh.execute("SELECT _parent_id, count(*) FROM orders__items GROUP BY 1").fetchall()}
print("orders whose item count in the child table differs from the source:",
      sum(len(source["orders"][n]["items"]) != rebuilt.get(n, 0) for n in source["orders"]))
# orders whose item count in the child table differs from the source: 0
print("the source's nested record:", json.dumps(source["orders"][10])[:120] + "...")
# the source's nested record: {"order_id": 10, "updated_at": "2026-03-01", "amount_cents": 8516, "customer": {"id": "C7", "address": {"city": "Austin"...

4. The bill. Simulate thirty daily syncs over a source with a million historical rows, two thousand new rows a day, and three hundred edits a day, once incrementally and once as a full refresh. Count rows moved and rows touched at least once, apply a hypothetical price per million distinct observed identities, and compare with an illustrative cost of building.

def month_of_syncs(mode, history=1_000_000, new_per_day=2000, edits_per_day=300):
    """Thirty daily syncs over a source with a million historical rows; count rows moved and rows touched at least once."""
    if mode not in {"incremental", "full refresh"}:
        raise ValueError("unsupported sync mode")
    if any(type(x) is not int or x < 0 for x in (history, new_per_day, edits_per_day)):
        raise ValueError("counts must be nonnegative integers")
    if edits_per_day > history + new_per_day:
        raise ValueError("too many edits for the initial population")
    r = random.Random(7)
    last_change = {n: -1 for n in range(1, history + 1)}
    touched, rows_moved = set(), 0
    for day in range(30):
        for _ in range(new_per_day):
            last_change[len(last_change) + 1] = day
        for n in r.sample(range(1, len(last_change) + 1), edits_per_day):
            last_change[n] = day
        rows = [n for n, changed in last_change.items() if changed == day] if mode == "incremental" else list(last_change)
        touched.update(rows)
        rows_moved += len(rows)
    return len(last_change), rows_moved, len(touched)

active_rows = {}
for mode in ("incremental", "full refresh"):
    size, moved, active_rows[mode] = month_of_syncs(mode)
    print(f"{mode:13}: {size:,} rows at the source, {moved:,} rows moved in the month, {active_rows[mode]:,} distinct identities observed")
# incremental  : 1,060,000 rows at the source, 68,983 rows moved in the month, 68,720 distinct identities observed
# full refresh : 1,060,000 rows at the source, 30,930,000 rows moved in the month, 1,060,000 distinct identities observed
price = 500
print(f"at an illustrative {price} per million observed identities: incremental {active_rows['incremental'] * price / 1e6:,.0f} a month,",
      f"full refresh {active_rows['full refresh'] * price / 1e6:,.0f} a month, for one source")
# at an illustrative 500 per million observed identities: incremental 34 a month, full refresh 530 a month, for one source
build_hours, rate = 240, 90
print(f"building and keeping one connector for a year at an illustrative {build_hours} hours and {rate} an hour is {build_hours * rate:,},",
      f"which the managed incremental connector reaches at {build_hours * rate / 12 / price:,.1f} million observed identities a month")
# building and keeping one connector for a year at an illustrative 240 hours and 90 an hour is 21,600, which the managed incremental connector reaches at 3.6 million observed identities a month

5. Failure visibility. Rotate the source’s token without telling the connector, sync, and compare the result with a custom job that catches the error and returns zero rows.

source["token"] = "rotated-token"
report = Connector().sync(token="secret-token")
print("managed connector with a stale credential:", report)
# managed connector with a stale credential: {'status': 'failed', 'error': '401 invalid token', 'rows': 0}

def custom_job(token):
    try:
        return len(api_fetch(token, "0000-00-00"))
    except Exception:
        return 0

print("a custom job that swallows the error loads", custom_job("secret-token"), "rows and exits 0")
# a custom job that swallows the error loads 0 rows and exits 0
source["token"] = "secret-token"
print("after the credential is fixed:", Connector().sync()["status"])
# after the credential is fixed: succeeded

6. The same source through dlt. Install dlt[duckdb] in a separate environment and run the pipeline below, which declares a resource with a primary key, a merge disposition, and an incremental cursor, loads two nested orders into DuckDB, then changes one order, adds one with a new field and a string amount, and runs again. The example was verified with dlt 1.30.0 and DuckDB 1.5.5. Use those versions to reproduce the recorded output; temporary data and pipeline state are removed on normal completion.

import os, tempfile
from pathlib import Path
os.environ["RUNTIME__DLTHUB_TELEMETRY"] = "false"
import dlt, duckdb

with tempfile.TemporaryDirectory(prefix="de012-dlt-") as work:
    SOURCE = [
        {"order_id": 1, "customer": {"id": "C1", "country": "US"}, "amount_cents": 6300, "updated_at": "2026-03-01",
         "items": [{"product_id": "P1", "qty": 1}, {"product_id": "P3", "qty": 2}]},
        {"order_id": 2, "customer": {"id": "C2", "country": "KR"}, "amount_cents": 1500, "updated_at": "2026-03-01",
         "items": [{"product_id": "P2", "qty": 1}]},
    ]

    @dlt.resource(name="orders", primary_key="order_id", write_disposition="merge")
    def orders(updated_at=dlt.sources.incremental("updated_at", initial_value="2026-01-01", primary_key=())):
        for row in SOURCE:
            if row["updated_at"] >= updated_at.start_value:
                yield row

    database = str(Path(work) / "shop.duckdb")
    pipeline = dlt.pipeline(pipeline_name="shop", pipelines_dir=str(Path(work) / "pipelines"),
                            destination=dlt.destinations.duckdb(database), dataset_name="raw")
    info = pipeline.run(orders())
    print("first run:", info.load_packages[0].state, "jobs:", len(info.load_packages[0].jobs["completed_jobs"]))
    con = duckdb.connect(database)
    print(con.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = 'raw' ORDER BY 1").fetchall())
    print(con.execute("SELECT order_id, customer__id, customer__country, amount_cents FROM raw.orders ORDER BY 1").fetchall())
    print(con.execute("SELECT product_id, qty FROM raw.orders__items ORDER BY 1").fetchall())
    print("cursor state:", pipeline.state["sources"]["shop"]["resources"]["orders"]["incremental"]["updated_at"]["last_value"])

    con.close()
    SOURCE[0]["amount_cents"] = 6800
    SOURCE[0]["updated_at"] = "2026-03-02"
    SOURCE.append({"order_id": 3, "customer": {"id": "C1", "country": "US"}, "amount_cents": "42.00", "updated_at": "2026-03-02",
                   "items": [], "channel": "app"})
    pipeline.run(orders())
    con = duckdb.connect(database)
    print(con.execute("SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = 'raw' AND table_name = 'orders' ORDER BY ordinal_position").fetchall())
    print(con.execute("SELECT order_id, amount_cents, amount_cents__v_text, channel FROM raw.orders ORDER BY 1").fetchall())
    print("rows in orders:", con.execute("SELECT count(*) FROM raw.orders").fetchone()[0], "after merging the second run")
    con.close()
    SOURCE[0]["amount_cents"] = 7100
    SOURCE[0]["items"] = []
    pipeline.run(orders())
    con = duckdb.connect(database)
    print("same-date edit:", con.execute("SELECT order_id, amount_cents FROM raw.orders WHERE order_id=1").fetchall())
    print("items after same-date edit:", con.execute("SELECT product_id, qty FROM raw.orders__items ORDER BY 1").fetchall())
    con.close()

first run: loaded jobs: 4
[('_dlt_loads',), ('_dlt_pipeline_state',), ('_dlt_version',), ('orders',), ('orders__items',)]
[(1, 'C1', 'US', 6300), (2, 'C2', 'KR', 1500)]
[('P1', 1), ('P2', 1), ('P3', 2)]
cursor state: 2026-03-01
[('order_id', 'BIGINT'), ('amount_cents', 'BIGINT'), ('updated_at', 'VARCHAR'), ('customer__id', 'VARCHAR'), ('customer__country', 'VARCHAR'), ('_dlt_load_id', 'VARCHAR'), ('_dlt_id', 'VARCHAR'), ('amount_cents__v_text', 'VARCHAR'), ('channel', 'VARCHAR')]
[(1, 6800, None, None), (2, 1500, None, None), (3, None, '42.00', 'app')]
rows in orders: 3 after merging the second run
same-date edit: [(1, 7100)]
items after same-date edit: [('P2', 1)]

7. Write the recommendation. For the subscription company’s sources, write the one-page build-versus-buy decision: the shape chosen per source, the volumes assumed, the prices used, the volume at which each choice flips, and the date to reassess.

References

Documentation checked on 2026-09-12. Pricing and deployment options depend on the current plan and supported connector. The cost exercise uses invented constants; the local library example was run with dlt 1.30.0 and DuckDB 1.5.5.

• Fivetran usage-based pricing and MAR exceptions
• Fivetran deployment models
• Airbyte sync modes
• Airbyte pricing plans
• dlt incremental loading
• dlt merge loading
• dlt schema evolution