データ基盤の構築において、データパイプラインの全体像を把握することは非常に重要です。しかし、複雑なデータフローを言葉だけで説明するのは難しく、チームメンバー間での認識のズレが生じやすくなります。

今回は、Mermaid を使ってデータ基盤のラインジ（データパイプライン）を図解する方法をご紹介します。ETL プロセス、DAG（有向非巡回グラフ）、品質チェックの全体像を視覚的に表現することで、データエンジニアリングチーム全体での理解を深められるでしょう。

背景

データ基盤を構築する際、データの流れは複数のステージを経て処理されます。

データは通常、以下のような流れで処理されます：

1. データソースからの抽出（Extract）
2. データの変換（Transform）
3. データウェアハウスへの格納（Load）
4. データ品質のチェック
5. 分析基盤への提供

これらのプロセスは相互に依存しており、適切な順序で実行される必要があります。

以下の図は、データ基盤における基本的なデータフローを示しています。

mermaidCopy codeflowchart LR
  source1[("データソース1<br/>MySQL")]
  source2[("データソース2<br/>API")]
  source3[("データソース3<br/>CSV")]

  extract["Extract<br/>データ抽出"]
  transform["Transform<br/>データ変換"]
  load["Load<br/>データ格納"]

  dwh[("データウェア<br/>ハウス")]

  source1 --> extract
  source2 --> extract
  source3 --> extract
  extract --> transform
  transform --> load
  load --> dwh

markdownCopy code```mermaid
flowchart LR
  source1[("データソース1<br/>MySQL")]
  source2[("データソース2<br/>API")]
  source3[("データソース3<br/>CSV")]

  extract["Extract<br/>データ抽出"]
  transform["Transform<br/>データ変換"]
  load["Load<br/>データ格納"]

  dwh[("データウェア<br/>ハウス")]

  source1 --> extract
  source2 --> extract
  source3 --> extract
  extract --> transform
  transform --> load
  load --> dwh
```

この図から、複数のデータソースから抽出されたデータが、変換・格納というステップを経てデータウェアハウスに集約される流れが理解できますね。

データ基盤の主要コンポーネント

データ基盤は、以下の主要なコンポーネントで構成されます。

#	コンポーネント	役割	例
1	データソース	生データの提供元	MySQL、API、CSV ファイル
2	ETL パイプライン	データの抽出・変換・格納	Apache Airflow、dbt
3	データウェアハウス	分析用データの保存	BigQuery、Snowflake
4	品質チェック機構	データ品質の検証	Great Expectations
5	オーケストレーター	処理の順序制御	Airflow、Prefect

これらのコンポーネントが連携することで、信頼性の高いデータ基盤が実現されるのです。

課題

データ基盤の設計と運用には、いくつかの課題があります。

複雑なデータフローの可視化

データパイプラインが複雑化すると、以下のような問題が発生します：

• 依存関係の把握が困難：どのタスクがどのタスクに依存しているか分かりにくい
• 障害時の影響範囲が不明：あるタスクが失敗した際、どこまで影響するか予測できない
• 新規メンバーのオンボーディングに時間がかかる：全体像の理解に多くの時間が必要

以下の図は、複雑化したデータパイプラインの依存関係を示しています。

mermaidCopy codeflowchart TD
  task1["タスクA<br/>顧客データ抽出"]
  task2["タスクB<br/>注文データ抽出"]
  task3["タスクC<br/>在庫データ抽出"]
  task4["タスクD<br/>データ統合"]
  task5["タスクE<br/>集計処理"]
  task6["タスクF<br/>レポート生成"]
  task7["タスクG<br/>品質チェック"]

  task1 --> task4
  task2 --> task4
  task3 --> task4
  task4 --> task7
  task7 --> task5
  task5 --> task6

