プロジェクト最適化とリリース — DataPipeline

> 出力： ローカルの DartPad または `dart run` で実行してください。この Dart コースの全例は Dart 3.x / Flutter 3.x ベースです。SDK バージョンにより結果が多少異なる場合があります。

import 'dart:isolate';

// 悪い例：静的チャンキング - 作業が偏る
List<List<T>> staticChunk<T>(List<T> data, int chunks) {
  final size = (data.length / chunks).ceil();
  return [
    for (var i = 0; i < chunks; i++)
      data.sublist(i * size, ((i + 1) * size).clamp(0, data.length))
  ];
}

// 良い例：SendPort による動的ワークスティーリング
class DynamicIsolatePool {
  final int _poolSize;
  var _pendingChunks = <List<Map<String, dynamic>>>[];
  final _results = <AnalysisResult>[];
  var _activeWorkers = 0;
  late Completer<List<AnalysisResult>> _completer;

  DynamicIsolatePool({int poolSize = 4}) : _poolSize = poolSize;

  Future<List<AnalysisResult>> process(
    List<Map<String, dynamic>> allOrders, {
    int chunkSize = 50000,
  }) async {
    _pendingChunks = [
      for (var i = 0; i < allOrders.length; i += chunkSize)
        allOrders.sublist(i, (i + chunkSize).clamp(0, allOrders.length))
    ];

    _completer = Completer<List<AnalysisResult>>();
    _activeWorkers = 0;

    final workers = <_Worker>[];
    for (var i = 0; i < _poolSize && _pendingChunks.isNotEmpty; i++) {
      final chunk = _pendingChunks.removeAt(0);
      workers.add(await _spawnWorker(chunk));
      _activeWorkers++;
    }

    return _completer.future;
  }

  Future<_Worker> _spawnWorker(List<Map<String, dynamic>> initialChunk) async {
    final receivePort = ReceivePort();
    final isolate = await Isolate.spawn(_workerEntry, receivePort.sendPort);
    final sendPort = await receivePort.first as SendPort;

    final responsePort = ReceivePort();
    sendPort.send(_WorkMessage(data: initialChunk, responsePort: responsePort.sendPort));

    responsePort.listen((result) {
      _results.add(result as AnalysisResult);
      _activeWorkers--;

      if (_pendingChunks.isNotEmpty) {
        final nextChunk = _pendingChunks.removeAt(0);
        sendPort.send(_WorkMessage(data: nextChunk, responsePort: responsePort.sendPort));
        _activeWorkers++;
      } else if (_activeWorkers == 0) {
        isolate.kill(priority: Isolate.immediate);
        receivePort.close();
        responsePort.close();
        _completer.complete(_results);
      }
    });

    return _Worker(isolate: isolate, sendPort: sendPort);
  }

  static void _workerEntry(SendPort mainSendPort) {
    final receivePort = ReceivePort();
    mainSendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message is _WorkMessage) {
        final orders = message.data.map((m) => Order(
          id: m['id'] as String,
          amount: (m['amount'] as num).toDouble(),
          status: m['status'] as String,
          category: m['category'] as String,
        )).toList();

        final result = const OrderAnalyzer().analyze(orders);
        message.responsePort.send(result);
      }
    });
  }
}

class _Worker {
  final Isolate isolate;
  final SendPort sendPort;
  const _Worker({required this.isolate, required this.sendPort});
}

class _WorkMessage {
  final List<Map<String, dynamic>> data;
  final SendPort responsePort;
  const _WorkMessage({required this.data, required this.responsePort});
}

import 'dart:async';

class BackpressurePipeline {
  final DataSource _source;
  final OrderParser _parser;
  final int _maxConcurrent;

  BackpressurePipeline({
    required DataSource source,
    required OrderParser parser,
    int maxConcurrent = 1000,
  })  : _source = source,
        _parser = parser,
        _maxConcurrent = maxConcurrent;

  Future<AnalysisResult> execute() async {
    final orders = <Order>[];
    var pending = 0;
    final completer = Completer<void>();

    final subscription = _source.readLines().listen(
      (line) {
        try {
          orders.add(_parser.parse(line));
        } on FormatException {
          return;
        }

        pending++;
        if (pending >= _maxConcurrent) {
          subscription.pause();
        }
      },
      onDone: () {
        completer.complete();
      },
      onError: (e) {
        completer.completeError(e);
      },
    );

    await completer.future;

    return const OrderAnalyzer().analyze(orders);
  }

