第11章 数据治理¶

# 数据目录实现示例（以 DataHub 为后端）
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.metadata.schema_classes import (
    DatasetPropertiesClass,
    OwnershipClass,
    OwnerClass,
    OwnershipTypeClass,
)

emitter = DatahubRestEmitter(gms_server="http://localhost:8080")

# 注册数据资产：定义表的属性
dataset_urn = "urn:li:dataset:(urn:li:dataPlatform:hive,sales_db.sales,PROD)"

# 发送数据集属性
emitter.emit(MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=DatasetPropertiesClass(
        name="sales",
        description="Daily sales data",
        customProperties={
            "team": "data_engineering",
            "update_frequency": "daily"
        }
    )
))

# 设置数据所有者
emitter.emit(MetadataChangeProposalWrapper(
    entityUrn=dataset_urn,
    aspect=OwnershipClass(
        owners=[
            OwnerClass(
                owner="urn:li:corpuser:data_team",
                type=OwnershipTypeClass.DATAOWNER
            )
        ]
    )
))

print(f"Dataset registered: {dataset_urn}")

class DataStandards:
    """数据标准类"""

    def __init__(self):
        self.standards = {}

    def define_naming_convention(self, convention_type):
        """定义命名规范"""
        if convention_type == 'table':
            self.standards['table'] = {
                'prefix': 'dwd_',
                'separator': '_',
                'case': 'lower',
                'max_length': 50
            }
        elif convention_type == 'column':
            self.standards['column'] = {
                'snake_case': True,
                'no_abbreviations': True,
                'max_length': 30
            }

    def validate_name(self, name, convention_type):
        """验证命名"""
        standards = self.standards.get(convention_type)

        if not standards:
            return True

        # 检查长度
        if len(name) > standards['max_length']:
            return False

        # 检查大小写
        if standards.get('case') == 'lower':
            if name != name.lower():
                return False

        return True

    def validate_data_type(self, value, expected_type):
        """验证数据类型"""
        if expected_type == 'integer':
            try:  # try/except捕获异常
                int(value)
                return True
            except:
                return False
        elif expected_type == 'float':
            try:
                float(value)
                return True
            except:
                return False
        elif expected_type == 'date':
            try:
                pd.to_datetime(value)
                return True
            except:
                return False

        return True

class ComplianceManager:
    """合规管理类"""

    def __init__(self):
        self.regulations = {}

    def define_regulation(self, regulation_name, requirements):
        """定义合规要求"""
        self.regulations[regulation_name] = {
            'requirements': requirements,
            'status': 'active'
        }

    def check_compliance(self, data, regulation_name):
        """检查合规性"""
        regulation = self.regulations.get(regulation_name)

        if not regulation:
            return {'compliant': True, 'issues': []}

        issues = []

        for requirement in regulation['requirements']:
            if not self._check_requirement(data, requirement):
                issues.append(requirement)

        return {
            'compliant': len(issues) == 0,
            'issues': issues
        }

    def _check_requirement(self, data, requirement):
        """检查单个要求"""
        if requirement['type'] == 'privacy':
            # 检查是否包含敏感信息
            sensitive_fields = ['phone', 'email', 'id_card']
            for field in sensitive_fields:
                if field in data and data[field]:
                    return False
            return True
        elif requirement['type'] == 'security':
            # 检查是否加密
            if 'password' in data:
                return not data['password'].startswith('***')
            return True
        elif requirement['type'] == 'access':
            # 检查访问控制
            if 'access_level' in data:
                return data['access_level'] in ['read', 'write']
            return True

        return True