etl-pipeline

建設データの ETL パイプライン

概要

DDC の方法論（第 4.2 章）に基づき、このスキルでは、さまざまなソースから情報を抽出し、有用な形式に変換して、ターゲットシステムに読み込んだり、レポートを生成したりする自動データパイプラインを構築できます。

書籍の参照先：「ETL и автоматизация процессов」／「ETL and Process Automation」

「ETL：手動管理から自動化へ移行することで、企業は人間が常時介入しなくてもデータを処理できるようになります。」 — DDC Book、第 4.2 章

ETL の構成要素

┌─────────┐    ┌───────────┐    ┌────────┐
│ EXTRACT │ -> │ TRANSFORM │ -> │  LOAD  │
└─────────┘    └───────────┘    └────────┘
   │               │               │
   ▼               ▼               ▼
 Sources        Process         Outputs
 - PDF          - Clean         - Excel
 - Excel        - Validate      - PDF
 - CSV          - Calculate     - Database
 - BIM          - Merge         - API
 - API          - Aggregate     - Dashboard

クイックスタート

import pandas as pd

# Simple ETL Pipeline
def simple_etl_pipeline(input_file, output_file):
    # EXTRACT
    df = pd.read_excel(input_file)

    # TRANSFORM
    df = df.dropna()  # Clean
    df['Total'] = df['Quantity'] * df['Unit_Price']  # Calculate
    summary = df.groupby('Category')['Total'].sum()  # Aggregate

    # LOAD
    summary.to_excel(output_file)
    return summary

# Run
result = simple_etl_pipeline("raw_data.xlsx", "processed_report.xlsx")

抽出：データソース

複数の Excel ファイルから

import pandas as pd
from pathlib import Path

def extract_excel_files(folder_path, pattern="*.xlsx"):
    """Extract data from multiple Excel files"""
    files = Path(folder_path).glob(pattern)
    all_data = []

    for file in files:
        try:
            df = pd.read_excel(file)
            df['_source_file'] = file.name
            all_data.append(df)
            print(f"Extracted: {file.name}")
        except Exception as e:
            print(f"Error reading {file.name}: {e}")

    if all_data:
        return pd.concat(all_data, ignore_index=True)
    return pd.DataFrame()

# Usage
df = extract_excel_files("./project_data/")

PDF ドキュメントから

import pdfplumber
import pandas as pd

def extract_from_pdfs(pdf_folder):
    """Extract tables from all PDFs in folder"""
    files = Path(pdf_folder).glob("*.pdf")
    all_tables = []

    for pdf_path in files:
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                tables = page.extract_tables()
                for table in tables:
                    if table and len(table) > 1:
                        df = pd.DataFrame(table[1:], columns=table[0])
                        df['_source'] = pdf_path.name
                        all_tables.append(df)

    return pd.concat(all_tables, ignore_index=True) if all_tables else pd.DataFrame()

API から

import requests
import pandas as pd

def extract_from_api(api_url, headers=None):
    """Extract data from REST API"""
    response = requests.get(api_url, headers=headers)

    if response.status_code == 200:
        data = response.json()
        return pd.DataFrame(data)
    else:
        raise Exception(f"API error: {response.status_code}")

# Usage
df = extract_from_api("https://api.example.com/projects")

データベースから

import pandas as pd
import sqlite3

def extract_from_database(db_path, query):
    """Extract data using SQL query"""
    conn = sqlite3.connect(db_path)
    df = pd.read_sql_query(query, conn)
    conn.close()
    return df

# Usage
df = extract_from_database(
    "construction.db",
    "SELECT * FROM elements WHERE category = 'Wall'"
)

変換：データ処理

データのクリーニング

def clean_construction_data(df):
    """Standard cleaning for construction data"""
    # Remove empty rows
    df = df.dropna(how='all')

    # Strip whitespace
    for col in df.select_dtypes(include=['object']).columns:
        df[col] = df[col].str.strip()

    # Standardize category names
    if 'Category' in df.columns:
        df['Category'] = df['Category'].str.title()

    # Convert numeric columns
    numeric_cols = ['Volume', 'Area', 'Length', 'Quantity', 'Cost']
    for col in numeric_cols:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')

    # Remove duplicates
    df = df.drop_duplicates()

    return df

データの検証