  // バッファ制御付きのポーズ可能な Stream
  Stream<Order> streamWithBackpressure() async* {
    var buffer = <Order>[];

    await for (final line in _source.readLines()) {
      try {
        buffer.add(_parser.parse(line));
      } on FormatException {
        continue;
      }

      if (buffer.length >= _maxConcurrent) {
        for (final order in buffer) {
          yield order;
        }
        buffer = [];
      }
    }

    for (final order in buffer) {
      yield order;
    }
  }
}

import 'dart:math';

class MemoryOptimizer {
  // 可能な場合は List ではなく Iterable を使用
  static Iterable<Order> lazyParse(Iterable<String> lines, OrderParser parser) sync* {
    for (final line in lines) {
      try {
        yield parser.parse(line);
      } on FormatException {
        continue;
      }
    }
  }

  // メモリ制限のためのチャンク処理
  static Future<AnalysisResult> processInChunks(
    Stream<String> lines,
    OrderParser parser, {
    int chunkSize = 100000,
  }) async {
    var chunk = <Order>[];
    var totalResult = _EmptyAnalysisResult();

    await for (final line in lines) {
      try {
        chunk.add(parser.parse(line));
      } on FormatException {
        continue;
      }

      if (chunk.length >= chunkSize) {
        final partial = const OrderAnalyzer().analyze(chunk);
        totalResult = totalResult.merge(partial);
        chunk = [];
      }
    }

    if (chunk.isNotEmpty) {
      final partial = const OrderAnalyzer().analyze(chunk);
      totalResult = totalResult.merge(partial);
    }

    return totalResult.toAnalysisResult();
  }
}

class _EmptyAnalysisResult {
  int totalOrders = 0;
  int completedOrders = 0;
  double revenue = 0;
  double tax = 0;
  final Map<String, double> revenueByCategory = {};

  _EmptyAnalysisResult merge(AnalysisResult other) {
    totalOrders += other.totalOrders;
    completedOrders += other.completedOrders;
    revenue += other.revenue;
    tax += other.tax;
    for (final entry in other.revenueByCategory.entries) {
      revenueByCategory.update(entry.key, (v) => v + entry.value, ifAbsent: () => entry.value);
    }
    return this;
  }

  AnalysisResult toAnalysisResult() => AnalysisResult(
    totalOrders: totalOrders,
    completedOrders: completedOrders,
    revenue: revenue,
    tax: tax,
    total: revenue + tax,
    revenueByCategory: Map.unmodifiable(revenueByCategory),
  );
}

# デバッグビルド（大きい、デバッグ情報を含む）
dart compile exe bin/data_pipeline.dart -o build/data_pipeline_debug

# リリースビルド（最適化済み、サイズが小さい）
dart compile exe bin/data_pipeline.dart -o build/data_pipeline \
  --define=MODE=release

# バイナリサイズを確認
ls -lh build/

# デバッグシンボルを削除（追加のサイズ削減）
# Linux/macOS の場合：
strip build/data_pipeline

// ツリーシェイク用の条件付きインポート
const _isRelease = bool.fromEnvironment('MODE.release');

void log(String message) {
  if (!_isRelease) {
    print('[DEBUG] $message');
  }
}

// このクラス全体はリリースモードでツリーシェイクされる
class DebugLogger {
  void log(String message) {
    if (!_isRelease) {
      print('[DEBUG] ${DateTime.now()}: $message');
    }
  }
}

# .github/workflows/ci.yml
name: DataPipeline CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  analyze-and-test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        sdk: [stable, beta]

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1
        with:
          sdk: ${{ matrix.sdk }}

      - name: Install dependencies
        run: dart pub get

      - name: Verify formatting
        run: dart format --output=none --set-exit-if-changed .

      - name: Analyze code
        run: dart analyze --fatal-infos

      - name: Run tests
        run: dart test --coverage=coverage

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: coverage/lcov.info

  build-and-release:
    needs: analyze-and-test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Install dependencies
        run: dart pub get

      - name: Compile AOT binary
        run: |
          dart compile exe bin/data_pipeline.dart -o build/data_pipeline

      - name: Run integration test
        run: |
          ./build/data_pipeline analyze --source csv --input test/fixtures/sample.csv

      - name: Upload binary artifact
        uses: actions/upload-artifact@v4
        with:
          name: data_pipeline-linux
          path: build/data_pipeline

  publish:
    needs: build-and-release
    runs-on: ubuntu-latest
    if: startsWith(github.ref, 'refs/tags/v')

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Install dependencies
        run: dart pub get

      - name: Publish to pub.dev
        run: dart pub publish --force
        env:
          PUB_CREDENTIALS: ${{ secrets.PUB_CREDENTIALS }}

name: data_pipeline
description: E-commerce data analytics CLI tool - process million-level orders with Stream + Isolate
version: 1.0.0

