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

主な考慮事項：

• 少なくとも1回の配信と厳密に1回の配信のセマンティクス
• バックプレッシャーへの対処
• 失敗時に備えたデッドレターキュー
• チェックポイントの管理

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 ポーリング、ウェブフック、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