数据转化逻辑指导

Below is a concise, implementation-ready description of the data transformations required to make your event log usable for A/B experiments, funnels, and retention analyses. It focuses on correctness under late-arriving and out-of-order events, UTC+8 time boundaries, and identity stitching between anonymous_id and uid.

1. Target outputs (curated data model)

• canonical_events: One row per cleaned event, ordered by event time, with standardized fields.
• user_identity_map: Mapping of anonymous_id → uid (many-to-one) with first_seen_at and last_seen_at.
• ab_assignments: First exposure row per user_key and experiment, with frozen variant.
• sessions: Sessionized user activity (optional but recommended for funnel and retention diagnostics).
• fact_orders and fact_order_items: Order-level and line-item facts derived from purchase events.
• daily_user_activity: One row per user per local day for retention and DAU/WAU/MAU (based on a defined activity set).

2. Core challenges to address

• Late and out-of-order events: Use event time semantics with watermarks; reprocess recent partitions.
• Time zone: Business day is UTC+8; conversion must be systematic and consistent.
• Identity stitching: Merge anonymous_id and uid into a durable user_key without contaminating A/B assignment.
• Schema normalization: Parse props(json), standardize field names/types, and deduplicate.

3. Transformations pipeline A. Parse and normalize the raw schema

• Input columns: event_name, user_id, ts, ab_group, device, props (JSON).
• Parse props(JSON) and extract canonical fields (e.g., product_id, cart_id, order_id, currency, price, quantity, anonymous_id, uid, experiment_id if present).
• Type cast all extracted fields (strings, numerics, ISO currency, etc.).
• Standardize device (e.g., device_type, os, app/web).

B. Timestamp handling and local day boundaries (UTC+8)

• Assume ts is recorded in UTC+8. Create:
  • event_time_local = parse(ts) in UTC+8.
  • event_time_utc = convert(event_time_local to UTC).
  • event_date_local = date(event_time_local).
• Add ingestion_time_utc (load time) if available.
• Use event_time_local for funnels, retention, and A/B day boundaries; use event_time_utc for cross-system joins.

C. Identity stitching (anonymous_id + uid)

• Extraction:
  • From props, extract anonymous_id and uid (do not rely solely on user_id).
• Mapping table (user_identity_map):
  • When an event carries both anonymous_id and uid, create or update a link anonymous_id → uid with first_seen_at = min(link times), last_seen_at = max.
  • Enforce many-to-one: if an anonymous_id later links to a different uid, keep the earliest uid and flag conflicts for QA.
• Durable user_key:
  • user_key = coalesce(uid, map(anonymous_id), anonymous_id).
  • Backfill: Once a link is established, backfill user_key for past events of that anonymous_id within a configurable lookback window (e.g., 30 days) to unify history.
• A/B caveat: For A/B analysis, freeze assignment at first exposure (see E). Do not let post-exposure identity merges change the assigned variant; handle collisions explicitly.

D. Event standardization and deduplication

• Event ID:
  • Use provided event_id if present in props; else create a deterministic id_hash = hash(source_id, user identifiers, event_name, critical keys like order_id/product_id, event_time_utc).
• Dedup rule:
  • Keep the latest record by ingestion_time_utc for identical event_id.
  • If no event_id, dedup within a time window (e.g., 24h) by (user_key, event_name, product_id/order_id, event_time_utc second-level) with last-write-wins.
• Ordering and late data:
  • Use event_time_local as the primary sort key; ingestion_time_utc as a tie-breaker.
  • Apply an event-time watermark (e.g., 72 hours) for streaming; in batch, reprocess the last N local days (e.g., 7 days) to capture late arrivals.

E. A/B experiment readiness

• Exposure definition:
  • Exposure occurs on the first event where ab_group (and optionally experiment_id) is observed or where props indicates assignment/exposure.
• ab_assignments table:
  • Keys: user_key, experiment_id (if multiple experiments), variant (ab_group normalized), exposure_time_local, exposure_date_local.
  • Freeze variant at first exposure_time_local. If a later event shows a different variant for the same user_key/experiment, flag as contamination and exclude or handle per analysis policy.
  • If identity merge links two identifiers with different variants, choose the earliest exposure and flag as conflict; define a rule (exclude from primary, include in sensitivity).
• Attach variant to events:
  • Left join canonical_events to ab_assignments on user_key and event_time_local >= exposure_time_local to tag post-exposure events for effect measurement.
  • For pre-exposure covariates, use events strictly before exposure_time_local.

F. Funnel readiness

• Flatten and standardize core events:
  • view_product: product_id, category, price, currency, referrer.
  • add_to_cart: product_id, quantity, price, currency, cart_id.
  • checkout: cart_id, step (if available), order_id (if available).
  • purchase: order_id (required), revenue_gross, discount, tax, shipping, currency; items[] if available.
• Derivations/joins:
  • If purchase is order-level with items[], explode to fact_order_items for product-level funnels.
  • Link checkout and add_to_cart via cart_id; link purchase via order_id. If missing IDs, use session- and time-based heuristics (nearest prior cart within same session).