markdownCopy code```mermaid
flowchart TD
  task1["タスクA<br/>顧客データ抽出"]
  task2["タスクB<br/>注文データ抽出"]
  task3["タスクC<br/>在庫データ抽出"]
  task4["タスクD<br/>データ統合"]
  task5["タスクE<br/>集計処理"]
  task6["タスクF<br/>レポート生成"]
  task7["タスクG<br/>品質チェック"]

  task1 --> task4
  task2 --> task4
  task3 --> task4
  task4 --> task7
  task7 --> task5
  task5 --> task6
```

この図を見ると、タスク間の依存関係が複雑に絡み合っていることが分かります。タスク D が失敗すると、その後の全てのタスクに影響が及びますね。

データ品質の保証

データ品質を保証するには、以下の課題に対処する必要があります：

#	課題	影響	対処の必要性
1	データの欠損	分析結果の信頼性低下	★★★
2	データ型の不一致	パイプラインの停止	★★★
3	重複データ	集計値の誤り	★★☆
4	異常値の混入	ビジネス判断の誤り	★★★
5	スキーマの変更	既存処理の破綻	★★★

これらの課題を適切に検出し、対処する仕組みが必要となります。

パイプラインの実行管理

データパイプラインの実行には、以下のような管理上の課題があります：

• 実行スケジュールの調整：各タスクをいつ実行するか
• リトライ制御：失敗時の再実行ロジック
• リソース管理：計算リソースの効率的な利用
• 監視とアラート：異常の早期検知

これらを適切に管理しないと、データの鮮度が保たれず、ビジネスに必要なタイミングでデータが提供できなくなってしまいます。

解決策

Mermaid を使ってデータ基盤のラインジを図解することで、これらの課題を解決できます。

ETL プロセスの可視化

ETL（Extract、Transform、Load）プロセスを Mermaid で図解することで、データの流れが明確になります。

以下は、実際の ETL プロセスを表現した図です。

mermaidCopy codeflowchart TB
  subgraph extract["Extract（抽出）"]
    db1[("MySQL<br/>顧客DB")]
    db2[("PostgreSQL<br/>注文DB")]
    api1["外部API<br/>在庫情報"]
  end

  subgraph transform["Transform（変換）"]
    clean["データクレンジング"]
    normalize["正規化処理"]
    enrich["データ統合"]
  end

  subgraph loadProcess["Load（格納）"]
    stage[("ステージング<br/>エリア")]
    dwh[("データウェア<br/>ハウス")]
  end

  db1 --> clean
  db2 --> clean
  api1 --> clean
  clean --> normalize
  normalize --> enrich
  enrich --> stage
  stage --> dwh

markdownCopy code```mermaid
flowchart TB
  subgraph extract["Extract（抽出）"]
    db1[("MySQL<br/>顧客DB")]
    db2[("PostgreSQL<br/>注文DB")]
    api1["外部API<br/>在庫情報"]
  end

  subgraph transform["Transform（変換）"]
    clean["データクレンジング"]
    normalize["正規化処理"]
    enrich["データ統合"]
  end

  subgraph loadProcess["Load（格納）"]
    stage[("ステージング<br/>エリア")]
    dwh[("データウェア<br/>ハウス")]
  end

  db1 --> clean
  db2 --> clean
  api1 --> clean
  clean --> normalize
  normalize --> enrich
  enrich --> stage
  stage --> dwh
```

この図から、各ステージでのデータ処理の流れが一目で理解できますね。抽出、変換、格納の 3 つのフェーズが明確に分離されています。

DAG（有向非巡回グラフ）による依存関係の表現

Apache Airflow などのワークフローエンジンでは、タスクの依存関係を DAG として表現します。

DAG を Mermaid で表現すると、以下のようになります：

mermaidCopy codeflowchart LR
  start["開始"]
  extract1["顧客データ<br/>抽出"]
  extract2["注文データ<br/>抽出"]
  validate1["データ検証1"]
  validate2["データ検証2"]
  merge["データ統合"]
  transform1["集計処理"]
  quality["品質チェック"]
  load_data["DWH格納"]
  finish["完了"]

  start --> extract1
  start --> extract2
  extract1 --> validate1
  extract2 --> validate2
  validate1 --> merge
  validate2 --> merge
  merge --> transform1
  transform1 --> quality
  quality --> load_data
  load_data --> finish

