Antigravity AI データパイプライン × ETL自動化 本番構築ガイド

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
// ステップ1: REST APIからデータを取得
async function fetchApiData(endpoint: string, apiKey: string) {
  const response = await fetch(endpoint, {
    headers: { Authorization: `Bearer ${apiKey}` },
  });
  const data = await response.json();
  return data;
}
 
// ステップ2: PostgreSQLから大規模データセットを取得
async function fetchPostgresData(
  connectionString: string,
  query: string
): Promise<Record<string, any>[]> {
  const { Client } = await import("pg");
  const pgClient = new Client({ connectionString });
  await pgClient.connect();
  const result = await pgClient.query(query);
  await pgClient.end();
  return result.rows;
}
 
// ステップ3: S3からJSONLファイルをストリーミング取得
async function fetchS3Data(
  bucket: string,
  key: string,
  awsAccessKey: string,
  awsSecretKey: string
) {
  const { S3Client, GetObjectCommand } = await import("@aws-sdk/client-s3");
  const s3Client = new S3Client({
    region: "us-east-1",
    credentials: {
      accessKeyId: awsAccessKey,
      secretAccessKey: awsSecretKey,
    },
  });
 
  const command = new GetObjectCommand({ Bucket: bucket, Key: key });
  const response = await s3Client.send(command);
  const stream = response.Body;
  const lines: Record<string, any>[] = [];
 
  return new Promise((resolve, reject) => {
    stream.on("data", (chunk: Buffer) => {
      const text = chunk.toString();
      text.split("\n").forEach((line) => {
        if (line.trim()) {
          lines.push(JSON.parse(line));
        }
      });
    });
    stream.on("end", () => resolve(lines));
    stream.on("error", reject);
  });
}

async function inferSchema(
  sampleData: Record<string, any>[],
  dataSourceName: string
) {
  // サンプルデータから統計情報を計算
  const stats = analyzeSampleStatistics(sampleData);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a data engineer analyzing a new data source.
 
Source: ${dataSourceName}
Sample records count: ${sampleData.length}
Data statistics:
${JSON.stringify(stats, null, 2)}
 
First 5 records:
${JSON.stringify(sampleData.slice(0, 5), null, 2)}
 
Your task:
1. Infer the schema (column names, types, constraints)
2. Identify business keys (likely primary keys)
3. Detect any data quality issues in the sample
4. Suggest normalization approach
 
Provide response in JSON format with keys:
- schema (array of {name, type, nullable, constraints})
- businessKeys (array)
- dataQualityIssues (array)
- normalizationStrategy (string)`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 2048,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    throw new Error("No text response from schema inference");
  }
 
  const jsonMatch = textContent.text.match(/\{[\s\S]*\}/);
  if (!jsonMatch) {
    throw new Error("Could not parse schema from response");
  }
 
  return JSON.parse(jsonMatch[0]);
}
 
// 統計情報の計算
function analyzeSampleStatistics(data: Record<string, any>[]) {
  const stats: Record<string, any> = {};
 
  if (data.length === 0) return stats;
 
  const firstRecord = data[0];
  for (const key of Object.keys(firstRecord)) {
    const values = data.map((r) => r[key]);
    const nullCount = values.filter((v) => v === null || v === undefined).length;
    const uniqueCount = new Set(values).size;
 
    stats[key] = {
      nullCount,
      nullPercentage: (nullCount / data.length) * 100,
      uniqueCount,
      uniquePercentage: (uniqueCount / data.length) * 100,
      sampleValues: values.slice(0, 3),
      types: [
        ...new Set(
          values.map((v) =>
            v === null ? "null" : typeof v === "object" ? "object" : typeof v
          )
        ),
      ],
    };
  }
 
  return stats;
}

interface CleaningRule {
  field: string;
  operation: string;
  parameters: Record<string, any>;
  description: string;
}
 
async function generateCleaningRules(
  schema: Record<string, any>,
  sampleData: Record<string, any>[],
  businessContext: string
): Promise<CleaningRule[]> {
  const dataIssues = detectDataQualityIssues(sampleData);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a data quality expert designing cleaning rules.
 
Business Context: ${businessContext}
 
Detected Schema:
${JSON.stringify(schema, null, 2)}
 
Data Quality Issues:
${JSON.stringify(dataIssues, null, 2)}
 
Problem records (showing issues):
${JSON.stringify(sampleData.slice(0, 3), null, 2)}
 
Generate data cleaning rules in JSON format. For each field with issues, create a rule with:
- field: column name
- operation: trim, normalize_email, remove_duplicates, fix_phone, parse_date, remove_outliers, etc.
- parameters: operation-specific config
- description: why this rule is needed
 
Return as JSON array of rules. Be specific and production-ready.`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 2048,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    throw new Error("No text response from rule generation");
  }
 
  const jsonMatch = textContent.text.match(/\[[\s\S]*\]/);
  if (!jsonMatch) {
    throw new Error("Could not parse rules from response");
  }
 
  return JSON.parse(jsonMatch[0]);
}
 