def validate_construction_data(df, rules):
    """
    Validate data against rules

    Args:
        rules: list of dicts like
        [{'column': 'Volume', 'rule': 'positive'},
         {'column': 'Category', 'rule': 'not_null'}]
    """
    errors = []

    for rule in rules:
        col = rule['column']
        rule_type = rule['rule']

        if col not in df.columns:
            errors.append(f"Missing column: {col}")
            continue

        if rule_type == 'positive':
            invalid = df[df[col] <= 0]
            if len(invalid) > 0:
                errors.append(f"{len(invalid)} rows with non-positive {col}")

        elif rule_type == 'not_null':
            null_count = df[col].isna().sum()
            if null_count > 0:
                errors.append(f"{null_count} null values in {col}")

        elif rule_type == 'unique':
            duplicates = df[col].duplicated().sum()
            if duplicates > 0:
                errors.append(f"{duplicates} duplicate values in {col}")

    return errors

# Usage
validation_rules = [
    {'column': 'Volume', 'rule': 'positive'},
    {'column': 'Category', 'rule': 'not_null'},
    {'column': 'ElementId', 'rule': 'unique'}
]
errors = validate_construction_data(df, validation_rules)

データの集計

def aggregate_by_hierarchy(df, hierarchy=['Project', 'Building', 'Level', 'Category']):
    """Aggregate data at different hierarchy levels"""
    results = {}

    for i in range(1, len(hierarchy) + 1):
        level_cols = hierarchy[:i]
        if all(col in df.columns for col in level_cols):
            agg = df.groupby(level_cols).agg({
                'Volume': 'sum',
                'Cost': 'sum',
                'ElementId': 'count'
            }).rename(columns={'ElementId': 'Count'})

            level_name = '_'.join(level_cols)
            results[level_name] = agg

    return results

# Usage
aggregations = aggregate_by_hierarchy(df)
for name, data in aggregations.items():
    print(f"\n{name}:")
    print(data.head())

データのエンリッチメント

def enrich_with_prices(df, prices_df):
    """Enrich element data with pricing information"""
    # Merge with price database
    enriched = df.merge(prices_df, on='Category', how='left')

    # Calculate costs
    enriched['Material_Cost'] = enriched['Volume'] * enriched['Unit_Price']
    enriched['Labor_Cost'] = enriched['Volume'] * enriched['Labor_Rate']
    enriched['Total_Cost'] = enriched['Material_Cost'] + enriched['Labor_Cost']

    return enriched

読み込み：出力の生成

Excel レポートの生成

def generate_excel_report(df, summary, output_path):
    """Generate formatted Excel report"""
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Raw data
        df.to_excel(writer, sheet_name='Data', index=False)

        # Summary by category
        summary.to_excel(writer, sheet_name='Summary')

        # Pivot table
        if 'Level' in df.columns and 'Category' in df.columns:
            pivot = pd.pivot_table(
                df, values='Volume',
                index='Level', columns='Category',
                aggfunc='sum', fill_value=0
            )
            pivot.to_excel(writer, sheet_name='By_Level')

    print(f"Report saved: {output_path}")

# Usage
generate_excel_report(df, summary, "project_report.xlsx")

PDF レポートの生成

from reportlab.lib import colors
from reportlab.lib.pagesizes import letter, A4
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph
from reportlab.lib.styles import getSampleStyleSheet

def generate_pdf_report(df, output_path, title="Construction Report"):
    """Generate PDF report from DataFrame"""
    doc = SimpleDocTemplate(output_path, pagesize=A4)
    elements = []
    styles = getSampleStyleSheet()

    # Title
    elements.append(Paragraph(title, styles['Title']))

    # Convert DataFrame to table
    data = [df.columns.tolist()] + df.values.tolist()
    table = Table(data)

    # Style the table
    table.setStyle(TableStyle([
        ('BACKGROUND', (0, 0), (-1, 0), colors.grey),
        ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
        ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
        ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
        ('FONTSIZE', (0, 0), (-1, 0), 10),
        ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
        ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
        ('GRID', (0, 0), (-1, -1), 1, colors.black)
    ]))

    elements.append(table)
    doc.build(elements)
    print(f"PDF saved: {output_path}")

# Usage
generate_pdf_report(summary, "report.pdf")

データベースへの読み込み

import sqlite3