markdownCopy code```mermaid
flowchart LR
  start["開始"]
  extract1["顧客データ<br/>抽出"]
  extract2["注文データ<br/>抽出"]
  validate1["データ検証1"]
  validate2["データ検証2"]
  merge["データ統合"]
  transform1["集計処理"]
  quality["品質チェック"]
  load_data["DWH格納"]
  finish["完了"]

  start --> extract1
  start --> extract2
  extract1 --> validate1
  extract2 --> validate2
  validate1 --> merge
  validate2 --> merge
  merge --> transform1
  transform1 --> quality
  quality --> load_data
  load_data --> finish
```

この DAG 図により、並列実行可能なタスク（extract1 と extract2）と、順次実行が必要なタスクが明確に区別できます。

図で理解できる要点：

• 並列実行可能なタスクを識別できる
• タスク間の依存関係が視覚的に把握できる
• ボトルネックとなるタスクを特定しやすい

データ品質チェックの組み込み

データパイプラインにおける品質チェックのポイントを図解します。

以下は、品質チェックを組み込んだデータフローです：

mermaidCopy codeflowchart TD
  input[("入力データ")]
  schema_check["スキーマ検証"]
  null_check["NULL値チェック"]
  range_check["値範囲チェック"]
  dup_check["重複チェック"]

  pass_check["検証OK"]
  fail_check["検証NG"]

  alert["アラート通知"]
  retry["リトライ処理"]
  manual["手動対応"]

  next_process["次の処理へ"]

  input --> schema_check
  schema_check -->|OK| null_check
  schema_check -->|NG| fail_check
  null_check -->|OK| range_check
  null_check -->|NG| fail_check
  range_check -->|OK| dup_check
  range_check -->|NG| fail_check
  dup_check -->|OK| pass_check
  dup_check -->|NG| fail_check

  pass_check --> next_process
  fail_check --> alert
  alert --> retry
  retry -->|再実行| input
  retry -->|失敗継続| manual

markdownCopy code```mermaid
flowchart TD
  input[("入力データ")]
  schema_check["スキーマ検証"]
  null_check["NULL値チェック"]
  range_check["値範囲チェック"]
  dup_check["重複チェック"]

  pass_check["検証OK"]
  fail_check["検証NG"]

  alert["アラート通知"]
  retry["リトライ処理"]
  manual["手動対応"]

  next_process["次の処理へ"]

  input --> schema_check
  schema_check -->|OK| null_check
  schema_check -->|NG| fail_check
  null_check -->|OK| range_check
  null_check -->|NG| fail_check
  range_check -->|OK| dup_check
  range_check -->|NG| fail_check
  dup_check -->|OK| pass_check
  dup_check -->|NG| fail_check

  pass_check --> next_process
  fail_check --> alert
  alert --> retry
  retry -->|再実行| input
  retry -->|失敗継続| manual
```

品質チェックを多段階で実施することで、データの信頼性を高めることができます。エラー発生時のリトライ処理や手動対応へのエスカレーションフローも明確ですね。

パイプラインの全体アーキテクチャ

データ基盤全体のアーキテクチャを、レイヤー構造で表現します。

mermaidCopy codeflowchart TB
  subgraph source_layer["ソースレイヤー"]
    src1[("業務DB")]
    src2[("外部API")]
    src3[("ファイル")]
  end

  subgraph ingestion_layer["インジェストレイヤー"]
    batch["バッチ取り込み"]
    stream["ストリーム取り込み"]
  end

  subgraph processing_layer["処理レイヤー"]
    raw[("Raw層")]
    staging[("Staging層")]
    curated[("Curated層")]
  end

  subgraph quality_layer["品質レイヤー"]
    validation["検証処理"]
    monitoring["モニタリング"]
  end

  subgraph consumption_layer["利用レイヤー"]
    bi["BIツール"]
    ml["機械学習"]
    app["アプリケーション"]
  end

  src1 --> batch
  src2 --> stream
  src3 --> batch

  batch --> raw
  stream --> raw

  raw --> validation
  validation --> staging
  staging --> validation
  validation --> curated

  validation -.-> monitoring

  curated --> bi
  curated --> ml
  curated --> app

