Skip to content

Commit 28eb646

Browse files
authored
Create ml_pipeline.py
1 parent 8ba213a commit 28eb646

1 file changed

Lines changed: 257 additions & 0 deletions

File tree

‎projects/advanced/ml_pipeline.py‎

Lines changed: 257 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,257 @@
1+
"""
2+
Advanced Project: Machine Learning Data Pipeline
3+
================================================
4+
5+
This project demonstrates building a production-ready ML data pipeline.
6+
It covers advanced Python concepts including:
7+
- Generator functions and iterators
8+
- Context managers
9+
- Decorators for timing and logging
10+
- Type hints and data validation
11+
- Async/await patterns
12+
- Data transformation pipelines
13+
14+
Features:
15+
- Lazy data loading with generators
16+
- Pipeline pattern for data transformations
17+
- Performance monitoring
18+
- Error handling and recovery
19+
- Batch processing
20+
21+
Requirements:
22+
pip install pandas numpy scikit-learn (optional, for full ML features)
23+
"""
24+
25+
import time
26+
import logging
27+
from datetime import datetime
28+
from typing import List, Dict, Any, Callable, Iterator, Optional, TypeVar, Generic
29+
from functools import wraps
30+
from contextlib import contextmanager
31+
from dataclasses import dataclass, field
32+
import json
33+
34+
# Configure logging
35+
logging.basicConfig(
36+
level=logging.INFO,
37+
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
38+
)
39+
logger = logging.getLogger(__name__)
40+
41+
42+
# Type variables for generic pipeline
43+
T = TypeVar('T')
44+
R = TypeVar('R')
45+
46+
47+
def timing_decorator(func: Callable) -> Callable:
48+
"""Decorator to measure function execution time"""
49+
@wraps(func)
50+
def wrapper(*args, **kwargs):
51+
start_time = time.perf_counter()
52+
result = func(*args, **kwargs)
53+
end_time = time.perf_counter()
54+
elapsed = end_time - start_time
55+
logger.info(f"{func.__name__} executed in {elapsed:.4f} seconds")
56+
return result
57+
return wrapper
58+
59+
60+
@contextmanager
61+
def pipeline_stage(stage_name: str):
62+
"""Context manager for pipeline stage monitoring"""
63+
logger.info(f"🚀 Starting stage: {stage_name}")
64+
start_time = time.time()
65+
try:
66+
yield
67+
elapsed = time.time() - start_time
68+
logger.info(f"✅ Completed stage: {stage_name} ({elapsed:.2f}s)")
69+
except Exception as e:
70+
elapsed = time.time() - start_time
71+
logger.error(f"❌ Failed stage: {stage_name} after {elapsed:.2f}s - {str(e)}")
72+
raise
73+
74+
75+
@dataclass
76+
class DataRecord:
77+
"""Represents a single data record"""
78+
id: int
79+
features: Dict[str, float]
80+
label: Optional[int] = None
81+
timestamp: datetime = field(default_factory=datetime.now)
82+
83+
def to_dict(self) -> Dict[str, Any]:
84+
return {
85+
'id': self.id,
86+
'features': self.features,
87+
'label': self.label,
88+
'timestamp': self.timestamp.isoformat()
89+
}
90+
91+
92+
class DataGenerator:
93+
"""
94+
Generator class for lazy data loading
95+
96+
Demonstrates:
97+
- Generator functions
98+
- Memory-efficient data loading
99+
- Iterator protocol
100+
"""
101+
102+
def __init__(self, n_samples: int = 1000):
103+
self.n_samples = n_samples
104+
105+
def generate_sample_data(self) -> Iterator[DataRecord]:
106+
"""Generate sample data records lazily"""
107+
for i in range(self.n_samples):
108+
record = DataRecord(
109+
id=i,
110+
features={
111+
'feature_1': float(i % 10),
112+
'feature_2': float(i % 7),
113+
'feature_3': float(i % 5)
114+
},
115+
label=i % 3
116+
)
117+
yield record
118+
119+
def batch_generator(self, batch_size: int = 32) -> Iterator[List[DataRecord]]:
120+
"""Generate batches of data"""
121+
batch = []
122+
for record in self.generate_sample_data():
123+
batch.append(record)
124+
if len(batch) >= batch_size:
125+
yield batch
126+
batch = []
127+
if batch:
128+
yield batch
129+
130+
131+
class PipelineStep(Generic[T, R]):
132+
"""Generic pipeline step with transform functionality"""
133+
134+
def __init__(self, name: str, transform_func: Callable[[T], R]):
135+
self.name = name
136+
self.transform_func = transform_func
137+
138+
@timing_decorator
139+
def execute(self, data: T) -> R:
140+
"""Execute the transformation"""
141+
return self.transform_func(data)
142+
143+
144+
class DataPipeline:
145+
"""
146+
Data Processing Pipeline
147+
148+
Demonstrates:
149+
- Chain of responsibility pattern
150+
- Generic type hints
151+
- Pipeline composition
152+
"""
153+
154+
def __init__(self, name: str = "ML Pipeline"):
155+
self.name = name
156+
self.steps: List[PipelineStep] = []
157+
158+
def add_step(self, name: str, transform_func: Callable) -> 'DataPipeline':
159+
"""Add a step to the pipeline"""
160+
step = PipelineStep(name, transform_func)
161+
self.steps.append(step)
162+
return self
163+
164+
@timing_decorator
165+
def run(self, data: Any) -> Any:
166+
"""Execute all pipeline steps sequentially"""
167+
logger.info(f"Running pipeline: {self.name}")
168+
result = data
169+
170+
for step in self.steps:
171+
with pipeline_stage(step.name):
172+
result = step.execute(result)
173+
174+
logger.info(f"Pipeline {self.name} completed successfully")
175+
return result
176+
177+
178+
def create_sample_pipeline() -> DataPipeline:
179+
"""Create a sample data processing pipeline"""
180+
181+
def normalize_features(data: List[DataRecord]) -> List[DataRecord]:
182+
"""Normalize feature values"""
183+
for record in data:
184+
max_val = max(record.features.values()) or 1
185+
record.features = {k: v/max_val for k, v in record.features.items()}
186+
return data
187+
188+
def filter_outliers(data: List[DataRecord]) -> List[DataRecord]:
189+
"""Filter out outlier records"""
190+
return [r for r in data if sum(r.features.values()) < 20]
191+
192+
def add_metadata(data: List[DataRecord]) -> List[DataRecord]:
193+
"""Add metadata to records"""
194+
for record in data:
195+
record.features['record_count'] = len(data)
196+
return data
197+
198+
pipeline = DataPipeline("Sample ML Pipeline")
199+
pipeline.add_step("Normalize Features", normalize_features)
200+
pipeline.add_step("Filter Outliers", filter_outliers)
201+
pipeline.add_step("Add Metadata", add_metadata)
202+
203+
return pipeline
204+
205+
206+
def main():
207+
"""Main entry point demonstrating the ML pipeline"""
208+
print("=" * 60)
209+
print("Advanced Project: ML Data Pipeline")
210+
print("=" * 60)
211+
212+
# Create data generator
213+
generator = DataGenerator(n_samples=100)
214+
215+
# Load first batch of data
216+
print("\n📊 Loading sample data...")
217+
data_batch = next(generator.batch_generator(batch_size=10))
218+
print(f" Loaded {len(data_batch)} records")
219+
220+
# Display sample record
221+
print("\n📋 Sample Record:")
222+
print(f" {json.dumps(data_batch[0].to_dict(), indent=2)}")
223+
224+
# Create and run pipeline
225+
print("\n⚙️ Creating data processing pipeline...")
226+
pipeline = create_sample_pipeline()
227+
228+
# Execute pipeline
229+
print("\n🔄 Running pipeline...")
230+
processed_data = pipeline.run(data_batch)
231+
232+
# Display results
233+
print(f"\n✅ Processed {len(processed_data)} records")
234+
print("\n📋 Processed Sample Record:")
235+
print(f" {json.dumps(processed_data[0].to_dict(), indent=2)}")
236+
237+
# Demonstrate async-style processing
238+
print("\n⚡ Demonstrating batch processing...")
239+
total_processed = 0
240+
for batch in generator.batch_generator(batch_size=20):
241+
processed = pipeline.run(batch)
242+
total_processed += len(processed)
243+
244+
print(f"\n✨ Total records processed: {total_processed}")
245+
246+
print("\n" + "=" * 60)
247+
print("ML Pipeline demonstration complete!")
248+
print("=" * 60)
249+
print("\n💡 To extend this pipeline:")
250+
print(" - Add database connectors")
251+
print(" - Integrate with scikit-learn models")
252+
print(" - Add distributed processing with Dask/Spark")
253+
print()
254+
255+
256+
if __name__ == '__main__':
257+
main()

0 commit comments

Comments
 (0)