数据清洗方案

{ "context": { "dataset": { "rows": 1200000, "columns": 26, "core_fields": [ "lead_id", "name", "email", "phone", "country_code", "consent_flag", "consent_ts", "source_channel", "utm_campaign", "utm_source", "page_url", "event_ts", "session_id", "agent_notes" ], "sources": ["website_form", "ads_export", "call_center_api"], "formats": ["CSV", "XLSX", "API_JSON"], "languages": ["zh", "en", "es"], "update_mode": ["daily_batch", "hourly_increment"], "pii_requirement": ["normalize", "mask"] }, "assumptions": [ "Time zone canonicalization target is UTC.", "Lead status may exist in source data even if not listed as a core field; resolve if present.", "Fuzzy matching will be conservative to avoid false merges.", "Consent must not be inferred; missing consent is treated as not granted." ] }, "controlled_vocabularies": { "source_channel": { "allowed": ["ads", "web_form", "call_center", "email", "social", "affiliate", "unknown"], "mapping_examples": { "ad": "ads", "Ads": "ads", "ADS": "ads", "web": "web_form", "call": "call_center" } }, "country_code": { "standard": "ISO_3166_1_alpha2", "case": "upper" }, "lead_status": { "allowed": ["new", "qualified", "won", "lost", "disqualified"], "priority": ["won", "qualified", "new"] } }, "standardized_output_schema": { "lead_id", "name_raw", "name_norm", "name_ascii", "email_raw", "email_norm", "email_domain", "email_domain_valid_flag", "email_canonical_key", "phone_raw", "phone_e164", "phone_valid_flag", "phone_country_inferred", "country_code_iso2", "consent_flag_std", "consent_ts_utc", "consent_valid_flag", "source_channel_raw", "source_channel_std", "utm_source_raw", "utm_source_std", "utm_campaign_raw", "utm_campaign_std", "utm_valid_flag", "page_url_norm", "event_ts_raw", "event_ts_utc", "timezone_source", "session_id", "agent_notes_sanitized", "lead_status_std", "dedupe_cluster_id", "master_lead_id", "pii_mask_email", "pii_mask_phone", "quality_issue_flags" }, "pipeline": [ { "id": "ingest_01", "name": "Unified Ingestion & Encoding Normalization", "objective": "Load all sources into a common staging area with consistent encoding.", "operations": [ "Convert CSV/XLSX/JSON to a unified table format.", "Detect and enforce UTF-8 (normalize BOM, reject invalid sequences).", "Persist raw columns with a raw suffix." ], "outputs": ["stg_leads_raw"] }, { "id": "schema_02", "name": "Schema Alignment & Typing", "objective": "Map disparate schemas to canonical fields and apply types.", "operations": [ "Field mapping from source-specific names to core fields.", "Type casting: email/name/page_url to string; event_ts/consent_ts to timestamp; consent_flag to boolean; country_code/source_channel to string.", "Create missing fields with nulls where absent." ], "outputs": ["stg_leads_aligned"] }, { "id": "unicode_03", "name": "Unicode Normalization", "objective": "Normalize multi-language text to consistent form.", "operations": [ "Apply Unicode NFKC to name, email, page_url, agent_notes.", "Trim leading/trailing whitespace; collapse internal multiple spaces." ], "outputs": ["stg_leads_unicode"] }, { "id": "email_04", "name": "Email Standardization & Validation", "objective": "Resolve case, aliases, illegal domains, and produce a canonical key.", "rules": { "normalization": [ "Lowercase entire email.", "Split local-part and domain; trim spaces.", "Remove plus-tags in local-part (e.g., user+tag@example.com → user@example.com).", "For gmail/googlemail: remove dots in local-part; map domain googlemail.com → gmail.com.", "Convert internationalized domains to punycode before validation." ], "validation": [ "RFC 5322 pattern check (conservative).", "Domain validation via public suffix list; flag disposable/invalid TLDs.", "Optional: MX lookup (record flag only, do not block pipeline)." ], "canonical_key": "email_canonical_key = domain + '|' + normalized local-part" }, "outputs": ["email_norm", "email_domain", "email_domain_valid_flag", "email_canonical_key"] }, { "id": "phone_05", "name": "Phone Normalization to E.164", "objective": "Normalize diverse formats, apply country, and validate.", "rules": { "cleaning": [ "Remove spaces, hyphens, parentheses, dots, and non-digits except leading '+'.", "If country_code present and phone lacks '+', prepend corresponding country calling code." ], "validation": [ "Parse using E.164 rules (e.g., libphonenumber class of validators).", "Flag invalid lengths or impossible numbers." ], "inference": [ "If country_code missing, infer from phone country or page_url TLD when confidence is high.", "Record inference source; do not overwrite explicit country_code." ] }, "outputs": ["phone_e164", "phone_valid_flag", "phone_country_inferred"] }, { "id": "country_06", "name": "Country Code Standardization", "objective": "Ensure ISO alpha-2 uppercase and reconcile with phone inference.", "rules": [ "Uppercase and validate against ISO 3166-1 alpha-2.", "If missing, use phone_country_inferred; else leave null.", "Flag mismatch between explicit country_code and phone country for review." ], "outputs": ["country_code_iso2"] }, { "id": "consent_07", "name": "Consent Normalization", "objective": "Standardize consent flag and timestamp safely.", "rules": [ "consent_flag_std = true only if explicit true from source; otherwise false.", "Parse consent_ts with source timezone; convert to UTC.", "consent_valid_flag = true if consent_flag_std=true and consent_ts_utc not null; else false." ], "outputs": ["consent_flag_std", "consent_ts_utc", "consent_valid_flag"] }, { "id": "utm_08", "name": "UTM Recovery & Canonicalization", "objective": "Fix missing/spelled UTM parameters and standardize.", "operations": [ "Parse page_url query for utm_source and utm_campaign if fields are null.", "Lowercase, trim, and percent-decode UTM values.", "Correct common misspellings (e.g., utm_srouce→utm_source; cmpaign→campaign).", "Restrict to alphanumerics, '-', '', and '.'; replace other chars with '_'.", "Set utm_valid_flag=false if both utm_source_std and utm_campaign_std are null." ], "outputs": ["utm_source_std", "utm_campaign_std", "utm_valid_flag"] }, { "id": "channel_09", "name": "Source Channel Standardization", "objective": "Map scattered enumerations to controlled vocabulary.", "rules": [ "Lowercase, trim.", "Apply mapping table; default to 'unknown' if unmapped." ], "outputs": ["source_channel_std"] }, { "id": "time_10", "name": "Event Timestamp & Timezone Reconciliation", "objective": "Resolve inconsistencies and normalize to UTC.", "rules": [ "Detect timezone from explicit offset in event_ts if present.", "If absent, derive timezone_source by precedence: session metadata > page_url (tz or locale) > country_code.", "Convert event_ts to event_ts_utc using derived timezone.", "Flag records where derivation is low confidence or contradicts source." ], "outputs": ["event_ts_utc", "timezone_source"] }, { "id": "name_11", "name": "Name Normalization & Encoding Unification", "objective": "Normalize multi-language names without losing original form.", "operations": [ "name_norm: NFKC + trim + collapse spaces.", "name_ascii: transliterate diacritics for Latin scripts; preserve CJK as-is.", "Optionally split into heuristics (given/family) only for downstream modeling; keep single name field canonical." ], "outputs": ["name_norm", "name_ascii"] }, { "id": "status_12", "name": "Lead Status Resolution", "objective": "Resolve overlapping statuses (New, Qualified, Won).", "rules": [ "Normalize to lowercase and map to allowed statuses.", "For multiple statuses across duplicates, pick highest priority: won > qualified > new.", "Record source of status and resolution action." ], "outputs": ["lead_status_std"] }, { "id": "dedupe_13", "name": "Cross-Source Deduplication", "objective": "Cluster duplicate leads across CRM, ads, and call center.", "method": { "blocking": [ "Block A: email_canonical_key exact match.", "Block B: phone_e164 exact match.", "Block C: name_norm + country_code_iso2 with phonetic key (for records lacking email/phone)." ], "similarity": [ "Email near-duplicate: same domain AND Levenshtein distance ≤1 on local-part after alias normalization.", "Name similarity: Jaro-Winkler ≥0.92 on name_norm.", "Temporal proximity: event_ts_utc within 30 days increases match confidence." ], "merge_thresholds": [ "Auto-merge if (email exact OR phone exact) OR (email near-duplicate AND name similarity AND same country).", "Manual review queue if similarity scores are borderline." ], "survivorship": [ "Primary key: master_lead_id = earliest valid lead_id or generated UUID.", "Prefer record with consent_valid_flag=true.", "Prefer Won > Qualified > New for lead_status_std.", "For each field, choose the most complete and most recent non-null value; keep provenance." ] }, "outputs": ["dedupe_cluster_id", "master_lead_id"] }, { "id": "notes_14", "name": "Agent Notes Sanitization", "objective": "Remove PII and normalize text.", "operations": [ "Redact patterns resembling emails and phone numbers.", "Normalize whitespace and control characters." ], "outputs": ["agent_notes_sanitized"] }, { "id": "pii_15", "name": "PII Masking", "objective": "Protect sensitive data while preserving utility.", "rules": [ "pii_mask_email: first char + '@' + domain (e.g., j@example.com).", "pii_mask_phone: show last 4 digits; mask others (e.g., ******1234).", "Retain full normalized values in restricted tables; expose masked values in analytics outputs." ], "outputs": ["pii_mask_email", "pii_mask_phone"] }, { "id": "quality_16", "name": "Quality Flags & Issue Cataloging", "objective": "Annotate records with detected issues for monitoring and remediation.", "operations": [ "quality_issue_flags: JSON array of codes (e.g., EMAIL_INVALID_DOMAIN, PHONE_INVALID, CONSENT_MISSING_TS, UTM_MISSING, TZ_DERIVED_LOW_CONF, STATUS_CONFLICT)." ], "outputs": ["quality_issue_flags"] }, { "id": "publish_17", "name": "Publish Standardized Outputs", "objective": "Produce curated tables for downstream use.", "outputs": [ "cur_leads_master (one row per master_lead_id)", "cur_leads_touchpoints (normalized events with event_ts_utc)", "dq_issues_log (record-level issues)" ] } ], "validation_rules": { "completeness": [ "lead_id not null in staging; master_lead_id not null in curated.", "event_ts_utc not null for publishable touchpoints.", "If consent_flag_std=true then consent_ts_utc must not be null." ], "validity": [ "email_domain_valid_flag must be true for email_norm to be considered valid.", "phone_valid_flag must be true for phone_e164 to be considered valid.", "country_code_iso2 must be in ISO list when present.", "source_channel_std must be in controlled vocabulary." ], "consistency": [ "Timezone_source derivation should be consistent with country_code when confidence is high; otherwise flag.", "lead_status_std must reflect survivorship priority after dedupe." ], "uniqueness": [ "master_lead_id unique in cur_leads_master.", "session_id uniqueness scoped per master_lead_id; duplicates indicate repeated events." ] }, "monitoring": { "cadence": ["hourly_increment", "daily_batch"], "metrics": [ "Email invalid rate (%)", "Phone invalid rate (%)", "Consent missing timestamp count", "UTM missing/invalid rate (%)", "Timezone derivation low-confidence rate (%)", "Dedup merge rate (%) and manual review queue size", "Source_channel unmapped rate (%)", "PII masking coverage (%)" ], "thresholds_examples": { "email_invalid_rate_pct": "<= 5%", "phone_invalid_rate_pct": "<= 8%", "utm_invalid_rate_pct": "<= 10%", "timezone_low_conf_pct": "<= 3%" }, "alerting": [ "Trigger alerts when thresholds are exceeded.", "Log top offending sources (by source_channel_raw and source system)." ] }, "error_handling": { "quarantine_buckets": [ "EMAIL_ILLEGAL_DOMAIN", "PHONE_UNPARSABLE", "CONSENT_FLAG_TRUE_TS_NULL", "UTM_IRRECOVERABLE", "TZ_CONFLICT", "STATUS_AMBIGUOUS" ], "actions": [ "Exclude quarantined records from master publish until resolved.", "Provide remediation reports to source owners." ] }, "implementation_notes": { "scalability": [ "Use distributed processing for 1.2M rows (e.g., Spark or equivalent) for dedupe clustering.", "Index on email_canonical_key and phone_e164 for blocking joins." ], "provenance": [ "Track source_system, ingestion_ts, and field-level lineage.", "Maintain pre-merge snapshots for auditability." ], "security": [ "Restrict access to unmasked PII tables.", "Encrypt at rest and in transit." ] } }

实施提示（与规模相关）：

• 分布式处理：使用分区窗口（按canonical_device_id）进行排序、去重与滚动统计，避免跨设备混算。12百万行规模建议批量清洗在对象存储侧（如Spark/SQL）并将规范化结果写入时序库。
• 规则管理：建立可配置的规则字典（单位识别、范围阈值、固件字段映射、时区推断），版本化管理，保证可回溯。
• 字段建议（新增列）：canonical_device_id、ts_utc、tz_offset_minutes、ts_source、ts_utc_imputed、temp_c、temp_unit_source、humidity_out_of_range、battery_jump_flag、lat_imputed/lng_imputed、quality_flags、payload_json_extra、source_schema_version、partition_key、processing_version。
• 验证与抽样：清洗后对每设备抽样检查时间序列单调性（排序）、重复事件剩余率、单位统一正确率、越界字段残留率等，确保清洗生效。