markdownCopy code```mermaid
flowchart TB
  subgraph source_layer["ソースレイヤー"]
    src1[("業務DB")]
    src2[("外部API")]
    src3[("ファイル")]
  end

  subgraph ingestion_layer["インジェストレイヤー"]
    batch["バッチ取り込み"]
    stream["ストリーム取り込み"]
  end

  subgraph processing_layer["処理レイヤー"]
    raw[("Raw層")]
    staging[("Staging層")]
    curated[("Curated層")]
  end

  subgraph quality_layer["品質レイヤー"]
    validation["検証処理"]
    monitoring["モニタリング"]
  end

  subgraph consumption_layer["利用レイヤー"]
    bi["BIツール"]
    ml["機械学習"]
    app["アプリケーション"]
  end

  src1 --> batch
  src2 --> stream
  src3 --> batch

  batch --> raw
  stream --> raw

  raw --> validation
  validation --> staging
  staging --> validation
  validation --> curated

  validation -.-> monitoring

  curated --> bi
  curated --> ml
  curated --> app
```

この図は、データ基盤を 5 つのレイヤーに分割し、各レイヤーの役割と連携を示しています。品質レイヤーが各処理段階を監視する構造が理解できますね。

具体例

実際のデータ基盤構築のシナリオを通じて、Mermaid による図解の活用例を見ていきましょう。

シナリオ：EC サイトのデータ基盤構築

EC サイトの注文データ、顧客データ、在庫データを統合し、日次でレポートを作成するデータ基盤を構築します。

データソースの定義

まず、データソースと抽出方法を定義します。

typescriptCopy code// データソースの設定
interface DataSource {
  name: string; // データソース名
  type: string; // データベースの種類
  connection: string; // 接続文字列
  extractSchedule: string; // 抽出スケジュール
}

// 顧客データソースの設定
const customerSource: DataSource = {
  name: 'customer_db',
  type: 'MySQL',
  connection: 'mysql://user:pass@host:3306/customers',
  extractSchedule: '0 1 * * *', // 毎日午前1時
};

このコードでは、データソースの接続情報と抽出スケジュールを定義しています。

typescriptCopy code// 注文データソースの設定
const orderSource: DataSource = {
  name: 'order_db',
  type: 'PostgreSQL',
  connection: 'postgresql://user:pass@host:5432/orders',
  extractSchedule: '0 1 * * *', // 毎日午前1時
};

// 在庫データソースの設定（API経由）
const inventorySource = {
  name: 'inventory_api',
  type: 'REST_API',
  endpoint: 'https://api.inventory.example.com/v1/stock',
  apiKey: 'YOUR_API_KEY',
  extractSchedule: '0 2 * * *', // 毎日午前2時
};

複数のデータソースから、それぞれ異なるタイミングでデータを抽出する設定です。

Airflow DAG の実装

Apache Airflow を使って、データパイプラインを実装します。

pythonCopy code# Airflow DAG の基本設定
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

# DAG のデフォルト引数
default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,  # 3回までリトライ
    'retry_delay': timedelta(minutes=5)
}

このコードでは、DAG の基本設定として、所有者、開始日、リトライ設定などを定義しています。

pythonCopy code# DAG の定義
dag = DAG(
    'ec_data_pipeline',
    default_args=default_args,
    description='EC サイトのデータ統合パイプライン',
    schedule_interval='0 3 * * *',  # 毎日午前3時実行
    catchup=False,
    tags=['production', 'daily']
)

DAG 全体のスケジュールを午前 3 時に設定しています。これは、各データソースからの抽出が完了した後のタイミングですね。

タスクの定義

各処理ステップをタスクとして定義します。

