使用说明
🏦 AI统一数据平台
实时和批量交易智能
一个完全可操作的数据工程管道，通过双摄取路径（Apache Spark批处理+Apache Kafka流）处理合成银行交易，将干净的数据存储在PostgreSQL中，并通过GPT-4o生成人工智能驱动的财务异常摘要。在Apache Airflow上进行端到端的编排。
🏗️ 建筑
╔══════════════════════════════════════════════════════════════╗
║ DATA SOURCES ║
║ CSV / Parquet Files Kafka Topic ║
║ (historical) (real-time) ║
╚══════════╦═══════════════════════════════════╦═══════════════╝
║ ║
┌──────────▼──────────┐ ┌────────────▼────────────┐
│ Spark Batch ETL │ │ Spark Structured │
│ │ │ Streaming │
│ • Multi-format read │ │ • foreachBatch() │
│ • Column normalize │ │ • 10s micro-batches │
│ • Type casting │ │ • Checkpoint recovery │
│ • Bad record split │ │ • Idle auto-stop │
└──────────┬──────────┘ └────────────┬────────────┘
║ ║
╔══════════▼═══════════════════════════════════▼═════════════╗
║ Shared Transformation Layer ║
║ clean_and_cast → add_error_column → finalize_good_data ║
╚══════════╦═══════════════════════════════════╦═════════════╝
║ ║
┌──────────▼──────────┐ ┌────────────▼────────────┐
│ PostgreSQL │ │ Bad Records Store │
│ • Schema evolution │ │ timestamped CSV │
│ • JDBC append │ │ error_reason column │
└──────────┬──────────┘ └─────────────────────────┘
║
╔══════════▼═════════════════════════════════════════════════╗
║ Insight Generation Layer ║
║ generate_insights_from_df() ║
║ → insight_spark_batch_{date}.json ║
║ → insight_kafka_stream_{date}.json ║
╚══════════╦═════════════════════════════════════════════════╝
║
╔══════════▼═════════════════════════════════════════════════╗
║ AI Insight Engine ║
║ ║
║ rule_engine.py → deterministic ground truth ║
║ ↓ ║
║ insight_engine.py → aggregated prompt (~800 tokens) ║
║ ↓ ║
║ github_client.py → GPT-4o via GitHub Models ║
║ ↓ ║
║ cache_store.py → MD5-keyed, no duplicate LLM calls ║
║ ↓ ║
║ ai_insights_combined_{date}.json ║
╚══════════╦═════════════════════════════════════════════════╝
║
╔══════════▼═════════════════════════════════════════════════╗
║ Apache Airflow Orchestration ║
║ ║
║ spark_batch_processing ║
║ └──→ ai_combined_insights ║
║ └──→ archive_insight_files ║
║ ║
║ Schedule: Daily 06:00 UTC ║
║ Retries: 2 × 5 min delay | SLA: 1 hr ║
╚════════════════════════════════════════════════════════════╝
🔬 关键工程决策
- 令牌高效LLM提示设计
一个简单的实现将所有行直接序列化为提示字符串。在225个分组行中，这产生了一个34286个字符的提示（约8500个令牌），达到了GPT-4o的GitHub模型限制，并导致管道故障。 修复：只发送聚合信号——规则引擎输出+前5个提款天数+前5次存款天数。无论交易量如何，提示大小现在都是恒定的。
| 数量 | 天真的方法 | 这条管道 |
|---|---|---|
| 728特克斯→ 225 分组行 | ~8500个标记❌ | ~800个代币✅ |
| 50000笔交易 | 约750000个代币❌ | ~800个代币✅ |
| 5000000笔交易 | 不可能❌ | ~800个代币✅ |
- 规则引擎作为不可变的基础真理
rule_engine.py 始终在任何LLM调用之前运行，并生成确定性聚合。法学硕士将这些数字作为固定事实接收，并被指示只叙述，从不重新计算。这可以防止金融产出中出现幻觉数字。
raw_transactions
↓
rule_engine.py → { total_txns, deposits, withdrawals, anomaly, confidence }
↓
LLM receives these as GROUND TRUTH — narrates, never recalculates
- 优雅的后退——永不沉默的失败
如果LLM调用因任何原因失败，管道将退回到完全由规则引擎输出构建的结构化自然语言摘要。具体的错误出现在JSON中。没有崩溃，没有空洞的回应。
{
"source_date": "2026-05-10",
"ai_summary": "LLM unavailable. Using deterministic rule-based insights.",
"rule_summary": {
"total_transactions": 67,
"total_deposit": 7000000.0,
"total_withdrawal": 6445957.0,
"avg_balance": 4188171.79,
"anomaly": "normal",
"confidence": 0.9
},
"mode": "fallback",
"error": "GitHub Model Error: Unauthorized\n",
"note": "Combined insight generated from Spark only. Only Spark pipeline ran today (/opt/project/insights/insight_spark_batch_2026-05-10.json). No Kafka file was available."
}
- 增量Kafka洞察合并
Kafka微批每10秒到达一次。一种简单的方法会覆盖每个批次的每日洞察文件。相反，使用加权平均余额计算合并批次，以便正确累积每日总计。
ex["avg_balance"] = (
(ex["avg_balance"] * n_old + r["avg_balance"] * n_new)
/ (n_old + n_new)
)
- MD5密钥洞察缓存
LLM电话费率有限且费用高昂。每个结果都使用从中导出的MD5密钥进行缓存 source_date +事务指纹。在相同数据上重新运行管道会完全跳过LLM。
📁 项目结构
ai-unified-data-platform/
│
├── spark_dynamic_pipeline.py # Spark batch ETL — ingest, clean, write, insights
├── kafka_streaming_pipeline.py # Kafka streaming — micro-batch processing
├── schema.py # Shared Spark schema definition
├── db_utils.py # JDBC config, schema evolution helpers
├── insight_store.py # DataFrame → aggregated insight JSON
│
├── ai_insight_engine/
│ ├── app.py # Entry point — merges spark + kafka insight files
│ ├── insight_engine.py # Orchestrates rule engine + LLM call + cache
│ ├── rule_engine.py # Deterministic aggregation and anomaly detection
│ ├── github_client.py # GitHub Models API client (GPT-4o)
│ └── cache_store.py # MD5-keyed result cache
│
├── dags/
│ └── batch_ai_pipeline_dag.py # Airflow DAG — scheduling, retries, archival
│
├── data/
│ ├── input_file/ # Drop CSV / Parquet files here
│ └── bad_records/ # Rejected rows — timestamped CSV per run
│
├── insights/
│ ├── insight_spark_batch_{date}.json
│ ├── insight_kafka_stream_{date}.json
│ ├── ai_insights_combined_{date}.json
│ └── archive/ # Previous day's files auto-moved by Airflow
│
├── checkpoint/ # Spark-managed Kafka stream checkpoint
├── .env.example
├── docker-compose.yml
└── requirements.txt
🚨 数据质量——不良记录处理
每一行在到达PostgreSQL之前都要经过四个验证门。不良记录被写入带有时间戳的CSV error_reason 列已填充。两条路都阻挡不了另一条路。
| 验证规则 | 错误标签 |
|---|---|
transaction_date 无法解析为 dd-MMM-yy |
Invalid Date |
account_no 为空白或空白 |
Missing Account |
withdrawal_amt 非空但非数字 |
Invalid Withdrawal Amt |
deposit_amt 非空但非数字 |
Invalid Deposit Amt |
截图
👩💻 作者
Priyusha-数据工程师·信息研究理学硕士生
5年以上跨批处理和流处理系统设计和构建数据管道的经验。该项目反映了全栈数据工程所有权——从原始摄入到人工智能增强分析——而不依赖于托管服务。
所展示的技能：
