data-source-audit

建設業向けデータソース監査

概要

建設業のデータソースを包括的に監査し、サイロを特定してデータフローをマッピングし、品質を評価して、統合戦略を計画します。デジタルトランスフォーメーションとデータドリブンな建設の取り組みに不可欠です。

ビジネス上の背景

建設関連の組織では通常、10～50以上のデータソースを利用しています。

• プロジェクト管理システム
• 見積もりソフトウェア
• スケジューリングツール
• 会計／ERPシステム
• BIMプラットフォーム
• ドキュメント管理システム
• 現場向けアプリ
• スプレッドシート

注： このスキルは特定のベンダーに依存せず、あらゆるデータソースで利用できます。他の箇所の例で言及されている製品名は、それぞれの所有者の商標です。

このスキルは、次のことに役立ちます。

• すべてのデータソースを発見する
• データフローと依存関係をマッピングする
• 統合の機会を特定する
• データ改善の取り組みに優先順位を付ける

技術的な実装

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Set
from enum import Enum
from datetime import datetime
import pandas as pd
import json

class DataSourceType(Enum):
    DATABASE = "database"
    API = "api"
    FILE_SHARE = "file_share"
    CLOUD_APP = "cloud_app"
    SPREADSHEET = "spreadsheet"
    LEGACY_SYSTEM = "legacy_system"
    IOT_SENSOR = "iot_sensor"
    MANUAL_ENTRY = "manual_entry"

class DataDomain(Enum):
    COST = "cost"
    SCHEDULE = "schedule"
    BIM = "bim"
    DOCUMENT = "document"
    FIELD = "field"
    SAFETY = "safety"
    QUALITY = "quality"
    HR = "hr"
    ACCOUNTING = "accounting"
    PROCUREMENT = "procurement"

@dataclass
class DataSource:
    name: str
    source_type: DataSourceType
    domains: List[DataDomain]
    owner: str
    department: str
    description: str
    # Technical details
    technology: str
    location: str  # cloud, on-prem, hybrid
    access_method: str  # API, ODBC, file export, manual
    # Data characteristics
    update_frequency: str  # real-time, daily, weekly, monthly, ad-hoc
    data_volume: str  # small, medium, large
    retention_period: str
    # Quality metrics
    completeness_score: float = 0.0
    accuracy_score: float = 0.0
    timeliness_score: float = 0.0
    # Integration status
    integrations: List[str] = field(default_factory=list)
    is_master: bool = False  # Is this the master source for any entity?
    master_for: List[str] = field(default_factory=list)
    # Issues
    known_issues: List[str] = field(default_factory=list)
    # Metadata
    last_audit_date: Optional[datetime] = None
    audit_notes: str = ""

@dataclass
class DataFlow:
    source: str
    target: str
    flow_type: str  # push, pull, bidirectional, manual
    frequency: str
    entities: List[str]  # What data entities flow
    transformation: str  # none, simple, complex
    status: str  # active, planned, deprecated

@dataclass
class DataSilo:
    name: str
    sources: List[str]
    impact: str  # high, medium, low
    description: str
    resolution_options: List[str]