pythonCopy code# データ抽出タスク
def extract_customer_data(**context):
    """顧客データを抽出する"""
    # MySQLからデータ抽出処理
    print("顧客データを抽出中...")
    # 実際の抽出ロジック
    return "customer_data_extracted"

extract_customers = PythonOperator(
    task_id='extract_customers',
    python_callable=extract_customer_data,
    dag=dag
)

顧客データを抽出するタスクです。実際の処理では、MySQL に接続してデータを取得します。

pythonCopy code# 注文データ抽出タスク
def extract_order_data(**context):
    """注文データを抽出する"""
    print("注文データを抽出中...")
    return "order_data_extracted"

extract_orders = PythonOperator(
    task_id='extract_orders',
    python_callable=extract_order_data,
    dag=dag
)

# 在庫データ抽出タスク
def extract_inventory_data(**context):
    """在庫データをAPIから取得する"""
    print("在庫データを抽出中...")
    return "inventory_data_extracted"

extract_inventory = PythonOperator(
    task_id='extract_inventory',
    python_callable=extract_inventory_data,
    dag=dag
)

注文データと在庫データの抽出タスクも同様に定義します。これら 3 つのタスクは並列実行が可能です。

データ検証タスク

抽出したデータの品質を検証します。

pythonCopy code# データ品質検証関数
def validate_data(data_type, **context):
    """データの品質を検証する"""
    checks = {
        'schema_check': True,    # スキーマ検証
        'null_check': True,      # NULL値チェック
        'range_check': True,     # 値範囲チェック
        'duplicate_check': True  # 重複チェック
    }

    print(f"{data_type}データの品質検証中...")

    for check_name, result in checks.items():
        if not result:
            raise ValueError(f"{check_name} failed for {data_type}")

    print(f"{data_type}データの検証完了")
    return f"{data_type}_validated"

この関数では、4 種類の品質チェックを実施しています。いずれかのチェックが失敗すると、エラーを発生させます。

pythonCopy code# 各データソースの検証タスク
validate_customers = PythonOperator(
    task_id='validate_customers',
    python_callable=validate_data,
    op_kwargs={'data_type': 'customer'},
    dag=dag
)

validate_orders = PythonOperator(
    task_id='validate_orders',
    python_callable=validate_data,
    op_kwargs={'data_type': 'order'},
    dag=dag
)

validate_inventory = PythonOperator(
    task_id='validate_inventory',
    python_callable=validate_data,
    op_kwargs={'data_type': 'inventory'},
    dag=dag
)

各データソースに対して、検証タスクを作成します。

データ変換・統合タスク

検証済みのデータを変換し、統合します。

pythonCopy code# データ統合処理
def merge_data(**context):
    """各データソースのデータを統合する"""
    print("データ統合処理を開始...")

    # 顧客、注文、在庫データを結合
    # JOIN処理やデータマッピングを実行

    print("データ統合完了")
    return "data_merged"

merge_task = PythonOperator(
    task_id='merge_data',
    python_callable=merge_data,
    dag=dag
)

3 つのデータソースを統合する処理です。実際には、顧客 ID や商品 ID をキーにして結合処理を行います。

pythonCopy code# 集計処理
def aggregate_data(**context):
    """統合データから集計値を算出する"""
    print("集計処理を開始...")

    # 日次売上集計
    # 顧客別購入金額集計
    # 商品別販売数集計

    print("集計処理完了")
    return "data_aggregated"

aggregate_task = PythonOperator(
    task_id='aggregate_data',
    python_callable=aggregate_data,
    dag=dag
)

統合したデータから、各種集計値を計算します。

データウェアハウスへの格納

処理済みデータをデータウェアハウスに格納します。

pythonCopy code# DWH格納処理
def load_to_dwh(**context):
    """データウェアハウスにデータを格納する"""
    print("DWHへのデータ格納を開始...")

    # BigQueryやSnowflakeなどへ格納
    # トランザクション制御を実施

    print("DWH格納完了")
    return "data_loaded"

load_task = PythonOperator(
    task_id='load_to_dwh',
    python_callable=load_to_dwh,
    dag=dag
)