def load_to_database(df, db_path, table_name, if_exists='replace'):
    """Load DataFrame to SQLite database"""
    conn = sqlite3.connect(db_path)
    df.to_sql(table_name, conn, if_exists=if_exists, index=False)
    conn.close()
    print(f"Loaded {len(df)} rows to {table_name}")

# Usage
load_to_database(df, "construction.db", "elements")

ETL パイプライン全体

class ConstructionETLPipeline:
    """Complete ETL pipeline for construction data"""

    def __init__(self, config):
        self.config = config
        self.data = None
        self.errors = []

    def extract(self):
        """Extract data from configured sources"""
        print("Extracting data...")
        sources = []

        # Excel files
        if 'excel_folder' in self.config:
            df = extract_excel_files(self.config['excel_folder'])
            sources.append(df)

        # PDF files
        if 'pdf_folder' in self.config:
            df = extract_from_pdfs(self.config['pdf_folder'])
            sources.append(df)

        self.data = pd.concat(sources, ignore_index=True)
        print(f"Extracted {len(self.data)} records")
        return self

    def transform(self):
        """Apply transformations"""
        print("Transforming data...")

        # Clean
        self.data = clean_construction_data(self.data)

        # Validate
        if 'validation_rules' in self.config:
            self.errors = validate_construction_data(
                self.data, self.config['validation_rules']
            )

        # Enrich with prices if available
        if 'prices_file' in self.config:
            prices = pd.read_excel(self.config['prices_file'])
            self.data = enrich_with_prices(self.data, prices)

        print(f"Transformed {len(self.data)} records")
        return self

    def load(self):
        """Load to configured outputs"""
        print("Loading data...")

        # Excel report
        if 'excel_output' in self.config:
            summary = self.data.groupby('Category').agg({
                'Volume': 'sum', 'Cost': 'sum'
            })
            generate_excel_report(
                self.data, summary, self.config['excel_output']
            )

        # Database
        if 'database' in self.config:
            load_to_database(
                self.data,
                self.config['database'],
                self.config.get('table_name', 'elements')
            )

        print("Pipeline complete!")
        return self

    def run(self):
        """Run complete pipeline"""
        return self.extract().transform().load()

# Usage
config = {
    'excel_folder': './input_data/',
    'prices_file': './prices.xlsx',
    'validation_rules': [
        {'column': 'Volume', 'rule': 'positive'},
        {'column': 'Category', 'rule': 'not_null'}
    ],
    'excel_output': './output/report.xlsx',
    'database': './output/project.db',
    'table_name': 'elements'
}

pipeline = ConstructionETLPipeline(config)
pipeline.run()

Airflow によるスケジュール設定

# airflow_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'construction_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'construction_etl',
    default_args=default_args,
    description='Daily construction data ETL',
    schedule_interval='@daily',
)

def extract_task():
    # Extract logic
    pass

def transform_task():
    # Transform logic
    pass

def load_task():
    # Load logic
    pass

t1 = PythonOperator(task_id='extract', python_callable=extract_task, dag=dag)
t2 = PythonOperator(task_id='transform', python_callable=transform_task, dag=dag)
t3 = PythonOperator(task_id='load', python_callable=load_task, dag=dag)

t1 >> t2 >> t3

クイックリファレンス

段階	タスク	ツール／メソッド
抽出	Excel の読み込み	pd.read_excel()
抽出	CSV の読み込み	pd.read_csv()
抽出	PDF の読み込み	pdfplumber
抽出	API の読み込み	requests.get()
変換	クリーニング	df.dropna(), df.str.strip()
変換	検証	カスタム検証関数
変換	計算	df['new'] = df['a'] * df['b']
変換	集計	df.groupby().agg()
読み込み	Excel	df.to_excel()
読み込み	PDF	reportlab
読み込み	データベース	df.to_sql()
読み込み	API	requests.post()

リソース

• 書籍：Artem Boiko 著「Data-Driven Construction」、第 4.2 章
• ウェブサイト：https://datadrivenconstruction.io
• Airflow：https://airflow.apache.org
• n8n：https://n8n.io

次のステップ

• BIM データの検証については、bim-validation-pipeline を参照してください
• 高度な PDF 生成については、pdf-report-generator を参照してください
• n8n との統合については、workflow-automation を参照してください