class DataSourceAuditor:
    """Audit and analyze construction data sources."""

    def __init__(self):
        self.sources: Dict[str, DataSource] = {}
        self.flows: List[DataFlow] = []
        self.silos: List[DataSilo] = []

    def add_source(self, source: DataSource):
        """Register a data source."""
        self.sources[source.name] = source

    def add_flow(self, flow: DataFlow):
        """Register a data flow between sources."""
        self.flows.append(flow)

    def discover_sources_from_survey(self, survey_responses: List[Dict]) -> List[DataSource]:
        """Create data sources from survey responses."""
        sources = []

        for response in survey_responses:
            source = DataSource(
                name=response['system_name'],
                source_type=DataSourceType(response['type']),
                domains=[DataDomain(d) for d in response['domains']],
                owner=response['owner'],
                department=response['department'],
                description=response['description'],
                technology=response['technology'],
                location=response['location'],
                access_method=response['access_method'],
                update_frequency=response['update_frequency'],
                data_volume=response['data_volume'],
                retention_period=response['retention_period'],
            )
            sources.append(source)
            self.add_source(source)

        return sources

    def identify_silos(self) -> List[DataSilo]:
        """Identify data silos based on integration analysis."""
        silos = []

        # Find sources with no integrations
        isolated_sources = [
            name for name, source in self.sources.items()
            if not source.integrations and source.source_type != DataSourceType.MANUAL_ENTRY
        ]

        if isolated_sources:
            silos.append(DataSilo(
                name="Isolated Systems",
                sources=isolated_sources,
                impact="high",
                description="Systems with no integrations, requiring manual data transfer",
                resolution_options=[
                    "Implement API integration",
                    "Set up automated file exports",
                    "Migrate to integrated platform"
                ]
            ))

        # Find duplicate data domains without master
        domain_sources: Dict[DataDomain, List[str]] = {}
        for name, source in self.sources.items():
            for domain in source.domains:
                if domain not in domain_sources:
                    domain_sources[domain] = []
                domain_sources[domain].append(name)

        for domain, sources in domain_sources.items():
            if len(sources) > 1:
                # Check if any is designated master
                masters = [s for s in sources if self.sources[s].is_master]
                if not masters:
                    silos.append(DataSilo(
                        name=f"No Master for {domain.value}",
                        sources=sources,
                        impact="medium",
                        description=f"Multiple sources for {domain.value} data without designated master",
                        resolution_options=[
                            "Designate master data source",
                            "Implement MDM solution",
                            "Create data reconciliation process"
                        ]
                    ))

        # Find one-way flows that should be bidirectional
        flow_pairs = {}
        for flow in self.flows:
            key = tuple(sorted([flow.source, flow.target]))
            if key not in flow_pairs:
                flow_pairs[key] = []
            flow_pairs[key].append(flow)

        for (s1, s2), flows in flow_pairs.items():
            if len(flows) == 1 and flows[0].flow_type != 'bidirectional':
                # Check if bidirectional would make sense
                s1_domains = set(self.sources[s1].domains)
                s2_domains = set(self.sources[s2].domains)
                if s1_domains & s2_domains:  # Overlapping domains
                    silos.append(DataSilo(
                        name=f"One-way flow: {s1} -> {s2}",
                        sources=[s1, s2],
                        impact="low",
                        description="Data flows one direction only between systems with overlapping domains",
                        resolution_options=[
                            "Evaluate need for bidirectional sync",
                            "Implement change data capture"
                        ]
                    ))

        self.silos = silos
        return silos

    def assess_source_quality(self, source_name: str, sample_data: pd.DataFrame) -> Dict[str, float]:
        """Assess data quality for a source based on sample data."""
        if source_name not in self.sources:
            raise ValueError(f"Unknown source: {source_name}")

        scores = {}

        # Completeness: % of non-null values
        completeness = 1 - (sample_data.isnull().sum().sum() / sample_data.size)
        scores['completeness'] = completeness

        # Uniqueness: % of unique rows (for key columns)
        if len(sample_data) > 0:
            uniqueness = len(sample_data.drop_duplicates()) / len(sample_data)
        else:
            uniqueness = 1.0
        scores['uniqueness'] = uniqueness

        # Validity: Basic format checks (simplified)
        validity_checks = 0
        total_checks = 0

        for col in sample_data.columns:
            if 'date' in col.lower():
                total_checks += 1
                try:
                    pd.to_datetime(sample_data[col], errors='raise')
                    validity_checks += 1
                except:
                    pass
            if 'email' in col.lower():
                total_checks += 1
                valid_emails = sample_data[col].str.contains(r'@.*\.', na=False).sum()
                if valid_emails / len(sample_data) > 0.9:
                    validity_checks += 1

        scores['validity'] = validity_checks / total_checks if total_checks > 0 else 1.0

        # Update source with scores
        self.sources[source_name].completeness_score = scores['completeness']
        self.sources[source_name].accuracy_score = scores['validity']

        return scores

    def create_data_catalog(self) -> pd.DataFrame:
        """Create a data catalog from all sources."""
        catalog_entries = []

        for name, source in self.sources.items():
            entry = {
                'Source Name': name,
                'Type': source.source_type.value,
                'Domains': ', '.join(d.value for d in source.domains),
                'Owner': source.owner,
                'Department': source.department,
                'Technology': source.technology,
                'Location': source.location,
                'Access Method': source.access_method,
                'Update Frequency': source.update_frequency,
                'Data Volume': source.data_volume,
                'Integrations': len(source.integrations),
                'Is Master': 'Yes' if source.is_master else 'No',
                'Quality Score': (source.completeness_score + source.accuracy_score) / 2,
                'Known Issues': len(source.known_issues),
            }
            catalog_entries.append(entry)

        return pd.DataFrame(catalog_entries)

    def generate_integration_matrix(self) -> pd.DataFrame:
        """Generate integration matrix showing connections between sources."""
        source_names = list(self.sources.keys())
        matrix = pd.DataFrame(
            index=source_names,
            columns=source_names,
            data=''
        )

        for flow in self.flows:
            if flow.source in source_names and flow.target in source_names:
                current = matrix.loc[flow.source, flow.target]
                symbol = '→' if flow.flow_type == 'push' else '←' if flow.flow_type == 'pull' else '↔'
                matrix.loc[flow.source, flow.target] = f"{current}{symbol}" if current else symbol

        return matrix

    def calculate_integration_score(self) -> Dict[str, float]:
        """Calculate overall integration score and breakdown."""
        if not self.sources:
            return {'overall': 0.0}

        scores = {}

        # Coverage: % of sources with at least one integration
        integrated = sum(1 for s in self.sources.values() if s.integrations)
        scores['coverage'] = integrated / len(self.sources)

        # Master data: % of domains with designated master
        domains_with_master = set()
        for source in self.sources.values():
            if source.is_master:
                domains_with_master.update(source.master_for)

        all_domains = set()
        for source in self.sources.values():
            all_domains.update(d.value for d in source.domains)

        scores['master_data'] = len(domains_with_master) / len(all_domains) if all_domains else 1.0

        # Data quality average
        quality_scores = [
            (s.completeness_score + s.accuracy_score) / 2
            for s in self.sources.values()
            if s.completeness_score > 0 or s.accuracy_score > 0
        ]
        scores['quality'] = sum(quality_scores) / len(quality_scores) if quality_scores else 0.0

        # Silo impact
        high_impact_silos = sum(1 for s in self.silos if s.impact == 'high')
        scores['silo_risk'] = 1 - (high_impact_silos * 0.2)  # Each high-impact silo reduces score

        # Overall
        scores['overall'] = (
            scores['coverage'] * 0.3 +
            scores['master_data'] * 0.25 +
            scores['quality'] * 0.25 +
            scores['silo_risk'] * 0.2
        )

        return scores

    def generate_audit_report(self) -> str:
        """Generate comprehensive audit report."""
        report = ["# Data Source Audit Report", ""]
        report.append(f"**Audit Date:** {datetime.now().strftime('%Y-%m-%d')}")
        report.append(f"**Total Sources:** {len(self.sources)}")
        report.append(f"**Total Data Flows:** {len(self.flows)}")
        report.append("")

        # Integration Score
        scores = self.calculate_integration_score()
        report.append("## Integration Maturity Score")
        report.append(f"**Overall Score:** {scores['overall']:.1%}")
        report.append(f"- Coverage: {scores['coverage']:.1%}")
        report.append(f"- Master Data: {scores['master_data']:.1%}")
        report.append(f"- Data Quality: {scores['quality']:.1%}")
        report.append(f"- Silo Risk: {scores['silo_risk']:.1%}")
        report.append("")

        # Sources by Type
        report.append("## Sources by Type")
        by_type = {}
        for source in self.sources.values():
            t = source.source_type.value
            by_type[t] = by_type.get(t, 0) + 1
        for t, count in sorted(by_type.items(), key=lambda x: -x[1]):
            report.append(f"- {t}: {count}")
        report.append("")

        # Data Silos
        report.append("## Identified Data Silos")
        if self.silos:
            for silo in self.silos:
                report.append(f"\n### {silo.name}")
                report.append(f"**Impact:** {silo.impact}")
                report.append(f"**Sources:** {', '.join(silo.sources)}")
                report.append(f"**Description:** {silo.description}")
                report.append("**Resolution Options:**")
                for opt in silo.resolution_options:
                    report.append(f"- {opt}")
        else:
            report.append("No significant data silos identified.")
        report.append("")

        # Recommendations
        report.append("## Recommendations")
        recommendations = self._generate_recommendations()
        for i, rec in enumerate(recommendations, 1):
            report.append(f"{i}. {rec}")

        return "\n".join(report)

    def _generate_recommendations(self) -> List[str]:
        """Generate recommendations based on audit findings."""
        recommendations = []

        scores = self.calculate_integration_score()

        if scores['coverage'] < 0.7:
            recommendations.append(
                "Increase integration coverage - over 30% of systems are isolated. "
                "Prioritize connecting high-value data sources."
            )

        if scores['master_data'] < 0.5:
            recommendations.append(
                "Implement Master Data Management - designate authoritative sources "
                "for key entities (projects, vendors, employees, cost codes)."
            )

        if scores['quality'] < 0.7:
            recommendations.append(
                "Improve data quality - implement validation rules at data entry points "
                "and automated quality monitoring."
            )

        # Check for spreadsheet dependency
        spreadsheets = [s for s in self.sources.values()
                       if s.source_type == DataSourceType.SPREADSHEET]
        if len(spreadsheets) > 3:
            recommendations.append(
                f"Reduce spreadsheet dependency - {len(spreadsheets)} spreadsheet-based "
                "data sources identified. Migrate critical data to proper databases."
            )

        # Check for legacy systems
        legacy = [s for s in self.sources.values()
                 if s.source_type == DataSourceType.LEGACY_SYSTEM]
        if legacy:
            recommendations.append(
                f"Plan legacy system migration - {len(legacy)} legacy systems identified. "
                "Create modernization roadmap."
            )

        return recommendations