このタスクでは、変換・集計済みのデータを最終的なデータウェアハウスに格納します。

タスク依存関係の定義

各タスクの実行順序を定義します。

pythonCopy code# タスク依存関係の設定
# 抽出 → 検証 → 統合 → 集計 → 格納 の順で実行

# 抽出後に検証を実行
extract_customers >> validate_customers
extract_orders >> validate_orders
extract_inventory >> validate_inventory

# 全ての検証が完了したら統合を実行
[validate_customers, validate_orders, validate_inventory] >> merge_task

# 統合後に集計を実行
merge_task >> aggregate_task

# 集計後にDWHへ格納
aggregate_task >> load_task

この依存関係により、並列実行可能な部分（抽出と検証）と、順次実行が必要な部分（統合、集計、格納）が制御されます。

パイプライン全体の図解

実装したパイプラインを Mermaid で図解すると、以下のようになります。

mermaidCopy codeflowchart TB
  start["DAG開始<br/>毎日03:00"]

  subgraph extract_phase["抽出フェーズ（並列実行）"]
    extract1["extract_customers<br/>顧客データ抽出"]
    extract2["extract_orders<br/>注文データ抽出"]
    extract3["extract_inventory<br/>在庫データ抽出"]
  end

  subgraph validate_phase["検証フェーズ（並列実行）"]
    validate1["validate_customers<br/>顧客データ検証"]
    validate2["validate_orders<br/>注文データ検証"]
    validate3["validate_inventory<br/>在庫データ検証"]
  end

  merge["merge_data<br/>データ統合"]
  aggregate["aggregate_data<br/>集計処理"]
  load_dwh["load_to_dwh<br/>DWH格納"]

  finish["DAG完了"]

  start --> extract1
  start --> extract2
  start --> extract3

  extract1 --> validate1
  extract2 --> validate2
  extract3 --> validate3

  validate1 --> merge
  validate2 --> merge
  validate3 --> merge

  merge --> aggregate
  aggregate --> load_dwh
  load_dwh --> finish

markdownCopy code```mermaid
flowchart TB
  start["DAG開始<br/>毎日03:00"]

  subgraph extract_phase["抽出フェーズ（並列実行）"]
    extract1["extract_customers<br/>顧客データ抽出"]
    extract2["extract_orders<br/>注文データ抽出"]
    extract3["extract_inventory<br/>在庫データ抽出"]
  end

  subgraph validate_phase["検証フェーズ（並列実行）"]
    validate1["validate_customers<br/>顧客データ検証"]
    validate2["validate_orders<br/>注文データ検証"]
    validate3["validate_inventory<br/>在庫データ検証"]
  end

  merge["merge_data<br/>データ統合"]
  aggregate["aggregate_data<br/>集計処理"]
  load_dwh["load_to_dwh<br/>DWH格納"]

  finish["DAG完了"]

  start --> extract1
  start --> extract2
  start --> extract3

  extract1 --> validate1
  extract2 --> validate2
  extract3 --> validate3

  validate1 --> merge
  validate2 --> merge
  validate3 --> merge

  merge --> aggregate
  aggregate --> load_dwh
  load_dwh --> finish
```

このフローチャートにより、実装したコードの全体像が視覚的に理解できますね。並列実行される抽出・検証フェーズと、順次実行される統合・集計・格納フェーズが明確に区別されています。

図で理解できる要点：

• 並列実行可能なタスクにより処理時間を短縮
• 各フェーズでのエラーハンドリングポイントを特定可能
• 全体の処理フローを一目で把握できる

モニタリングとアラート設定

データパイプラインの監視体制を構築します。

typescriptCopy code// アラート設定の型定義
interface AlertConfig {
  taskId: string; // 監視対象タスクID
  alertChannel: string; // 通知先チャネル
  conditions: {
    onFailure: boolean; // 失敗時の通知
    onRetry: boolean; // リトライ時の通知
    onSuccess: boolean; // 成功時の通知
  };
  sla: number; // SLA（秒）
}

