Exporting cold partitions to S3 before expiry
A rail operator runs axle-bearing acceleration and temperature sensors across a fleet of freight wagons, and safety regulators require every raw reading to be retrievable for seven years after collection. Keeping seven years of high-rate vibration data live in InfluxDB is neither affordable nor necessary: the query workload only ever touches the last few weeks, so the raw data lands in a hot bucket, is downsampled, and ages into a bearing-cold bucket with a short retention window. The problem is the last mile — once a day falls outside that window, the retention sweeper drops the shard group whole and the regulatory copy is gone forever. You cannot rely on retention alone as an archive, and you cannot lengthen retention to seven years without paying to keep cold data hot. The fix is a scheduled export job that pages the oldest un-archived day out to object storage, proves the copy is intact, and records a watermark before expiry is allowed to reclaim it. This page builds that job concretely, following the staged model in Cold-Storage Archival to Object Store.
The ordering constraint is the whole game. If $t_{export}$ is the timestamp of the newest day the archiver has successfully written to S3 and $t_{expiry}$ is the retention boundary (now minus the cold bucket’s retention window), the invariant you must never violate is:
$$ t_{export} \geq t_{expiry} + \Delta_{margin} $$
The export watermark has to lead the expiry boundary by a safety margin $\Delta_{margin}$ — at least one shard-group duration plus one export interval — so that no day can age past the boundary before the archiver has committed it to object storage and verified it.
Prerequisites
Solution walkthrough
Step 1 — Find the oldest un-archived day
The job must be self-scheduling: each run archives exactly one calendar day — the oldest day present in bearing-cold that has not already been recorded in the ledger. Reading the watermark from the ledger rather than trusting the schedule keeps the job idempotent and lets it catch up if a run was skipped. Query the ledger for the last committed day, then target the next one.
import os
from datetime import datetime, timedelta, timezone
from influxdb_client import InfluxDBClient
INFLUX_URL = os.environ["INFLUX_URL"]
COLD_TOKEN = os.environ["INFLUX_COLD_TOKEN"]
ORG = "rail-telemetry"
def last_archived_day(client: InfluxDBClient) -> datetime | None:
flux = '''
from(bucket: "archive-ledger")
|> range(start: -3y)
|> filter(fn: (r) => r._measurement == "export_watermark")
|> filter(fn: (r) => r._field == "day_epoch")
|> last()
'''
tables = client.query_api().query(flux, org=ORG)
for table in tables:
for rec in table.records:
return datetime.fromtimestamp(rec.get_value(), tz=timezone.utc)
return None
def next_day_to_export(client: InfluxDBClient) -> datetime:
watermark = last_archived_day(client)
if watermark is None:
# Bootstrap: start at the oldest data in the cold bucket.
flux = '''
from(bucket: "bearing-cold")
|> range(start: -35d)
|> filter(fn: (r) => r._measurement == "bearing_accel")
|> keep(columns: ["_time"])
|> first()
'''
for table in client.query_api().query(flux, org=ORG):
for rec in table.records:
t = rec.get_time().astimezone(timezone.utc)
return t.replace(hour=0, minute=0, second=0, microsecond=0)
raise RuntimeError("bearing-cold is empty; nothing to archive")
return watermark + timedelta(days=1)
Anchoring each export to a UTC calendar day (midnight-to-midnight) makes the S3 key layout deterministic and the row-count check meaningful. The three-year range() on the ledger is a cheap bound — the watermark is a single point, but Flux needs a start time.
Step 2 — Page the day out as line protocol into a date-partitioned S3 key
Export one day, streamed in time-ordered pages so a busy day never has to fit in memory at once. Reconstruct InfluxDB line protocol from each record — it is the loss-free, restore-ready format — and write it to a key partitioned by an RFC 3339 calendar date. A date-partitioned key (year=/month=/day=) is what makes a seven-year archive browsable and lets a restore fetch exactly one day without scanning the whole prefix.
import gzip, io
import boto3
from influxdb_client import InfluxDBClient
S3 = boto3.client("s3")
ARCHIVE_BUCKET = "rail-bearing-archive"
def escape(v: str) -> str:
return v.replace(" ", "\\ ").replace(",", "\\,").replace("=", "\\=")
def day_to_line_protocol(client: InfluxDBClient, day: datetime) -> tuple[bytes, int]:
start = day.isoformat()
stop = (day + timedelta(days=1)).isoformat()
flux = f'''
from(bucket: "bearing-cold")
|> range(start: {start}, stop: {stop})
|> filter(fn: (r) => r._measurement == "bearing_accel")
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
'''
buf = io.BytesIO()
rows = 0
with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
for record in client.query_api().query_stream(flux, org=ORG):
tags = f'wagon_id={escape(record["wagon_id"])},axle={escape(record["axle"])}'
ns = int(record.get_time().timestamp() * 1e9)
fields = f'rms={record["rms"]},temp_c={record["temp_c"]}'
gz.write(f'bearing_accel,{tags} {fields} {ns}\n'.encode())
rows += 1
return buf.getvalue(), rows
def s3_key(day: datetime) -> str:
return (f"bearing_accel/year={day:%Y}/month={day:%m}/day={day:%d}/"
f"bearing_accel-{day:%Y-%m-%d}.lp.gz")
def export_day(client: InfluxDBClient, day: datetime) -> tuple[str, int]:
payload, rows = day_to_line_protocol(client, day)
if rows == 0:
raise RuntimeError(f"No rows for {day:%Y-%m-%d}; refusing to write empty archive")
key = s3_key(day)
S3.put_object(
Bucket=ARCHIVE_BUCKET, Key=key, Body=payload,
ContentType="application/x-influxdb-line-protocol",
ContentEncoding="gzip",
Metadata={"row_count": str(rows), "source_day": f"{day:%Y-%m-%d}"},
)
return key, rows
query_stream yields records one at a time so the wagon fleet’s readings are gzipped as they arrive rather than materialised into a giant list. Stamping row_count into the object’s user metadata lets Step 3 verify without re-reading the body, and refusing to write an empty archive prevents a query bug from committing a zero-row watermark that would later let a real day expire unarchived.
Step 3 — Verify the object, then commit the watermark
The export is only trustworthy once you have proven the object exists and holds the row count you exported. Re-read the source count straight from InfluxDB, compare it against what landed in S3 via a HEAD request, and only on an exact match write the watermark to the ledger. This is the gate that authorises expiry.
def source_row_count(client: InfluxDBClient, day: datetime) -> int:
flux = f'''
from(bucket: "bearing-cold")
|> range(start: {day.isoformat()}, stop: {(day + timedelta(days=1)).isoformat()})
|> filter(fn: (r) => r._measurement == "bearing_accel" and r._field == "rms")
|> count()
'''
for table in client.query_api().query(flux, org=ORG):
for rec in table.records:
return rec.get_value()
return 0
def verify_and_commit(client: InfluxDBClient, day: datetime, key: str, exported_rows: int):
head = S3.head_object(Bucket=ARCHIVE_BUCKET, Key=key)
stored_rows = int(head["Metadata"]["row_count"])
source_rows = source_row_count(client, day)
if not (exported_rows == stored_rows == source_rows):
raise RuntimeError(
f"Row-count mismatch for {day:%Y-%m-%d}: "
f"source={source_rows} exported={exported_rows} s3={stored_rows}; "
f"NOT committing watermark, day stays eligible for re-export")
point = (f"export_watermark day_epoch={int(day.timestamp())}i,"
f"rows={source_rows}i,etag=\"{head['ETag'].strip('\"')}\" "
f"{int(datetime.now(timezone.utc).timestamp() * 1e9)}")
client.write_api().write(bucket="archive-ledger", org=ORG, record=point)
Because the watermark is written only after the three counts agree, a failure anywhere upstream leaves the day un-committed and eligible for a clean re-export on the next run. Overwriting the same deterministic S3 key with the same day’s data is safe and idempotent — object versioning keeps the prior copy while the ledger advances exactly once. For hardening the write path with retries, honoring Retry-After, and dead-lettering, see Python client orchestration patterns.
Step 4 — Restore a day from the archive
An archive you cannot restore is not an archive. When an audit or investigation needs a specific day back, fetch the object, decompress it, and replay the line protocol into a temporary bucket with the InfluxDB write API — timestamps are preserved because you stored nanosecond precision.
def restore_day(client: InfluxDBClient, day: datetime, target_bucket: str):
obj = S3.get_object(Bucket=ARCHIVE_BUCKET, Key=s3_key(day))
lines = gzip.decompress(obj["Body"].read()).decode()
client.write_api().write(bucket=target_bucket, org=ORG, record=lines.splitlines())
print(f"Restored {day:%Y-%m-%d} into {target_bucket}")
Gotchas and edge cases
Committing the watermark before verifying is silent, permanent data loss. If the job writes the watermark first and the PutObject later fails, the ledger says the day is safe and the retention sweeper will reclaim it — with nothing in S3. Order is non-negotiable: export, HEAD-verify the row count, and only then advance the watermark. Any exception between those steps must leave the day un-committed so the next run redoes it. This ordering discipline is the same one that governs retention policy design — expiry always trails a proven downstream copy, never leads it.
A too-short cold retention window can outrun a lagging archiver. If the job falls behind — a long outage, a throttled S3 endpoint — days keep aging toward the boundary while the watermark stalls. Size the cold bucket’s retention so it holds several days more than the export interval (the $\Delta_{margin}$ term above), and alert when now − watermark exceeds that margin so an operator intervenes before a day is lost rather than after.
Non-deterministic S3 keys break idempotent re-export. Keying on the wall-clock run time instead of the data’s calendar day means a re-run writes a second, differently-named object for the same day, and the row-count check compares against the wrong file. Derive the key purely from the day being exported so a re-export overwrites the identical key, and let bucket versioning — not a new key — preserve history.
Verification
After a run, confirm the watermark advanced and leads the retention boundary with comfortable margin. Compare the ledger watermark against the oldest day still present in the cold bucket:
watermark = from(bucket: "archive-ledger")
|> range(start: -3y)
|> filter(fn: (r) => r._measurement == "export_watermark" and r._field == "day_epoch")
|> last()
|> yield(name: "last_archived_day_epoch")
from(bucket: "bearing-cold")
|> range(start: -40d)
|> filter(fn: (r) => r._measurement == "bearing_accel")
|> first()
|> keep(columns: ["_time"])
|> yield(name: "oldest_live_day")
The oldest live day in bearing-cold should sit newer than or equal to the watermark — never older. An oldest-live day that precedes the watermark means a day aged out without being archived, which is a page-out-of-bed alert, not a warning. Cross-check the object landed with aws s3api head-object --bucket rail-bearing-archive --key <key> and confirm the row_count metadata matches the source count.
FAQ
Why export line protocol instead of CSV or Parquet?
Line protocol is InfluxDB’s native, loss-free serialisation: it round-trips measurement, tags, fields, and nanosecond timestamps back through the write API with no schema mapping. CSV loses type fidelity and Parquet needs a column schema you must maintain as fields evolve. For a compliance archive whose only job is faithful restore, line protocol is the lowest-risk format; convert to Parquet later for analytics if a query engine needs it.
How do I archive many measurements or a very large fleet per day?
Run the export per measurement (and, for very high cardinality, per wagon-group shard) so each object stays a manageable size and each row-count check is scoped. Parallelise the branches with the fan-out patterns from the orchestration section, but keep a single ledger watermark per partition so each stream advances independently and no partition can expire ahead of its own archive.
Can the retention sweeper and the archiver race?
Only if the margin is too thin. Because expiry acts on whole shard groups and the archiver commits whole calendar days, keeping the cold retention window longer than one shard-group duration plus the export interval guarantees the sweeper never reaches a day the archiver has not already committed. Monitor the watermark-to-boundary gap and treat a shrinking gap as the early warning.
Related
- Cold-Storage Archival to Object Store — the staged archival model this export job implements, and how object-store tiers fit the overall lifecycle.
- Retention Policy Design — sizing the cold bucket’s expiry window so it always trails the export watermark.
- Python Client Orchestration Patterns — retry, backoff, and idempotency patterns for the write path this archiver depends on.