第01课：数据工程概述

🌐 数据工程概述

数据工程是构建数据基础设施的工程学科，负责数据的采集、存储、处理和交付，使数据能够在组织中高效流动并被有效利用。

🐍 Python实战：核心实战

import time, json, hashlib, random
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass

@dataclass
class DataSource:
    name: str
    source_type: str
    schema: dict
    record_count: int = 0
    
    def generate_record(self):
        record = {}
        for field_name, field_type in self.schema.items():
            if field_type == 'int': record[field_name] = random.randint(1, 1000)
            elif field_type == 'float': record[field_name] = round(random.uniform(0, 100), 2)
            elif field_type == 'str': record[field_name] = f"{field_name}_{random.randint(1000,9999)}"
            elif field_type == 'timestamp': record[field_name] = datetime.now().isoformat()
        self.record_count += 1
        return record

class DataIngestion:
    def __init__(self):
        self.buffer = []
        self.cdc_log = []
        self.stats = defaultdict(lambda: {'batch': 0, 'cdc': 0})
    
    def batch_ingest(self, source, count):
        records = [source.generate_record() for _ in range(count)]
        self.buffer.extend(records)
        self.stats[source.name]['batch'] += count
        return records
    
    def cdc_ingest(self, source):
        changes = []
        for _ in range(random.randint(1, 5)):
            record = source.generate_record()
            change = {'operation': random.choice(['INSERT','UPDATE','DELETE']),
                'source': source.name, 'after': record, 'timestamp': datetime.now().isoformat()}
            changes.append(change)
            self.cdc_log.append(change)
        self.stats[source.name]['cdc'] += len(changes)
        return changes

class DataTransformer:
    def __init__(self): self.transform_log = []
    def clean(self, records, rules):
        cleaned, removed = [], 0
        for r in records:
            skip = False
            for field, rule in rules.items():
                if field in r:
                    val = r[field]
                    if rule.get('not_null') and val is None: skip = True; removed += 1; break
                    if rule.get('min') is not None and isinstance(val,(int,float)) and val < rule['min']: skip = True; removed += 1; break
            if not skip: cleaned.append(r)
        self.transform_log.append(f"清洗: {len(records)}→{len(cleaned)} (移除{removed})")
        return cleaned
    def normalize(self, records, field, method='minmax'):
        values = [r[field] for r in records if field in r and isinstance(r[field],(int,float))]
        if not values: return records
        min_v, max_v = min(values), max(values)
        result = []
        for r in records:
            r = r.copy()
            if field in r and isinstance(r[field],(int,float)) and max_v > min_v:
                r[f'{field}_norm'] = round((r[field]-min_v)/(max_v-min_v), 4)
            result.append(r)
        return result
    def aggregate(self, records, group_by, agg_field, agg_fn='sum'):
        groups = defaultdict(list)
        for r in records:
            if group_by in r and agg_field in r: groups[r[group_by]].append(r[agg_field])
        result = {}
        for key, vals in groups.items():
            if agg_fn == 'sum': result[key] = sum(vals)
            elif agg_fn == 'avg': result[key] = round(sum(vals)/len(vals),2)
            elif agg_fn == 'count': result[key] = len(vals)
        return result

class DataLake:
    def __init__(self, name):
        self.name = name
        self.zones = {'raw':[],'cleaned':[],'curated':[]}
    def write(self, zone, records, table_name):
        entry = {'table':table_name,'records':records,'count':len(records),
            'created_at':datetime.now().isoformat(),
            'checksum':hashlib.md5(json.dumps(records,default=str).encode()).hexdigest()[:8]}
        self.zones[zone].append(entry)
        return entry
    def stats(self):
        return {zone:{'tables':len(e),'total_records':sum(x['count'] for x in e)}
            for zone,e in self.zones.items()}

print("=" * 60)
print("数据工程生态系统模拟")
print("=" * 60)
sources = {
    'orders': DataSource('orders','database',{'order_id':'int','amount':'float','customer':'str','ts':'timestamp'}),
    'clicks': DataSource('clicks','log',{'session_id':'str','page':'str','duration':'float','ts':'timestamp'}),
}
print("数据源:", list(sources.keys()))
ingestion = DataIngestion()
batch = ingestion.batch_ingest(sources['orders'], 100)
print(f"批量采集: {len(batch)}条")
cdc = ingestion.cdc_ingest(sources['orders'])
print(f"CDC采集: {len(cdc)}事件")
transformer = DataTransformer()
cleaned = transformer.clean(batch, {'amount':{'not_null':True,'min':0}})
normalized = transformer.normalize(cleaned, 'amount')
agg = transformer.aggregate(cleaned, 'customer', 'amount', 'sum')
print(f"清洗: {len(batch)}→{len(cleaned)}, 聚合: {len(agg)}组")
lake = DataLake("demo")
lake.write('raw', batch, 'raw_orders')
lake.write('cleaned', cleaned, 'clean_orders')
print(f"数据湖: {lake.stats()}")
print("✅ 验证通过 - 生态系统模型运行正常")

📊 数据工程师核心技能矩阵

🐍 性能分析：数据量级与处理策略

import time, random
from collections import defaultdict

def benchmark(sizes, fn):
    results = {}
    for size in sizes:
        data = list(range(size)); random.shuffle(data)
        start = time.perf_counter()
        result = fn(data)
        elapsed = time.perf_counter() - start
        results[size] = {'ms': round(elapsed*1000,2), 'tps': round(size/elapsed,0)}
    return results

sizes = [1000, 5000, 10000, 50000, 100000]
print("排序性能:")
for size, m in benchmark(sizes, sorted).items():
    print(f"  {size:>8,}条: {m['ms']:>8.2f}ms, {m['tps']:>10,.0f}条/秒")

def chunked_sort(data, chunk_size=10000):
    chunks = [sorted(data[i:i+chunk_size]) for i in range(0,len(data),chunk_size)]
    result = []
    for c in chunks: result.extend(c)
    return result

print("分块排序:")
for size, m in benchmark(sizes, chunked_sort).items():
    print(f"  {size:>8,}条: {m['ms']:>8.2f}ms, {m['tps']:>10,.0f}条/秒")

print("✅ 验证通过 - 性能分析完成")

🏛️ 数据工程发展简史

📖 原理深入：ETL设计核心原则

🏆 最佳实践：生产环境ETL检查清单