ingesting-data

数据摄取模式

此技能提供从外部来源将数据导入系统的模式。

何时使用此技能

• 导入 CSV、JSON、Parquet 或 Excel 文件
• 从 S3、GCS 或 Azure Blob 存储加载数据
• 消费 REST/GraphQL API 数据流
• 构建 ETL/ELT 管道
• 数据库迁移和 CDC（变更数据捕获）
• 从 Kafka/Kinesis 流式摄取数据

摄取模式决策树

What is your data source?
├── Cloud Storage (S3, GCS, Azure) → See cloud-storage.md
├── Files (CSV, JSON, Parquet) → See file-formats.md
├── REST/GraphQL APIs → See api-feeds.md
├── Streaming (Kafka, Kinesis) → See streaming-sources.md
├── Legacy Database → See database-migration.md
└── Need full ETL framework → See etl-tools.md

按语言快速入门

Python（推荐用于 ETL）

dlt（data load tool）——现代 Python ETL：

import dlt

# Define a source
@dlt.source
def github_source(repo: str):
    @dlt.resource(write_disposition="merge", primary_key="id")
    def issues():
        response = requests.get(f"https://api.github.com/repos/{repo}/issues")
        yield response.json()
    return issues

# Load to destination
pipeline = dlt.pipeline(
    pipeline_name="github_issues",
    destination="postgres",  # or duckdb, bigquery, snowflake
    dataset_name="github_data"
)

load_info = pipeline.run(github_source("owner/repo"))
print(load_info)

使用 Polars 处理文件（比 pandas 更快）：

import polars as pl

# Read CSV with schema inference
df = pl.read_csv("data.csv")

# Read Parquet (columnar, efficient)
df = pl.read_parquet("s3://bucket/data.parquet")

# Read JSON lines
df = pl.read_ndjson("events.jsonl")

# Write to database
df.write_database(
    table_name="events",
    connection="postgresql://user:pass@localhost/db",
    if_table_exists="append"
)

TypeScript/Node.js

S3 摄取：

import { S3Client, GetObjectCommand } from "@aws-sdk/client-s3";
import { parse } from "csv-parse/sync";

const s3 = new S3Client({ region: "us-east-1" });

async function ingestFromS3(bucket: string, key: string) {
  const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
  const body = await response.Body?.transformToString();

  // Parse CSV
  const records = parse(body, { columns: true, skip_empty_lines: true });

  // Insert to database
  await db.insert(eventsTable).values(records);
}

API 数据流轮询：

import { Hono } from "hono";

// Webhook receiver for real-time ingestion
const app = new Hono();

app.post("/webhooks/stripe", async (c) => {
  const event = await c.req.json();

  // Validate webhook signature
  const signature = c.req.header("stripe-signature");
  // ... validation logic

  // Ingest event
  await db.insert(stripeEventsTable).values({
    eventId: event.id,
    type: event.type,
    data: event.data,
    receivedAt: new Date()
  });

  return c.json({ received: true });
});

Rust

高性能文件摄取：

use polars::prelude::*;
use aws_sdk_s3::Client;

async fn ingest_parquet(client: &Client, bucket: &str, key: &str) -> Result<DataFrame> {
    // Download from S3
    let resp = client.get_object()
        .bucket(bucket)
        .key(key)
        .send()
        .await?;

    let bytes = resp.body.collect().await?.into_bytes();

    // Parse with Polars
    let df = ParquetReader::new(Cursor::new(bytes))
        .finish()?;

    Ok(df)
}

Go

并发文件处理：

package main

import (
    "context"
    "encoding/csv"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

func ingestCSV(ctx context.Context, client *s3.Client, bucket, key string) error {
    resp, err := client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: &bucket,
        Key:    &key,
    })
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    reader := csv.NewReader(resp.Body)
    records, err := reader.ReadAll()
    if err != nil {
        return err
    }

    // Batch insert to database
    return batchInsert(ctx, records)
}

摄取模式

1. 批量摄取（文件/存储）

用于定期批量加载：

Source → Extract → Transform → Load → Validate
  ↓         ↓          ↓         ↓        ↓
 S3      Download   Clean/Map  Insert   Count check

关键注意事项：

• 对大型文件（>100MB）使用分块读取
• 使用校验和实现幂等性
• 跟踪文件处理状态
• 处理部分失败

2. 流式摄取（实时）

用于持续数据流：

Source → Buffer → Process → Load → Ack
  ↓        ↓         ↓        ↓      ↓
Kafka   In-memory  Transform  DB   Commit offset

关键注意事项：

• 至少一次与恰好一次语义
• 背压处理
• 为失败情况设置死信队列
• 检查点管理

3. API 轮询（数据流）

用于外部 API 数据：

Schedule → Fetch → Dedupe → Load → Update cursor
   ↓         ↓        ↓       ↓         ↓
 Cron     API call  By ID   Insert   Last timestamp

关键注意事项：

• 速率限制和退避
• 增量加载（游标、时间戳）
• 处理 API 分页
• 使用指数退避重试

4. 变更数据捕获（CDC）

用于数据库复制：

Source DB → Capture changes → Transform → Target DB
    ↓             ↓               ↓            ↓
 Postgres    Debezium/WAL      Map schema   Insert/Update

关键注意事项：

• 初始快照 + 流式变更
• 处理架构演进
• 顺序保证
• 冲突解决

库推荐

用例	Python	TypeScript	Rust	Go
ETL 框架	dlt, Meltano, Dagster	-	-	-
云存储	boto3, gcsfs, adlfs	@aws-sdk/, @google-cloud/	aws-sdk-s3, object_store	aws-sdk-go-v2
文件处理	polars, pandas, pyarrow	papaparse, xlsx, parquetjs	polars-rs, arrow-rs	encoding/csv, parquet-go
流式处理	confluent-kafka, aiokafka	kafkajs	rdkafka-rs	franz-go, sarama
CDC	Debezium, pg_logical	-	-	-

参考文档

• references/cloud-storage.md - S3、GCS、Azure Blob 模式
• references/file-formats.md - CSV、JSON、Parquet、Excel 处理
• references/api-feeds.md - REST 轮询、webhook、GraphQL 订阅
• references/streaming-sources.md - Kafka、Kinesis、Pub/Sub
• references/database-migration.md - 架构迁移、CDC 模式
• references/etl-tools.md - dlt、Meltano、Airbyte、Fivetran

脚本

• scripts/validate_csv_schema.py - 根据预期架构验证 CSV
• scripts/test_s3_connection.py - 测试 S3 存储桶连通性
• scripts/generate_dlt_pipeline.py - 生成 dlt 管道脚手架

与数据库技能串联

摄取完成后，串联到相应的数据库技能：

目标	串联到技能
PostgreSQL、MySQL	databases-relational
MongoDB、DynamoDB	databases-document
Qdrant、Pinecone	databases-vector（嵌入后）
ClickHouse、TimescaleDB	databases-timeseries
Neo4j	databases-graph

对于向量数据库，通过 ai-data-engineering 串联以生成嵌入：

ingesting-data → ai-data-engineering → databases-vector