function detectDataQualityIssues(data: Record<string, any>[]) {
  const issues: Record<string, string[]> = {};
 
  data.forEach((record, idx) => {
    Object.entries(record).forEach(([key, value]) => {
      if (!issues[key]) issues[key] = [];
 
      if (value === null || value === undefined) {
        issues[key].push(`Row ${idx}: null/undefined`);
      } else if (typeof value === "string") {
        if (value.trim() === "") issues[key].push(`Row ${idx}: empty string`);
        if (value !== value.toLowerCase() && /\d/.test(value)) {
          issues[key].push(`Row ${idx}: mixed case with numbers`);
        }
      } else if (typeof value === "number") {
        if (value < 0 && key.includes("count")) {
          issues[key].push(`Row ${idx}: negative count value`);
        }
      }
    });
  });
 
  return Object.fromEntries(
    Object.entries(issues).filter(([_, v]) => v.length > 0)
  );
}
 
// クレンジング実装
function applyCleaningRules(
  record: Record<string, any>,
  rules: CleaningRule[]
): Record<string, any> {
  const cleaned = { ...record };
 
  rules.forEach((rule) => {
    const value = cleaned[rule.field];
 
    switch (rule.operation) {
      case "trim":
        cleaned[rule.field] =
          typeof value === "string" ? value.trim() : value;
        break;
 
      case "normalize_email":
        cleaned[rule.field] =
          typeof value === "string" ? value.toLowerCase().trim() : value;
        break;
 
      case "parse_date":
        try {
          cleaned[rule.field] = new Date(value as string).toISOString();
        } catch {
          cleaned[rule.field] = null;
        }
        break;
 
      case "remove_outliers":
        const { threshold } = rule.parameters;
        if (
          typeof value === "number" &&
          Math.abs(value) > threshold
        ) {
          cleaned[rule.field] = null;
        }
        break;
 
      case "fix_phone":
        if (typeof value === "string") {
          cleaned[rule.field] = value.replace(/\D/g, "");
        }
        break;
    }
  });
 
  return cleaned;
}

async function enrichWithExternalData(
  record: Record<string, any>,
  enrichmentConfig: {
    geoLookup?: boolean;
    currencyConversion?: boolean;
    entityMatching?: boolean;
  }
): Promise<Record<string, any>> {
  const enriched = { ...record };
 
  if (enrichmentConfig.geoLookup && enriched.ip_address) {
    // IP アドレスから地理情報を取得（疑似実装）
    enriched.geo_country = "JP";
    enriched.geo_city = "Tokyo";
  }
 
  if (enrichmentConfig.currencyConversion && enriched.amount) {
    // 通貨換算（ExchangeRate API利用）
    enriched.amount_jpy = (enriched.amount as number) * 130;
  }
 
  if (enrichmentConfig.entityMatching && enriched.company_name) {
    // 企業マスタとのマッチング
    const messages: Anthropic.Messages.MessageParam[] = [
      {
        role: "user",
        content: `Find matching company entity for: "${enriched.company_name}"
 
Return JSON with:
- matched_name (exact company name or null)
- confidence (0-1)
- entity_id (internal ID or null)`,
      },
    ];
 
    const response = await client.messages.create({
      model: "claude-3-5-sonnet-20241022",
      max_tokens: 256,
      messages,
    });
 
    const textContent = response.content.find((block) => block.type === "text");
    if (textContent && textContent.type === "text") {
      const jsonMatch = textContent.text.match(/\{[\s\S]*\}/);
      if (jsonMatch) {
        const match = JSON.parse(jsonMatch[0]);
        Object.assign(enriched, match);
      }
    }
  }
 
  return enriched;
}