クイックスタート

# Initialize auditor
auditor = DataSourceAuditor()

# Add known sources
auditor.add_source(DataSource(
    name="Procore",
    source_type=DataSourceType.CLOUD_APP,
    domains=[DataDomain.DOCUMENT, DataDomain.FIELD, DataDomain.SCHEDULE],
    owner="Project Controls",
    department="Operations",
    description="Primary project management platform",
    technology="SaaS",
    location="cloud",
    access_method="API",
    update_frequency="real-time",
    data_volume="large",
    retention_period="7 years",
    integrations=["Sage 300", "Primavera P6"],
    is_master=True,
    master_for=["projects", "documents"]
))

auditor.add_source(DataSource(
    name="Sage 300",
    source_type=DataSourceType.DATABASE,
    domains=[DataDomain.COST, DataDomain.ACCOUNTING],
    owner="Finance",
    department="Accounting",
    description="ERP and job costing system",
    technology="SQL Server",
    location="on-prem",
    access_method="ODBC",
    update_frequency="daily",
    data_volume="medium",
    retention_period="10 years",
    is_master=True,
    master_for=["costs", "vendors", "invoices"]
))

# Add data flows
auditor.add_flow(DataFlow(
    source="Procore",
    target="Sage 300",
    flow_type="push",
    frequency="daily",
    entities=["change_orders", "budget_changes"],
    transformation="simple",
    status="active"
))

# Identify silos
silos = auditor.identify_silos()

# Generate report
report = auditor.generate_audit_report()
print(report)

# Create data catalog
catalog = auditor.create_data_catalog()
catalog.to_excel("data_catalog.xlsx", index=False)

調査テンプレート

この調査を使用して、組織全体のデータソースを把握します。

System Survey:
  - system_name: "What is the name of this system?"
  - type: "What type of system is it?"
    options: [database, api, file_share, cloud_app, spreadsheet, legacy_system]
  - domains: "What types of data does it contain?"
    options: [cost, schedule, bim, document, field, safety, quality, hr, accounting]
  - owner: "Who is the system owner?"
  - department: "Which department uses this system?"
  - technology: "What technology/platform is it built on?"
  - location: "Where is the system hosted?"
    options: [cloud, on-prem, hybrid]
  - access_method: "How can data be accessed?"
    options: [api, odbc, file_export, manual]
  - update_frequency: "How often is data updated?"
    options: [real-time, daily, weekly, monthly, ad-hoc]
  - integrations: "What other systems does it connect to?"

リソース

• DAMA DMBOK：データマネジメント知識体系
• データガバナンスフレームワーク：DCAM、EDM Council
• 統合パターン：書籍『Enterprise Integration Patterns』