45-EFT.WP.Data.Pipeline v1.0 | 第4章 数据源与摄取

目录 ／ 文档-技术白皮书（V5.05） ／ 45-EFT.WP.Data.Pipeline v1.0

第4章 数据源与摄取

I. 章节目的与范围

II. 术语与依赖

• 术语：source/connector、ingest、checkpoint、cursor/watermark、idempotency、dedupe_key、SLA/SLO、orchestrator。
• 依赖：数据契约与导出见《Core.DataSpec v1.0》；单位与量纲校核见《Core.Metrology v1.0》；与切分/质量的一致性回指《DatasetCards v1.0》；与特征/I-O 假设一致性回指《ModelCards v1.0》。
• 数学与符号：QPS、T_inf、ρ 等内联符号一律用反引号；含除号/积分/复合算符的表达必须加括号；如涉路径量 T_arr，登记 gamma(ell) 与 d ell；公式/符号/定义禁用中文。

III. 字段与结构（规范性）

stage:

name: "<src.kind.name>"

type: "source.<s3|gcs|fs|db|kafka|http|custom>"

impl: "I16-1.<impl_id>"

params:

endpoint: "<url-or-bootstrap>"

bucket_or_db: "<bucket|db>"

prefix_or_table: "<prefix|schema.table>"

query_or_pattern: "<sql|glob>"

credentials_ref: "secrets://path/to/credential"

format: "<json|parquet|csv|avro|binary>"

watermark:

field: "<updated_at|offset|lsn>"

start: "<ISO8601|offset>"

step: "<PT5M|1000>"

checkpoint:

path: "s3://.../chk/<stage>"

mode: "exactly-once|at-least-once"

dedupe_key: ["<pk>", "<ts>"]

outputs: ["raw_blob|raw_rows|events"]

idempotent: true

retries: {max: 3, backoff: "expo", jitter_ms: 200}

timeout_s: 1800

on_fail: "quarantine|skip|block"

schema_ref: "<contracts/raw@vX.Y>"

IV. 连接器类型与规范

• 对象存储/文件系统（source.s3|gcs|fs）：支持分片与断点续传；prefix + pattern 明确扫描域；记录 etag/mtime/size 以便去重。
• 数据库（source.db）：强制变更水位（watermark.field）或增量日志（lsn/scn）；SQL 带参数化与上限；导出表结构快照。
• 消息/流（source.kafka|pulsar）：消费组与偏移管理；按分区并行；提供重放策略与最大滞后阈值。
• HTTP/自定义（source.http|custom）：节流与速率限制；幂等签名或去重键；错误分类与回退。
• 格式与契约：format 与 schema_ref 强绑定；若为半结构/二进制，需在下游 validate 层完成解码与校验。

V. 幂等、重试与断点续传

• 幂等键：优先使用来源端主键/偏移/水位（dedupe_key）；写入端以幂等插入或UPSERT保障无重复。
• 重试：指数退避（backoff:"expo"）+ 抖动；错误分类（可重试/不可重试/告警升级）；单次尝试与总时限（timeout_s）必须显式。
• 断点：checkpoint.mode:"exactly-once|at-least-once" 与提交原子性说明（提交点包含：偏移、游标、目标写入完成标志）。

VI. 去重与顺序保证

• 去重：对批式，按 dedupe_key + watermark 去重；对流式，分区内保持单调偏移与幂等汇聚。
• 顺序：如业务要求强序，需声明分区键、排序字段与迟到窗口；否则标注“分区内有序，跨分区不保证顺序”。

VII. 计量与单位（SI）

• 吞吐：QPS = ( count / Δt )（1/s）；
• 时延：T_inf 以 ms，报告 {p50,p95,p99}；
• 利用率：ρ = ( λ / μ )；
• 带宽/存储：net_mbps、size_bytes；
• metrology:{units:"SI", check_dim:true} 为强制；所有复合度量合成前先做单位归一。

VIII. 安全、凭据与合规

• 凭据：仅引用 credentials_ref；禁止明文密钥；支持轮换与最小权限（只读原则）。
• 隐私：如含 PII/敏感字段，需执行最小化与脱敏；对外区域限制（如 EU-GDPR）在 export_manifest.references[] 中体现。
• 审计：所有连接/拉取/失败事件写入安全审计流；重大失败进入 on_fail:"quarantine" 并触发告警。

IX. 机器可读片段（可直接嵌入）

layers:

- name: "ingest"

stages:

- name: "src.s3.pull"

type: "source.s3"

impl: "I16-1.s3_pull"

params:

endpoint: "https://s3.amazonaws.com"

bucket_or_db: "eift-data"

prefix_or_table: "raw/2025/09/"

query_or_pattern: "*.jsonl"

credentials_ref: "secrets://aws/ingest_ro"

format: "json"

watermark: {field:"updated_at", start:"2025-09-01T00:00:00Z", step:"PT5M"}

checkpoint: {path:"s3://eift-meta/chk/src.s3.pull", mode:"at-least-once"}

dedupe_key: ["id","updated_at"]

outputs: ["raw_blob"]

idempotent: true

retries: {max:3, backoff:"expo", jitter_ms:200}

timeout_s: 1800

on_fail: "quarantine"

schema_ref: "contracts/raw_json@v1.2"

X. Lint 规则（节选，规范性）

lint_rules:

- id: SRC.TYPE_ALLOWED

when: "$.layers[*].stages[*].type"

assert: "value in ['source.s3','source.gcs','source.fs','source.db','source.kafka','source.http','source.custom']"

level: error

- id: SRC.CREDENTIALS_REF

when: "$.layers[*].stages[?(@.type^='source.')].params"

assert: "has_key('credentials_ref') and not has_key('plain_secret')"

level: error

- id: SRC.CHECKPOINT_DEFINED

when: "$.layers[*].stages[?(@.type^='source.')].params"

assert: "has_key('checkpoint') and has_key('watermark')"

level: error

- id: SRC.DEDUPE_OR_EXACTLY_ONCE

when: "$.layers[*].stages[?(@.type^='source.')]"

assert: "has_key('params.dedupe_key') or $.params.checkpoint.mode == 'exactly-once'"

level: error

- id: METROLOGY.SI_AND_CHECKDIM

when: "$.metrology"

assert: "units=='SI' and check_dim==true"

level: error

XI. 导出清单与审计轨

export_manifest:

version: "v1.0"

artifacts:

- {path:"ingest/pulled.manifest.json", sha256:"..."}

- {path:"ingest/checkpoint.meta.json", sha256:"..."}

- {path:"security/audit.log", sha256:"..."}

references:

- "EFT.WP.Core.DataSpec v1.0:EXPORT"

- "EFT.WP.Core.Metrology v1.0:check_dim"

- "EFT.WP.Data.DatasetCards v1.0:Ch.6"

XII. 本章合规自检

• 连接器 type/impl/params 完整，credentials_ref 引用有效，未出现明文密钥。
• watermark 与 checkpoint 已配置；dedupe_key 或 exactly-once 保障不重复。
• 重试/超时/幂等策略明确；失败分级与 on_fail 行为可追溯。
• metrology.units="SI" 且 check_dim=true；QPS/T_inf/ρ/net_mbps 等指标单位一致。
• schema_ref 与上游契约对表；审计工件与引用锚点已在 export_manifest 登记并具 sha256。