import { Client } from "pg";
 
async function loadToPostgres(
  connectionString: string,
  tableName: string,
  records: Record<string, any>[],
  schema: Record<string, any>
): Promise<{ insertedRows: number; failedRows: number }> {
  const pgClient = new Client({ connectionString });
  await pgClient.connect();
 
  try {
    // テーブル作成（既存の場合はスキップ）
    const createTableSQL = generateCreateTableSQL(tableName, schema);
    await pgClient.query(createTableSQL);
 
    // COPY（高速バルク挿入）でデータロード
    const copySQL = generateCopySQL(tableName, schema);
    const copyStream = pgClient.query(
      new (require("pg").QueryStream)(copySQL)
    );
 
    let insertedRows = 0;
    let failedRows = 0;
 
    for (const record of records) {
      try {
        const values = schema.fields.map(
          (field: { name: string }) => record[field.name]
        );
        await pgClient.query(
          `INSERT INTO ${tableName} (${schema.fields.map((f: { name: string }) => f.name).join(", ")})
           VALUES (${schema.fields.map((_, i) => `$${i + 1}`).join(", ")})`,
          values
        );
        insertedRows++;
      } catch (err) {
        console.error(`Failed to insert record:`, err);
        failedRows++;
      }
    }
 
    return { insertedRows, failedRows };
  } finally {
    await pgClient.end();
  }
}
 
function generateCreateTableSQL(
  tableName: string,
  schema: Record<string, any>
): string {
  const columns = schema.fields
    .map((field: { name: string; type: string; nullable?: boolean }) => {
      const typeMap: Record<string, string> = {
        string: "TEXT",
        number: "NUMERIC",
        integer: "INTEGER",
        boolean: "BOOLEAN",
        datetime: "TIMESTAMP",
        date: "DATE",
      };
 
      const sqlType = typeMap[field.type] || "TEXT";
      const nullable = field.nullable !== false ? "" : " NOT NULL";
 
      return `"${field.name}" ${sqlType}${nullable}`;
    })
    .join(", ");
 
  return `
    CREATE TABLE IF NOT EXISTS ${tableName} (
      id SERIAL PRIMARY KEY,
      ${columns},
      created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
  `;
}
 
function generateCopySQL(
  tableName: string,
  schema: Record<string, any>
): string {
  const columns = schema.fields
    .map((f: { name: string }) => `"${f.name}"`)
    .join(", ");
  return `COPY ${tableName} (${columns}) FROM STDIN WITH (FORMAT csv, DELIMITER ',')`;
}

import { BigQuery } from "@google-cloud/bigquery";
 
async function loadToBigQuery(
  projectId: string,
  datasetId: string,
  tableId: string,
  records: Record<string, any>[],
  schema: Record<string, any>,
  mode: "INSERT" | "OVERWRITE"
) {
  const bigquery = new BigQuery({ projectId });
  const dataset = bigquery.dataset(datasetId);
  const table = dataset.table(tableId);
 
  const bqSchema = schema.fields.map((field: { name: string; type: string }) => {
    const typeMap: Record<string, string> = {
      string: "STRING",
      number: "FLOAT64",
      integer: "INT64",
      boolean: "BOOL",
      datetime: "TIMESTAMP",
      date: "DATE",
      object: "JSON",
    };
 
    return {
      name: field.name,
      type: typeMap[field.type] || "STRING",
      mode: "NULLABLE",
    };
  });
 
  const options = {
    schema: bqSchema,
    writeDisposition:
      mode === "OVERWRITE"
        ? "WRITE_TRUNCATE"
        : "WRITE_APPEND",
    skipLeadingRows: 0,
  };
 
  try {
    const [job] = await table.insert(records, options);
    await job.waitForCompletion();
 
    console.log(
      `Loaded ${records.length} rows to BigQuery table ${tableId}`
    );
 
    return { success: true, rowsLoaded: records.length };
  } catch (err) {
    console.error("BigQuery load failed:", err);
    throw err;
  }
}

