Introducing The COPY Skippr Pipeline
October 2026
Yep, that's right. Skippr's COPY Pipeline Type is literally the cp of the data integration world.
And it amazes me that I've only just added such a "simple" feature.
This week, a client (Series A UK tech startup) needed to address some GDPR compliance issues in their S3 data archive.
Their event sourcing platform lands nested JSON from field devices into one S3 prefix all day. It's their source of truth (of last resort). Unfortunately, AWS Firehose has been dumping the data with mixed entity types in each object. For that data, the only prefix partitioning is by datetime (year=2021/month=... etc.) — count your blessings, I suppose.
The data is ingested by Skippr ELT pipelines to build the client's Athena data lake. A frustration has been that it's not possible to replay a single event type without parsing all the source data due to the mixed entity types in each object. Particularly painful on the old Hive tables in AWS Glue… where ALTER DDL statements essentially are not supported. Thankfully, we're migrating them to Iceberg... more in a future post.
The ask was to optionally apply custom S3 lifecycle events to each tenant's data — as each tenant has their own interpretation of GDPR (welcome to Europe). The bucket is large: billions of objects, hundreds of billions of records, and still growing.
They are not asking for the world here. No new typed warehouse table. Nothing changes in Iceberg. No flattening or cleansing. Not a SQL reshape of nested payloads. Just copy the JSON, put it in folders by tenant and event type so it's easy to find, ingest, and delete in future. Just copy the data, right…!?
I had anticipated simply pointing Skippr or something else at the source data and hey presto. It was not to be, as we will learn. Skippr has sophisticated Ingest and CDC Pipeline Types. Ingest infers a schema, auto-fixes format errors, technically cleanses data, and writes Arrow. CDC does all that, plus, well... CDC (Change Data Capture). So while Skippr is tremendously powerful, much of that got in the way of simply lifting and shifting the JSON records. So we added a third pipeline type: Copy.
The Naive Script
The naive approach to this problem would be to write something like the following. List the prefix, peek at a couple of fields, batch into JSONL, put_object under a new layout.
import json
from collections import defaultdict
import boto3
src = boto3.client("s3")
dst = boto3.client("s3")
batches = defaultdict(list)
token = None
while True:
kw = {"Bucket": "devices", "Prefix": "raw/"}
if token:
kw["ContinuationToken"] = token
page = src.list_objects_v2(**kw)
for obj in page.get("Contents", []):
body = src.get_object(Bucket="devices", Key=obj["Key"])["Body"].read()
rec = json.loads(body)
part = f"p_tenant_id={rec['tenant_id']}/p_year={rec['ts'][:4]}/…"
batches[part].append(body.rstrip() + b"\n")
token = page.get("NextContinuationToken")
if not token:
break
for part, rows in batches.items():
dst.put_object(Bucket="lake", Key=f"landing/{part}/batch.jsonl", Body=b"".join(rows))A few lines of Python that work on my laptop. It immediately fails, of course. Crash halfway through the listing and you do not know which objects you already wrote. A retry either duplicates JSONL or you're inventing a homegrown checkpointing system. Try to hold every partition in memory or local disk until the prefix is exhausted and years of Firehose won't fit. Suddenly you're faced with building a data integration platform... to copy some JSON about.
Still, scratch further through the surface, and the actual work required for the client ask is even harder than it first appears. Years of Firehose JSON with breaking schema changes, only date prefixes, mixed entity types in each object, no universal field naming convention to route on values. You have to actually read event_type, a timestamp, and detail.tenant_id (or tenant_id, tid, truck.tenant_id, device.tenant_id or something_else, depending on the event) to pick the destination object key — without reshaping the JSON. The JSON record body that lands MUST be byte-equal to the body you read. Then you write it, in efficient batches to:
dest/{event_type}/p_tenant_id=…/p_year=…/p_month=…/p_day=…
Exact same JSON bytes. New layout. Bucket to bucket. The raw Firehose prefixes must stay untouched to preserve the archive; we're looking for a parallel cut-over migration where we must lose zero records, zero byte-drift, across billions of events.
And if you flush every minute you get millions of tiny JSONL files. Ideally the copy should buffer on a day scale so each slice grows into a large object. Your PUTs will lag GETs by a tremendous degree. How will you buffer that data? How will you greedily consume S3 prefixes with thousands of concurrent GET requests?
Skippr already had the tricky engine pieces: S3 source offsets so a restart does not re-read committed objects, pending data sink manifests so a retry does not double-write the same JSONL, a multi-threaded work-stealing scheduler to keep a 32-core Spot box (c7i.8xlarge) busy.
Ironically, we were missing the simpler pipeline mechanics that would passthrough the JSON instead of invoking all the default Skippr schema inference magic.
Why Didn't You Just Use [my favourite tool]
Because I like to build, use and consult with my favourite tool... obviously 😉
It's important though to evaluate alternative tooling to Skippr. Partly as a duty to provide the client with the best possible solution, and also to promote Skippr's virtues in this blog (see what I did there!?).
Let me say upfront, I love working with most of these tools; they are exquisite examples of engineering and product. However, each one misses on delivering byte-equal JSON in new prefixes from an S3 bucket listing.
DuckDB COPY … TO 's3://…' (FORMAT JSON, PARTITION_BY …) is a fine one-shot SELECT - very elegant and ergonomic. It parses JSON into typed columns and serialises again. The trouble is nested objects become VARCHAR or STRUCT, the bytes change! There is no durable cursor over years of S3 prefixes. It just globs the query’s file list. Rerun the query and you rewrite the destination, duplicating data, maybe also adding new records... who knows, no way to tell. The query is memory bound, not a pipeline that can buffer each event_type and write out large object chunks. If the job is “reshape this moderately sized snapshot once,” DuckDB can do it. If the job is “keep copying a live archive, byte-equal, into p_tenant_id= folders,” DuckDB's not the tool — it's not designed to be.
AWS Glue / Spark is the natural next option given the client already uses Athena. But like DuckDB (and Skippr prior to 18.2.0 COPY Pipeline Type), crawlers and from_json force a typed schema. DynamicFrames flatten or stringify nests — definitely not an immutable byte-equal copy. Worse, you pay for a Spark cluster ($$$ infra) to shuffle bytes you did not want typed, then you own jobs ($$$ flesh and blood), IAM, bookmarking, and all the recovery paths when the JSON is “wrong” (which it always is all the time). How would you implement bookmarking per record within an object? (Clue: months at your desk with your head in your hands, wondering how you got yourself into this and when it will end.)
Databricks Autoloader looks closer. CloudFiles can incrementally ingest a prefix, which is the right shape of listing. The default sink is Delta or Parquet. JSON into a DataFrame still types the row. You freeze cloudFiles.schemaEvolutionMode and primitivesAsString and you still re-encode. Checkpointing is Spark offsets, not Skippr’s “this object, this JSONL key, once.” A Series A team now owns a Databricks workspace to move JSON to JSON.
S3 Batch / DistCp / aws s3 cp / sync all copy S3 object keys. They will bit-copy year=2021/month=10/…/firehose-xxxxx to another bucket with the same key. They cannot open the object, read event_type and detail.tenant_id, and write dest/{event_type}/p_tenant_id=…/p_year=…. Some can incrementally skip “newer than this mtime,” so at least you can repeatedly get it wrong without idempotence.
Airbyte, Fivetran, Secor (a source of great inspiration to Skippr) want a warehouse table as a sink. Not what we're trying to do here.
Streaming, ELT, databases, decoupled query engines, object stores, modelling tools, vendor-specific ingress pipelines, and VC-funded marketing budgets. None of them will list years of JSON objects on S3, peek at a couple of nested fields, and write the same bytes under a new key.
Told you this was a conceptually simple, yet surprisingly difficult problem.
Skippr Couldn't Do It Either
I've spent years toiling on Skippr, agonising over the most interesting and difficult data integration problems. Discover and evolve backwards-compatible schemas in streams, auto-tune scaling, SQL query on the WAL stream, land crafted Parquet with optimised row groups into any warehouse... Hell, build your lineage on the fly. CDC sir? Absolutely sir?
Yet incredibly, until 18.2.0, Skippr couldn't support simply bit-copying the source data and reorganising its disk layout a bit.
The existing Ingest pipeline type is JSON → Arrow (recursive nested schemas and all) → Parquet (typically). So our simple “copy” requirement paid the full CPU overhead for serde and type inference, to mutate the data we explicitly want to be immutably copied. The Skippr S3 source plugin could already remember which objects and records it had already processed. Phew, so you can see why I was tempted to believe Skippr was a shoo-in. But the bytes going through Skippr's WAL were Arrow, not the original JSON.
Welcome pipeline.type: copy (Skippr 18.2.0) pipeline type. Decode enough to build the destination S3 key and write the exact record-level bytes.
What COPY Looks Like Now
The skippr.yml type: copy config option (PipelineType.COPY in Python) acts as a passthrough of all the normal Skippr ingestion logic but maintains the exactly-once and recovery semantics of the Skippr WAL. The source reads JSON (or CSV, or XML) and keeps the source data byte-equal.
The client use case was limited to JSON, but there's support for CSV and XML. If you configure json → csv, nested fields are compacted to JSON strings in each CSV cell.
The copy pipeline peeks at fields to build the output prefixes: namespace_fields (event_type), batch_partition_fields (detail.tenant_id,truck.tenant_id, etc.), and batch_time_fields. Ideally I'd take lessons from 1BRC and improve the peek to raw byte slices rather than string parsing… another day, another prompt.
While reorganising the layout of these JSON records, we also want to take the opportunity to use the longer-standing Skippr feature buffer_threshold_seconds. Setting it to 86400 so a historical copy grows into large JSONL slices at the S3 sink, solving the client's existing tiny-object problem.
Public contract: pipeline type, S3 source, S3 destination.
Walk through
Let's look at how we'd configure Skippr via the Python SDK.
Or, you can use yaml or pure cli flags if you prefer (Docs).
from skippr import (
BatchTimeUnit,
Config,
DataSinkS3,
DataSourceS3,
LocalStorage,
Pipeline,
PipelineType,
Transform,
)
cfg = Config().workspace("copy-demo").storage(LocalStorage())
src = cfg.data_source(
"sample",
DataSourceS3(s3_bucket="my-source-bucket", s3_prefix="ill-just-dump-this-here"),
)
landing = cfg.data_sink(
"landing",
DataSinkS3(
s3_bucket="your-output-bucket",
s3_prefix="copy/phew-thats-a-month-and-a-gdpr-fine-saved",
format="jsonl",
),
)
cfg.pipeline(
"gdpr_compliant_migration",
Pipeline(
data_source=src,
data_sink=landing,
type=PipelineType.COPY,
buffer_threshold_seconds=86400,
transform=Transform(
batch_partition_fields="detail.tenant_id,truck.tenant_id",
batch_time_fields="time,system_datetime,date",
batch_time_unit=BatchTimeUnit.DAY,
time_partition_prefix="p_",
),
),
)
cfg.save("skippr.yml")skippr:
workspace: copy-demo
skipprd_el_storage_mode: local
pipelines:
gdpr_compliant_migration:
type: copy
buffer_threshold_seconds: 86400
data_source: data_sources.sample
data_sink: data_sinks.landing
transform:
batch_partition_fields: detail.tenant_id,truck.tenant_id
batch_time_fields: time,system_datetime,date
batch_time_unit: day
time_partition_prefix: p_
data_sources:
sample:
S3:
s3_bucket: my-source-bucket
s3_prefix: ill-just-dump-this-here
data_sinks:
landing:
S3:
s3_bucket: your-output-bucket
s3_prefix: copy/phew-thats-a-month-and-a-gdpr-fine-saved
format: jsonlpip install skippr==18.2.1
skipprd sync --pipeline gdpr_compliant_migration --onceThat's it! Objects with bit-equal records land under s3://your-output-bucket/copy/phew-thats-a-month-and-a-gdpr-fine-saved/<namespace>/p_detail.tenant_id=…/p_year=…/p_month=…/p_day=…/apply-….jsonl (or p_truck.tenant_id=… when that field is present). time_partition_prefix: p_ is what turns year= into p_year=.
Run --once again — it duplicates nothing. Skippr already recorded those source objects; you get exactly-once replication.
Drop a new object into the prefix. Another --once copies only that object — even if it's a needle in a haystack of billions.
When to Use Each Pipeline Type
copy is immutable passthrough: bucket to bucket, file to bucket, bucket to Kafka (you get the idea). Byte-equal JSON, new prefix layout, peek at fields only to choose the key.
Ingest will parse into Arrow and a warehouse, database, or other schema'd system. It discovers and evolves schemas, writes Parquet to Snowflake, Athena, Iceberg, etc.
CDC is ingest plus change data capture from compatible sources and sinks.
All three can stream to and from file, S3, Kafka, stdout, HTTP, MQTT — you name it.
If you need your source payloads intact, Hive folder layouts you can lifecycle per tenant, and an idempotent replay, that is COPY. If you need a queryable table, that is ingest.
More in the engine docs. Python enums in the Python guide. Engine 18.2.1 is on PyPI: pip install skippr==18.2.1.