pipeline agent parsing timestamps in mixed formats: canonicalize first

TL;DR

The agent parses timestamps from multiple sources with mixed formats, so some parse wrong and ordering breaks. Canonicalize first: convert every timestamp to UTC ISO 8601 at ingestion, in one place, before any downstream logic ever sees the raw string.

Symptom

A pipeline agent ingests timestamps in mixed formats: epoch seconds from one source, MM/DD/YYYY from another, ISO strings from a third. Some parse into the wrong date, some fail outright, and event ordering across sources is silently wrong because the formats were never unified.

When to use

• Sources mix epoch, US-style dates, and ISO strings
• Event ordering across sources looks wrong
• Parses fail on some sources but not others

When not to use

• Timestamps are already uniform and the bug is display-side (thats formatting, not parsing)
• The data has no timezone info and none can be assumed (flag it, do not guess silently)
• You need sub-second precision the parser drops (fix the parser, not the pipeline shape)

Steps

1. Survey the actual formats in the data before writing any parser:

from collections import Counter
print(Counter(sample_formats))  # e.g. epoch, MM/DD/YYYY, ISO

Expected output: the real format inventory. Agents usually discover a third format only after it breaks something; find them all up front.

2. Write one canonicalize function that handles every known format and always emits UTC ISO 8601:

from datetime import timezone
from dateutil import parser
def canonicalize(raw, assume_tz=timezone.utc):
    dt = parser.parse(str(raw))
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=assume_tz)
    return dt.astimezone(timezone.utc).isoformat()

Expected output: every input format maps to one output shape. Naive datetimes get the documented assumed timezone instead of the machine's local one.

3. Call it at the ingestion boundary, before branching logic:

record["event_time"] = canonicalize(record["event_time"])

Expected output: everything downstream (sorting, windowing, dedupe) works on uniform UTC strings. Parsing never happens twice in two different ways.

4. Reject or quarantine values that do not parse instead of coercing them silently:

try:
    record["event_time"] = canonicalize(raw)
except Exception:
    quarantine.append(record)  # never silently default to now()

Expected output: bad timestamps land in a quarantine set for review. Defaulting unparseable values to the current time corrupts ordering invisibly.

Variant phrasings

mixed date formats breaking sort order

String-sorting mixed formats orders them lexicographically, not chronologically. Canonicalize to ISO 8601 and the string sort becomes a time sort.

epoch vs ISO timestamps in the same column

Detect numerics first and convert from epoch before the string parser ever sees them.

Why it happens

Each source emits its own convention, and agents tend to parse each source with whatever worked for the first one. The result is N parsers with N edge cases instead of one normalization point. Ordering and windowing then compare values that were never in the same frame of reference.

Edge cases

• Daylight-saving transitions make some local times ambiguous. Record the source timezone explicitly rather than letting the parser guess.
• Epoch values in milliseconds vs seconds differ by a factor of 1000. Sanity-check magnitudes: a 13-digit epoch is milliseconds.
• MM/DD/YYYY vs DD/MM/YYYY is unresolvable from the string alone for days 1-12. Get the source's locale from documentation, not from the data.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_ntLpsxYauyRD1N8STfC7Kg