import {
  S3Client,
  PutObjectCommand,
} from "@aws-sdk/client-s3";
 
async function loadToS3DataLake(
  bucket: string,
  records: Record<string, any>[],
  sourceSystem: string,
  partitionDate: string,
  awsAccessKey: string,
  awsSecretKey: string
) {
  const s3Client = new S3Client({
    region: "us-east-1",
    credentials: {
      accessKeyId: awsAccessKey,
      secretAccessKey: awsSecretKey,
    },
  });
 
  // パーティション階層: s3://bucket/source/year/month/day/hour/data.jsonl
  const [year, month, day] = partitionDate.split("-");
  const hour = new Date().getHours().toString().padStart(2, "0");
 
  const s3Key = `data-lake/${sourceSystem}/year=${year}/month=${month}/day=${day}/hour=${hour}/data_${Date.now()}.jsonl`;
 
  const jsonlData = records
    .map((r) => JSON.stringify(r))
    .join("\n");
 
  const putCommand = new PutObjectCommand({
    Bucket: bucket,
    Key: s3Key,
    Body: jsonlData,
    ContentType: "application/x-ndjson",
  });
 
  try {
    await s3Client.send(putCommand);
    console.log(`Loaded ${records.length} records to S3: s3://${bucket}/${s3Key}`);
    return { success: true, s3Path: s3Key };
  } catch (err) {
    console.error("S3 load failed:", err);
    throw err;
  }
}

interface AnomalyDetectionConfig {
  zscore_threshold?: number;
  iqr_multiplier?: number;
  pattern_rules?: Array<{
    field: string;
    pattern: string;
    description: string;
  }>;
}
 
async function detectAnomalies(
  records: Record<string, any>[],
  schema: Record<string, any>,
  config: AnomalyDetectionConfig = {}
): Promise<
  Array<{
    recordIndex: number;
    anomalies: Array<{ field: string; reason: string; severity: string }>;
  }>
> {
  const zscoreThreshold = config.zscore_threshold || 3;
  const anomalousRecords = [];
 
  // 数値フィールドの統計情報を計算
  const numericStats = computeNumericStats(records, schema);
 
  for (let idx = 0; idx < records.length; idx++) {
    const record = records[idx];
    const anomalies: Array<{
      field: string;
      reason: string;
      severity: string;
    }> = [];
 
    // Zスコアによる外れ値検知
    schema.fields.forEach((field: { name: string; type: string }) => {
      if (field.type === "number" && numericStats[field.name]) {
        const value = record[field.name];
        if (typeof value === "number") {
          const stats = numericStats[field.name];
          const zscore = Math.abs((value - stats.mean) / stats.stddev);
 
          if (zscore > zscoreThreshold) {
            anomalies.push({
              field: field.name,
              reason: `Z-score ${zscore.toFixed(2)} exceeds threshold ${zscoreThreshold}`,
              severity: zscore > zscoreThreshold * 1.5 ? "CRITICAL" : "WARNING",
            });
          }
        }
      }
    });
 
    // パターンベースのルール
    if (config.pattern_rules) {
      for (const rule of config.pattern_rules) {
        const value = record[rule.field];
        const regex = new RegExp(rule.pattern);
 
        if (value && !regex.test(String(value))) {
          anomalies.push({
            field: rule.field,
            reason: `Does not match pattern: ${rule.description}`,
            severity: "WARNING",
          });
        }
      }
    }
 
    if (anomalies.length > 0) {
      anomalousRecords.push({ recordIndex: idx, anomalies });
    }
  }
 
  return anomalousRecords;
}
 
function computeNumericStats(
  records: Record<string, any>[],
  schema: Record<string, any>
): Record<
  string,
  { mean: number; stddev: number; min: number; max: number }