• Sequencing:
  • Order events per user_key by event_time_local to build funnel steps.
  • Optional product-consistent funnels: require the same product_id across steps; otherwise do cart/order-consistent funnels.

G. Retention readiness

• Activity set:
  • Define active_event_flag = 1 for events considered “active” (e.g., any of view_product, add_to_cart, checkout, purchase; adjust as needed).
• Cohort logic:
  • cohort_event default: first observed active event for the user_key (configurable).
  • cohort_date_local = date of cohort_event.event_time_local.
• Daily activity:
  • daily_user_activity aggregates by user_key, event_date_local; include device, ab variant, and activity flags.
  • Retention computed as returning activity on day N after cohort_date_local (D1, D7, etc.) using event_date_local.

4. Canonical event schema (recommended)

• Core fields:
  • event_id, event_name
  • event_time_local, event_time_utc, event_date_local
  • user_key, uid, anonymous_id
  • device_type, os, platform (from device and props)
  • ab_group_raw, experiment_id (if present), ab_variant (normalized)
  • product_id, category, price, currency, quantity
  • cart_id, order_id, revenue_gross, discount, tax, shipping
  • ingestion_time_utc, source (pipeline source)
• Notes:
  • Keep raw props as a JSON blob for auditability in addition to flattened fields.

5. Event definitions and conversion rules (implementation-ready)

• view_product
  • Required: product_id
  • Optional: price, currency, category, referrer
  • Dedup recommended: one view per user_key-product_id per session to avoid overcount inflation (configurable).
• add_to_cart
  • Required: product_id, cart_id (if available), quantity >= 1
  • Revenue fields optional; if missing, inherit latest known price for that product in session.
• checkout
  • Required: cart_id; if order_id already known, capture it.
  • Treat first checkout per cart_id as “checkout_start.”
• purchase
  • Required: order_id, currency; revenue fields (revenue_gross, discount, tax, shipping) strongly recommended.
  • If items[] present, explode into fact_order_items with product_id, quantity, price.
• Conversion rules (parameterized; defaults are placeholders):
  • Funnel conversion window: Analyze with event_time_local ordering; default window e.g., 7 days from first step (configurable).
  • Uniqueness: One conversion per user_key per funnel instance; define per cart_id/order_id for cart/order funnels; per product_id for product funnels.
  • Attribution: Attribute purchase to the most recent eligible funnel path for the same cart_id/order_id; if absent, use latest session path.

6. Handling late and out-of-order data

• Watermarks and reprocessing:
  • Streaming: allow 72h allowed lateness (tune to actual delay distribution).
  • Batch: daily job reprocesses last 7 local days; weekly job reprocesses last 30 days for identity backfills and order corrections.
• Idempotency:
  • Dedup by event_id; backfill identity safely (do not change frozen A/B variants).
• Monitoring:
  • Track late ratio by delay buckets (0–1h, 1–24h, 1–7d), out-of-order rate, and dedup rate.

7. A/B analysis safeguards

• Freeze variant at first exposure_time_local in ab_assignments.
• Exclude users with variant collisions by default from primary reads (publish a collision rate metric).
• For cross-device merges, apply earliest exposure after identity stitch; do not reassign retroactively.
• Anchor all experiment day boundaries to event_date_local (UTC+8).

8. Sessions (recommended)

• Build sessions per user_key using 30-minute inactivity gap on event_time_local.
• Store session_id, session_start_local, session_end_local, device, ab_variant at session start.
• Use session_id to constrain funnels and to impute missing cart_id/order_id when needed.

9. Data quality checks

• Required fields non-null rates per event_name (e.g., purchase must have order_id).
• Currency and price sanity (positive, valid ISO code).
• Identity consistency: anonymous_id mapping cardinality; collision rate.
• Time sanity: no future timestamps; distribution of negative intervals (purchase before view) after ordering by event_time_local.
• A/B coverage: share of events/users with ab assignment; variant balance.

10. Example implementation notes (SQL-level guidance)

• Flatten props: SELECT json_extract_scalar(props, '$.product_id') AS product_id, ...
• Identity map: INSERT INTO user_identity_map when anonymous_id and uid coexist in the same event; coalesce in canonical_events using left join to map table.
• Canonical events: Build via staging → dedup (by event_id/hash) → enrichment (device/identity/AB) → publish to canonical_events partitioned by event_date_local.
• AB assignments: SELECT first_value(ab_group) over (partition by user_key, experiment_id order by event_time_local) as frozen variant at first exposure.
• Watermark: filter or tag events with event_time_local older than watermark; reprocess recent partitions to integrate late data.

Key conclusions

• Use event-time processing with UTC+8 business-day boundaries to ensure correct funnels, retention, and A/B windows.
• Create a durable user_key via anonymous_id-to-uid stitching with backfill, but freeze A/B assignment at first exposure to avoid contamination.
• Standardize and deduplicate core events, enrich with order/cart links, and expose both order-level and item-level facts.
• Implement watermarks and scheduled reprocessing to capture late and out-of-order arrivals reliably.
• Maintain explicit ab_assignments and daily_user_activity tables to streamline experiment reads, funnel building, and retention views.