homepage: https://github.com/bob/datapipeline
repository: https://github.com/bob/datapipeline
issue_tracker: https://github.com/bob/datapipeline/issues

environment:
  sdk: ^3.0.0

dependencies:
  args: ^2.4.2
  http: ^1.1.0
  csv: ^6.0.0

dev_dependencies:
  test: ^1.24.0
  build_runner: ^2.4.0
  json_serializable: ^6.7.0

executables:
  data_pipeline: data_pipeline

# CHANGELOG

## 7. リリースノート

### (1) 1.0.0 - 2025-01-15

#### 機能
- CSV と JSON データソースのサポート
- バックプレッシャー付き Stream ベースパイプライン
- 並列処理用 Isolate プール
- analyze/export/validate サブコマンド付き CLI
- パターンマッチング + シールクラスアーキテクチャ

### (2) パフォーマンス
- 100 万件の注文を約 8 秒で処理（4 コア）
- 100 万件のメモリ使用量 200MB 以下

### (3) テスト
- 95% コードカバレッジ
- ユニット + 統合 + E2E テスト

# 公開前チェックリスト

# 1. 公開ドライラン
dart pub publish --dry-run

# 2. pub.dev でスコアを確認
dart pub publish --dry-run 2>&1 | grep -E "score|warning|error"

# 3. API ドキュメントを生成
dart doc .

# 4. パッケージ内容を検証
dart pub pack  # 検査用の .tar.gz を作成

# 5. 最終公開
dart pub publish

/// EC 注文データを分析して統計結果を生成します。
///
/// [OrderAnalyzer] をデータ分析の主要なエントリーポイントとして使用してください。
/// 対象地域に基づいて [taxRate] を設定してください。
///
/// 例：
/// ```dart
/// final analyzer = OrderAnalyzer(taxRate: 0.08);
/// final result = analyzer.analyze(orders);
/// print('Revenue: \$${result.revenue} USD');
/// ```
///
/// 関連項目：
/// - 出力構造については [AnalysisResult]
/// - 入力モデルについては [Order]
class OrderAnalyzer {
  /// 指定された [taxRate] でアナライザーを作成します。
  ///
  /// デフォルトは 0.08（米国税率 8%）。
  /// 地域別税率を使用：US=0.08、EU=0.20、UK=0.15。
  const OrderAnalyzer({this.taxRate = 0.08});

  /// 完了済み注文に適用される税率。
  final double taxRate;

  /// [orders] のリストを分析して集約された統計を返します。
  ///
  /// ステータスが 'completed' の注文のみが売上計算に含まれます。
  /// 他のステータスの注文は [AnalysisResult.totalOrders] でカウントされますが、
  /// 売上からは除外されます。
  AnalysisResult analyze(List<Order> orders) {
    // ... 実装
  }
}

// ============================================
// DataPipeline パフォーマンスベンチマーク
// Alice のエンドツーエンドストレステスト
// ============================================

import 'dart:async';
import 'dart:io';

class Benchmark {
  final String name;
  final Stopwatch _stopwatch;

  Benchmark(this.name) : _stopwatch = Stopwatch();

  Future<T> run<T>(Future<T> Function() action) async {
    _stopwatch.reset();
    _stopwatch.start();
    final result = await action();
    _stopwatch.stop();
    print('$name: ${_stopwatch.elapsedMilliseconds}ms');
    return result;
  }
}

// テストデータを生成
List<String> generateOrders(int count) {
  final categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final regions = ['US', 'EU', 'UK', 'JP'];

  return List.generate(count, (i) {
    final category = categories[i % categories.length];
    final status = statuses[i % statuses.length];
    final region = regions[i % regions.length];
    final amount = (100 + (i * 37) % 5000).toDouble();
    return 'ORD-${(i + 1).toString().padLeft(7, '0')},$amount,$status,$category,$region';
  });
}