> {
  const stats: Record<
    string,
    { mean: number; stddev: number; min: number; max: number }
  > = {};
 
  schema.fields.forEach((field: { name: string; type: string }) => {
    if (field.type === "number") {
      const values = records
        .map((r) => r[field.name])
        .filter((v) => typeof v === "number");
 
      if (values.length > 0) {
        const mean = values.reduce((a, b) => a + b, 0) / values.length;
        const variance =
          values.reduce((sum, v) => sum + Math.pow(v - mean, 2), 0) /
          values.length;
        const stddev = Math.sqrt(variance);
 
        stats[field.name] = {
          mean,
          stddev: stddev || 1,
          min: Math.min(...values),
          max: Math.max(...values),
        };
      }
    }
  });
 
  return stats;
}

async function validateBusinessRules(
  records: Record<string, any>[],
  businessRulesDescription: string
): Promise<Array<{ recordIndex: number; violations: string[] }>> {
  const sampleRecords = records.slice(0, 10);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a business rule validator. Check these sample records against the described business rules.
 
Business Rules:
${businessRulesDescription}
 
Sample Records to Validate:
${JSON.stringify(sampleRecords, null, 2)}
 
For each record, identify any rule violations. Return as JSON:
[
  {
    recordIndex: number,
    violations: ["violation description 1", "violation description 2"]
  }
]
 
Only include records with violations.`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 1024,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    return [];
  }
 
  const jsonMatch = textContent.text.match(/\[[\s\S]*\]/);
  if (!jsonMatch) {
    return [];
  }
 
  return JSON.parse(jsonMatch[0]);
}

interface PipelineStep {
  name: string;
  fn: (context: PipelineContext) => Promise<void>;
  retryConfig?: { maxRetries: number; backoffMs: number };
  timeout?: number;
  dependencies?: string[];
}
 
interface PipelineContext {
  sourceData: Record<string, any>[];
  schema: Record<string, any>;
  cleaningRules?: any[];
  cleanedData?: Record<string, any>[];
  validationResults?: any[];
  loadResults?: Record<string, any>;
  metadata: {
    startTime: Date;
    pipelineId: string;
    executionMode: "full" | "incremental";
  };
}
 
async function executePipelineWithRetry(
  step: PipelineStep,
  context: PipelineContext,
  attempt: number = 1
): Promise<void> {
  const maxRetries = step.retryConfig?.maxRetries || 3;
  const backoffMs = step.retryConfig?.backoffMs || 5000;
 
  try {
    const timeoutPromise = new Promise<void>((_, reject) => {
      setTimeout(() => reject(new Error("Step timeout")), step.timeout || 300000);
    });
 
    const executionPromise = step.fn(context);
 
    await Promise.race([executionPromise, timeoutPromise]);
  } catch (err) {
    if (attempt < maxRetries) {
      console.log(
        `Step "${step.name}" failed (attempt ${attempt}/${maxRetries}). Retrying in ${backoffMs}ms...`
      );
      await new Promise((resolve) => setTimeout(resolve, backoffMs));
      await executePipelineWithRetry(step, context, attempt + 1);
    } else {
      console.error(`Step "${step.name}" failed after ${maxRetries} attempts:`, err);
      throw err;
    }
  }
}
 
async function runFullPipeline(
  steps: PipelineStep[],
  initialContext: PipelineContext
) {
  const context = initialContext;
  const executedSteps = new Set<string>();
  const stepsByName = new Map(steps.map((s) => [s.name, s]));
 
  while (executedSteps.size < steps.length) {
    let stepExecuted = false;
 
    for (const step of steps) {
      if (executedSteps.has(step.name)) continue;
 
      const depsReady =
        !step.dependencies ||
        step.dependencies.every((dep) => executedSteps.has(dep));
 
      if (depsReady) {
        console.log(`Executing step: ${step.name}`);
        const startTime = Date.now();
 
        try {
          await executePipelineWithRetry(step, context);
          executedSteps.add(step.name);
 
          const duration = ((Date.now() - startTime) / 1000).toFixed(2);
          console.log(`✓ ${step.name} completed in ${duration}s`);
 
          stepExecuted = true;
        } catch (err) {
          console.error(`✗ ${step.name} failed:`, err);
          throw err;
        }
      }
    }
 
    if (!stepExecuted) {
      throw new Error(
        "Circular dependency or unsatisfiable dependencies detected"
      );
    }
  }
 
  console.log("Pipeline completed successfully");
  return context;
}
 
// 使用例
const pipelineSteps: PipelineStep[] = [
  {
    name: "fetch_data",
    fn: async (ctx) => {
      ctx.sourceData = await fetchApiData(
        "YOUR_API_ENDPOINT",
        "YOUR_API_KEY"
      );
    },
    timeout: 60000,
  },
  {
    name: "infer_schema",
    fn: async (ctx) => {
      ctx.schema = await inferSchema(ctx.sourceData, "api_source");
    },
    dependencies: ["fetch_data"],
  },
  {
    name: "generate_rules",
    fn: async (ctx) => {
      ctx.cleaningRules = await generateCleaningRules(
        ctx.schema,
        ctx.sourceData,
        "Business context"
      );
    },
    dependencies: ["infer_schema"],
  },
  {
    name: "clean_data",
    fn: async (ctx) => {
      ctx.cleanedData = ctx.sourceData.map((r) =>
        applyCleaningRules(r, ctx.cleaningRules || [])
      );
    },
    dependencies: ["generate_rules"],
  },
  {
    name: "validate_data",
    fn: async (ctx) => {
      ctx.validationResults = await detectAnomalies(
        ctx.cleanedData || [],
        ctx.schema
      );
    },
    dependencies: ["clean_data"],
    retryConfig: { maxRetries: 2, backoffMs: 3000 },
  },
  {
    name: "load_to_postgres",
    fn: async (ctx) => {
      ctx.loadResults = await loadToPostgres(
        "postgresql://user:pass@host/dbname",
        "data_table",
        ctx.cleanedData || [],
        ctx.schema
      );
    },
    dependencies: ["validate_data"],
  },
];

import cron from "node-cron";
 
interface ScheduleConfig {
  cronExpression: string;
  pipelineName: string;
  steps: PipelineStep[];
  alertOnFailure: {
    enabled: boolean;
    webhookUrl?: string;
    slackChannel?: string;
  };
}
 
class PipelineOrchestrator {
  private schedules = new Map<string, any>();
 
  scheduleRecurringPipeline(config: ScheduleConfig) {
    const job = cron.schedule(config.cronExpression, async () => {
      console.log(`Starting scheduled pipeline: ${config.pipelineName}`);
 
      const context: PipelineContext = {
        sourceData: [],
        schema: {},
        metadata: {
          startTime: new Date(),
          pipelineId: `${config.pipelineName}_${Date.now()}`,
          executionMode: "full",
        },
      };
 
      try {
        await runFullPipeline(config.steps, context);
 
        // 成功ログを記録
        this.logExecution(config.pipelineName, "SUCCESS", context);
      } catch (err) {
        console.error(`Pipeline failed: ${err}`);
 
        if (config.alertOnFailure.enabled) {
          await this.sendAlert(config, err as Error);
        }
 
        this.logExecution(config.pipelineName, "FAILED", context, err as Error);
      }
    });
 
    this.schedules.set(config.pipelineName, job);
    console.log(
      `Scheduled pipeline "${config.pipelineName}" with cron: ${config.cronExpression}`
    );
  }
 
  private async sendAlert(config: ScheduleConfig, error: Error) {
    if (config.alertOnFailure.webhookUrl) {
      await fetch(config.alertOnFailure.webhookUrl, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          pipeline: config.pipelineName,
          status: "FAILED",
          error: error.message,
          timestamp: new Date().toISOString(),
        }),
      });
    }
 
    if (config.alertOnFailure.slackChannel) {
      console.log(`Alert sent to Slack: ${config.alertOnFailure.slackChannel}`);
      // Slack SDK使用（省略）
    }
  }
 
  private logExecution(
    pipelineName: string,
    status: string,
    context: PipelineContext,
    error?: Error
  ) {
    const duration = Date.now() - context.metadata.startTime.getTime();
    console.log(
      `[${new Date().toISOString()}] ${pipelineName} - ${status} (${(duration / 1000).toFixed(2)}s)`
    );
 
    if (error) {
      console.error(`Error details: ${error.message}`);
    }
  }
 
  stopPipeline(pipelineName: string) {
    const job = this.schedules.get(pipelineName);
    if (job) {
      job.stop();
      this.schedules.delete(pipelineName);
      console.log(`Stopped pipeline: ${pipelineName}`);
    }
  }
}

// 推論結果をそのまま使わず、重要カラムは型を固定する
const lockedSchema = {
  ...inferredSchema,
  revenue: { type: "number", nullable: true, default: 0 },
  date: { type: "date", nullable: false },
  appId: { type: "string", nullable: false },
};

interface DataLineageRecord {
  recordId: string;
  sourceSystem: string;
  sourceTimestamp: Date;
  transformations: Array<{
    step: string;
    ruleId: string;
    timestamp: Date;
    changes?: Record<string, { before: any; after: any }>;
  }>;
  destinationSystems: Array<{
    system: string;
    loadTimestamp: Date;
    recordCount: number;
  }>;
  qualityScore: number;
}
 
async function trackDataLineage(
  record: Record<string, any>,
  sourceSystem: string,
  transformationHistory: Array<{
    step: string;
    ruleId: string;
  }>
): Promise<DataLineageRecord> {
  return {
    recordId: `${sourceSystem}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
    sourceSystem,
    sourceTimestamp: new Date(),
    transformations: transformationHistory.map((t) => ({
      step: t.step,
      ruleId: t.ruleId,
      timestamp: new Date(),
    })),
    destinationSystems: [],
    qualityScore: 0.95, // AIエージェントによる品質スコア
  };
}
 