アラート設定の型定義です。各タスクに対して、通知条件と SLA を設定できます。

typescriptCopy code// モニタリング設定
const monitoringConfig: AlertConfig[] = [
  {
    taskId: 'extract_customers',
    alertChannel: '#data-alerts',
    conditions: {
      onFailure: true,
      onRetry: true,
      onSuccess: false,
    },
    sla: 600, // 10分以内に完了すべき
  },
  {
    taskId: 'load_to_dwh',
    alertChannel: '#data-alerts',
    conditions: {
      onFailure: true,
      onRetry: true,
      onSuccess: true, // 成功時も通知
    },
    sla: 1800, // 30分以内に完了すべき
  },
];

重要なタスクに対して、詳細なモニタリング設定を行います。特に DWH への格納は成功時も通知することで、データの更新を確認できますね。

モニタリングのフローを図解すると、以下のようになります：

mermaidCopy codeflowchart LR
  task_exec["タスク実行"]
  monitor["モニタリング<br/>システム"]

  check_status["ステータス<br/>チェック"]
  check_sla["SLA<br/>チェック"]

  normal["正常"]
  alert_fail["失敗アラート"]
  alert_sla["SLAアラート"]

  slack["Slack通知"]
  email["Email通知"]
  pagerduty["PagerDuty"]

  task_exec --> monitor
  monitor --> check_status
  monitor --> check_sla

  check_status -->|成功| normal
  check_status -->|失敗| alert_fail
  check_sla -->|超過| alert_sla

  alert_fail --> slack
  alert_fail --> email
  alert_fail --> pagerduty

  alert_sla --> slack
  alert_sla --> email

markdownCopy code```mermaid
flowchart LR
  task_exec["タスク実行"]
  monitor["モニタリング<br/>システム"]

  check_status["ステータス<br/>チェック"]
  check_sla["SLA<br/>チェック"]

  normal["正常"]
  alert_fail["失敗アラート"]
  alert_sla["SLAアラート"]

  slack["Slack通知"]
  email["Email通知"]
  pagerduty["PagerDuty"]

  task_exec --> monitor
  monitor --> check_status
  monitor --> check_sla

  check_status -->|成功| normal
  check_status -->|失敗| alert_fail
  check_sla -->|超過| alert_sla

  alert_fail --> slack
  alert_fail --> email
  alert_fail --> pagerduty

  alert_sla --> slack
  alert_sla --> email
```

この図により、タスクの実行状態と SLA を監視し、適切な通知チャネルにアラートを送信するフローが理解できます。

まとめ

Mermaid を使ったデータ基盤のラインジ図解により、以下のメリットが得られます。

可視化による理解促進

データパイプラインの全体像を図解することで、チームメンバー全員が同じ認識を持てるようになります。新規メンバーのオンボーディングも大幅に短縮できるでしょう。

依存関係の明確化

DAG を使った図解により、タスク間の依存関係が一目で分かります。これにより、並列実行可能な部分を特定し、処理時間を最適化できますね。

品質保証の組み込み

データ品質チェックのポイントを図に含めることで、品質保証の仕組みが明確になります。エラー時のリトライやエスカレーションフローも視覚的に把握できます。

運用保守の効率化

モニタリングとアラートのフローを図解することで、障害時の対応手順が明確になります。どのタスクが失敗した際にどこに通知されるかが一目で分かるのです。

Mermaid による図解は、Markdown ファイルとして管理でき、バージョン管理システムで変更履歴を追跡できます。ドキュメントとコードを一元管理し、常に最新の状態を保てるでしょう。

データ基盤の構築・運用において、Mermaid を活用した図解は強力なツールとなります。ぜひ実際のプロジェクトで活用してみてください。

関連リンク

• Mermaid 公式ドキュメント
• Apache Airflow 公式ドキュメント
• Great Expectations 公式サイト
• dbt 公式ドキュメント
• Google Cloud BigQuery ドキュメント
• Snowflake ドキュメント