void main() async {
  print('=== DataPipeline パフォーマンスベンチマーク ===\n');

  final sizes = [10000, 100000, 1000000];

  for (final size in sizes) {
    final label = size >= 1000000 ? '${size ~/ 1000000}M' : '${size ~/ 1000}K';
    print('--- $label 件の注文 ---');

    // データを生成して一時ファイルに保存
    final data = generateOrders(size);
    final tempFile = File('${Directory.systemTemp.path}/bench_${label}.csv');
    await tempFile.writeAsString(data.join('\n'));

    // ベンチマーク 1: Stream 処理
    final streamResult = await Benchmark('Stream ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      final parser = const CsvOrderParser();
      final pipeline = Pipeline(source: source, parser: parser);
      return pipeline.execute();
    });

    // ベンチマーク 2: メモリ効率の良いチャンク処理
    final chunkedResult = await Benchmark('Chunked ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      return MemoryOptimizer.processInChunks(
        source.readLines(),
        const CsvOrderParser(),
        chunkSize: 50000,
      );
    });

    // 結果の一致を検証
    final streamRev = streamResult.revenue.toStringAsFixed(2);
    final chunkedRev = chunkedResult.revenue.toStringAsFixed(2);
    print('売上一致: ${streamRev == chunkedRev ? "PASS" : "FAIL"}');
    print('売上: \$${streamRev} USD\n');

    await tempFile.delete();
  }

  // ベンチマーク 3: AOT バイナリサイズチェック
  print('--- ビルド成果物 ---');
  final exe = File('build/data_pipeline');
  if (await exe.exists()) {
    final sizeKB = await exe.length() ~/ 1024;
    print('AOT バイナリ: ${sizeKB}KB');
  } else {
    print('AOT バイナリ: 未ビルド（実行: dart compile exe bin/data_pipeline.dart）');
  }

  print('\n=== ベンチマーク完了 ===');
  print('Alice 受け入れ基準: 100 万件 < 10 秒、メモリ < 300MB、バイナリ < 20MB');
}

// ---- 必要なスタブ（前のレッスンからインポート） ----

sealed class DataSource {
  const DataSource();
  String get displayName;
  Stream<String> readLines();
}

class CsvSource extends DataSource {
  final String path;
  const CsvSource({required this.path});
  @override
  String get displayName => 'CSV: $path';
  @override
  Stream<String> readLines() =>
    File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

abstract class OrderParser {
  const OrderParser();
  Order parse(String raw);
}

class CsvOrderParser extends OrderParser {
  const CsvOrderParser();
  @override
  Order parse(String raw) {
    final parts = raw.split(',');
    if (parts.length < 4) throw FormatException('Invalid row');
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
      region: parts.length > 4 ? parts[4].trim() : 'US',
    );
  }
}

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;
  final String region;
  const Order({required this.id, required this.amount, required this.status, required this.category, this.region = 'US'});
  double get tax => amount * 0.08;
}

class AnalysisResult {
  final int totalOrders;
  final int completedOrders;
  final double revenue;
  final double tax;
  final double total;
  final Map<String, double> revenueByCategory;
  const AnalysisResult({required this.totalOrders, required this.completedOrders, required this.revenue, required this.tax, required this.total, required this.revenueByCategory});
}

class OrderAnalyzer {
  const OrderAnalyzer();
  AnalysisResult analyze(List<Order> orders) {
    final completed = orders.where((o) => o.status == 'completed').toList();
    final revenue = completed.fold<double>(0, (s, o) => s + o.amount);
    final tax = completed.fold<double>(0, (s, o) => s + o.tax);
    final byCategory = <String, double>{};
    for (final o in completed) {
      byCategory.update(o.category, (v) => v + o.amount, ifAbsent: () => o.amount);
    }
    return AnalysisResult(totalOrders: orders.length, completedOrders: completed.length, revenue: revenue, tax: tax, total: revenue + tax, revenueByCategory: byCategory);
  }
}

class Pipeline {
  final DataSource source;
  final OrderParser parser;
  const Pipeline({required this.source, required this.parser});
  Future<AnalysisResult> execute() async {
    final orders = <Order>[];
    await for (final line in source.readLines()) {
      try { orders.add(parser.parse(line)); } on FormatException { continue; }
    }
    return const OrderAnalyzer().analyze(orders);
  }
}

class MemoryOptimizer {
  static Future<AnalysisResult> processInChunks(Stream<String> lines, OrderParser parser, {int chunkSize = 50000}) async {
    var chunk = <Order>[];
    int totalOrders = 0, completedOrders = 0;
    double revenue = 0, tax = 0;
    final byCategory = <String, double>{};

    await for (final line in lines) {
      try { chunk.add(parser.parse(line)); } on FormatException { continue; }
      if (chunk.length >= chunkSize) {
        final r = const OrderAnalyzer().analyze(chunk);
        totalOrders += r.totalOrders;
        completedOrders += r.completedOrders;
        revenue += r.revenue;
        tax += r.tax;
        for (final e in r.revenueByCategory.entries) {
          byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
        }
        chunk = [];
      }
    }
    if (chunk.isNotEmpty) {
      final r = const OrderAnalyzer().analyze(chunk);
      totalOrders += r.totalOrders;
      completedOrders += r.completedOrders;
      revenue += r.revenue;
      tax += r.tax;
      for (final e in r.revenueByCategory.entries) {
        byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
      }
    }
    return AnalysisResult(totalOrders: totalOrders, completedOrders: completedOrders, revenue: revenue, tax: tax, total: revenue + tax, revenueByCategory: byCategory);
  }
}