interface PipelineMetrics {
  totalRecordsProcessed: number;
  successRate: number;
  averageProcessingTime: number;
  anomaliesDetected: number;
  validationFailures: number;
  dataQualityScore: number;
}
 
function computePipelineMetrics(
  context: PipelineContext
): PipelineMetrics {
  const duration =
    Date.now() - context.metadata.startTime.getTime();
  const totalRecords = context.sourceData.length;
  const anomalyCount =
    context.validationResults?.filter(
      (r) => r.anomalies?.length > 0
    ).length || 0;
 
  return {
    totalRecordsProcessed: totalRecords,
    successRate:
      context.loadResults?.insertedRows /
        (context.loadResults?.insertedRows + context.loadResults?.failedRows || 1) || 0,
    averageProcessingTime: duration / totalRecords,
    anomaliesDetected: anomalyCount,
    validationFailures: (context.validationResults?.length || 0),
    dataQualityScore:
      1 - (anomalyCount + (context.validationResults?.length || 0)) / totalRecords,
  };
}

async function processBatchesInParallel(
  records: Record<string, any>[],
  batchSize: number = 10000,
  processFunction: (batch: Record<string, any>[]) => Promise<Record<string, any>[]>,
  concurrency: number = 4
) {
  const batches = [];
  for (let i = 0; i < records.length; i += batchSize) {
    batches.push(records.slice(i, i + batchSize));
  }
 
  const results: Record<string, any>[] = [];
 
  // concurrencyレベルで並列処理
  for (let i = 0; i < batches.length; i += concurrency) {
    const batchGroup = batches.slice(i, i + concurrency);
    const groupResults = await Promise.all(
      batchGroup.map((batch) => processFunction(batch))
    );
 
    results.push(...groupResults.flat());
 
    console.log(
      `Processed ${Math.min((i + concurrency) * batchSize, records.length)} / ${records.length} records`
    );
  }
 
  return results;
}

import { Transform, pipeline } from "stream";
import { createReadStream } from "fs";
 
async function processLargeFileWithStream(
  inputPath: string,
  outputPath: string,
  transformFn: (record: Record<string, any>) => Promise<Record<string, any>>
) {
  const inputStream = createReadStream(inputPath, { encoding: "utf8" });
 
  const transformStream = new Transform({
    objectMode: true,
    async transform(chunk: string, _, callback) {
      try {
        const record = JSON.parse(chunk.trim());
        const transformed = await transformFn(record);
        callback(null, JSON.stringify(transformed) + "\n");
      } catch (err) {
        callback(err);
      }
    },
  });
 
  const outputStream = createReadStream(outputPath, { encoding: "utf8" });
 
  return new Promise((resolve, reject) => {
    pipeline(inputStream, transformStream, outputStream, (err) => {
      if (err) reject(err);
      else resolve(null);
    });
  });
}