512 lines
16 KiB
Rust
512 lines
16 KiB
Rust
//! Enterprise-grade ETL (Extract, Transform, Load) system for `RustyTorch`
|
|
//!
|
|
//! This crate provides comprehensive ETL capabilities including:
|
|
//! - DAG-based task scheduling with Airflow-like workflow management
|
|
//! - Stream processing with windowing and exactly-once semantics
|
|
//! - Incremental processing with watermarks and state management
|
|
//! - Data lineage tracking and impact analysis
|
|
//! - Transformation pipelines with declarative specifications
|
|
//! - Data quality monitoring with automated checks and remediation
|
|
//! - Multi-source data connectors for files, databases, and streams
|
|
//! - Production-ready monitoring and observability
|
|
//!
|
|
//! # Architecture
|
|
//!
|
|
//! The ETL system is built around several core components:
|
|
//!
|
|
//! ## DAG Engine
|
|
//! - **Task Scheduling**: Priority-based scheduling with resource allocation
|
|
//! - **Dependency Resolution**: Topological sorting with cycle detection
|
|
//! - **Parallel Execution**: Thread pool management with work stealing
|
|
//! - **Retry Logic**: Exponential backoff with jitter and circuit breakers
|
|
//! - **Conditional Execution**: Complex dependency logic with branching
|
|
//!
|
|
//! ## Stream Processing
|
|
//! - **Event Time Processing**: Watermark-based event time handling
|
|
//! - **Windowing**: Tumbling, sliding, and session window operations
|
|
//! - **Exactly-Once Semantics**: Idempotent processing with checkpointing
|
|
//! - **Late Data Handling**: Configurable late arrival policies
|
|
//! - **Stream Analytics**: Real-time aggregations and complex event processing
|
|
//!
|
|
//! ## Incremental Processing
|
|
//! - **Change Detection**: Efficient delta computation algorithms
|
|
//! - **Watermark Management**: Time-ordered data processing
|
|
//! - **State Management**: Persistent state with recovery capabilities
|
|
//! - **Checkpointing**: Fault-tolerant state management
|
|
//! - **Merge Strategies**: Append, upsert, and overwrite operations
|
|
//!
|
|
//! ## Data Lineage
|
|
//! - **Metadata Collection**: Comprehensive transformation tracking
|
|
//! - **Lineage Graphs**: Upstream/downstream relationship modeling
|
|
//! - **Impact Analysis**: Change propagation modeling
|
|
//! - **Schema Evolution**: Compatibility analysis and versioning
|
|
//! - **Audit Trails**: Compliance and debugging support
|
|
//!
|
|
//! # Examples
|
|
//!
|
|
//! ## Basic ETL Pipeline
|
|
//!
|
|
//! ```rust
|
|
//! use rtx_etl::{
|
|
//! EtlEngine, EtlConfig, Task, TaskGraph, DataSource,
|
|
//! Transformation, DataSink
|
|
//! };
|
|
//!
|
|
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
|
|
//! // Create ETL engine with configuration
|
|
//! let config = EtlConfig::builder()
|
|
//! .with_dag_scheduling(true)
|
|
//! .with_stream_processing(true)
|
|
//! .with_lineage_tracking(true)
|
|
//! .build()?;
|
|
//!
|
|
//! let mut engine = EtlEngine::new(config).await?;
|
|
//!
|
|
//! // Define data pipeline
|
|
//! let extract_task = Task::new("extract_users")
|
|
//! .with_source(DataSource::database("postgresql://localhost/app"))
|
|
//! .with_sql("SELECT * FROM users WHERE updated_at > ?");
|
|
//!
|
|
//! let transform_task = Task::new("transform_users")
|
|
//! .depends_on("extract_users")
|
|
//! .with_transformation(Transformation::sql(
|
|
//! "SELECT user_id, UPPER(name) as name, age + 1 as next_age FROM users"
|
|
//! ));
|
|
//!
|
|
//! let load_task = Task::new("load_users")
|
|
//! .depends_on("transform_users")
|
|
//! .with_sink(DataSink::parquet("s3://bucket/processed/users.parquet"));
|
|
//!
|
|
//! // Create and execute DAG
|
|
//! let mut dag = TaskGraph::new();
|
|
//! dag.add_task(extract_task);
|
|
//! dag.add_task(transform_task);
|
|
//! dag.add_task(load_task);
|
|
//!
|
|
//! let execution_result = engine.execute_dag(dag).await?;
|
|
//! println!("Pipeline completed: {:?}", execution_result);
|
|
//! # Ok(())
|
|
//! # }
|
|
//! ```
|
|
//!
|
|
//! ## Stream Processing Pipeline
|
|
//!
|
|
//! ```rust
|
|
//! use rtx_etl::{
|
|
//! StreamProcessor, StreamConfig, Window, WindowType,
|
|
//! StreamSource, StreamSink
|
|
//! };
|
|
//! use chrono::Duration;
|
|
//!
|
|
//! # async fn stream_example() -> Result<(), Box<dyn std::error::Error>> {
|
|
//! let config = StreamConfig::builder()
|
|
//! .with_exactly_once_semantics(true)
|
|
//! .with_watermark_strategy("bounded_out_of_orderness", Duration::minutes(5))
|
|
//! .build()?;
|
|
//!
|
|
//! let mut processor = StreamProcessor::new(config).await?;
|
|
//!
|
|
//! // Define stream processing pipeline
|
|
//! let kafka_source = StreamSource::kafka("user_events")
|
|
//! .with_bootstrap_servers("localhost:9092")
|
|
//! .with_consumer_group("etl_pipeline");
|
|
//!
|
|
//! let tumbling_window = Window::tumbling(Duration::minutes(5))
|
|
//! .trigger_on_watermark()
|
|
//! .allowed_lateness(Duration::minutes(1));
|
|
//!
|
|
//! processor
|
|
//! .from_source(kafka_source)
|
|
//! .key_by("user_id")
|
|
//! .window(tumbling_window)
|
|
//! .aggregate("COUNT(*) as event_count, MAX(timestamp) as last_seen")
|
|
//! .to_sink(StreamSink::redis("user_metrics"));
|
|
//!
|
|
//! processor.start().await?;
|
|
//! # Ok(())
|
|
//! # }
|
|
//! ```
|
|
//!
|
|
//! ## Incremental Processing with State Management
|
|
//!
|
|
//! ```rust
|
|
//! use rtx_etl::{
|
|
//! IncrementalProcessor, StateManager, Checkpoint,
|
|
//! ChangeDetectionStrategy, MergeStrategy
|
|
//! };
|
|
//!
|
|
//! # async fn incremental_example() -> Result<(), Box<dyn std::error::Error>> {
|
|
//! let mut processor = IncrementalProcessor::builder()
|
|
//! .with_state_backend(StateManager::redis("redis://localhost:6379"))
|
|
//! .with_change_detection(ChangeDetectionStrategy::timestamp_based("updated_at"))
|
|
//! .with_checkpoint_interval(std::time::Duration::from_secs(300))
|
|
//! .build().await?;
|
|
//!
|
|
//! // Process incremental changes
|
|
//! let checkpoint = processor.load_checkpoint("user_processing").await?;
|
|
//! let changes = processor.detect_changes("users_table", checkpoint).await?;
|
|
//!
|
|
//! for batch in changes.into_batches(1000) {
|
|
//! let transformed = processor.transform(batch).await?;
|
|
//! processor.merge(transformed, MergeStrategy::Upsert).await?;
|
|
//! processor.save_checkpoint("user_processing").await?;
|
|
//! }
|
|
//! # Ok(())
|
|
//! # }
|
|
//! ```
|
|
|
|
#![allow(clippy::missing_errors_doc, clippy::module_name_repetitions)]
|
|
|
|
use chrono::{DateTime, Duration, Utc};
|
|
use serde::{Deserialize, Serialize};
|
|
use std::collections::HashMap;
|
|
use uuid::Uuid;
|
|
|
|
// Core modules
|
|
pub mod connectors;
|
|
pub mod dag;
|
|
pub mod engine;
|
|
pub mod incremental;
|
|
pub mod lineage;
|
|
pub mod monitoring;
|
|
pub mod quality;
|
|
pub mod state;
|
|
pub mod stream;
|
|
pub mod transform;
|
|
|
|
// Re-exports for convenient access
|
|
pub use connectors::{
|
|
Connector, ConnectorConfig, DataSink, DataSource, DatabaseConnector, FileConnector,
|
|
StreamConnector, StreamSink, StreamSource,
|
|
};
|
|
pub use dag::{
|
|
DagConfig, DagExecution, DagScheduler, ExecutionContext, Task, TaskBuilder, TaskGraph,
|
|
TaskResult, TaskStatus,
|
|
};
|
|
pub use engine::{EtlConfig, EtlConfigBuilder, EtlEngine, ExecutionResult};
|
|
pub use incremental::{
|
|
ChangeDetectionStrategy, ChangeDetector, Checkpoint, IncrementalConfig, IncrementalProcessor,
|
|
MergeStrategy,
|
|
};
|
|
pub use lineage::{
|
|
DataLineage, ImpactAnalysis, LineageConfig, LineageGraph, LineageNode, LineageTracker,
|
|
SchemaEvolutionTracker,
|
|
};
|
|
pub use monitoring::{AlertManager, Dashboard, EtlMetrics, MetricCollector, PerformanceMonitor};
|
|
pub use quality::{
|
|
DataProfiler, QualityAlert, QualityCheck, QualityConfig, QualityMonitor, QualityReport,
|
|
QualityRule,
|
|
};
|
|
pub use state::{
|
|
PostgresStateManager, RedisStateManager, StateBackend, StateManager, StateRecovery,
|
|
StateSnapshot,
|
|
};
|
|
pub use stream::{EventTime, StreamConfig, StreamProcessor, Watermark, Window, WindowType};
|
|
pub use transform::{
|
|
SqlTransformation, TransformFunction, Transformation, TransformationBuilder,
|
|
TransformationPipeline,
|
|
};
|
|
|
|
/// Comprehensive error types for ETL operations
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum EtlError {
|
|
/// Task execution error
|
|
#[error("Task '{task_id}' failed: {message}")]
|
|
TaskExecution { task_id: String, message: String },
|
|
|
|
/// DAG validation error (cycles, missing dependencies, etc.)
|
|
#[error("DAG validation error: {0}")]
|
|
DagValidation(String),
|
|
|
|
/// Stream processing error
|
|
#[error("Stream processing error: {0}")]
|
|
StreamProcessing(String),
|
|
|
|
/// Incremental processing error
|
|
#[error("Incremental processing error: {0}")]
|
|
IncrementalProcessing(String),
|
|
|
|
/// Data transformation error
|
|
#[error("Transformation error: {0}")]
|
|
Transformation(String),
|
|
|
|
/// Data source connection error
|
|
#[error("Data source error: {0}")]
|
|
DataSource(String),
|
|
|
|
/// Data sink error
|
|
#[error("Data sink error: {0}")]
|
|
DataSink(String),
|
|
|
|
/// Data validation error
|
|
#[error("Data validation error: {0}")]
|
|
Validation(String),
|
|
|
|
/// State management error
|
|
#[error("State management error: {0}")]
|
|
StateManagement(String),
|
|
|
|
/// Data quality check failure
|
|
#[error("Data quality check failed: {0}")]
|
|
DataQuality(String),
|
|
|
|
/// Lineage tracking error
|
|
#[error("Lineage tracking error: {0}")]
|
|
Lineage(String),
|
|
|
|
/// Configuration error
|
|
#[error("Configuration error: {0}")]
|
|
Config(String),
|
|
|
|
/// Resource management error (memory, CPU, etc.)
|
|
#[error("Resource error: {0}")]
|
|
Resource(String),
|
|
|
|
/// Timeout error
|
|
#[error("Operation timed out after {timeout_ms}ms: {operation}")]
|
|
Timeout { operation: String, timeout_ms: u64 },
|
|
|
|
/// Serialization/deserialization error
|
|
#[error("Serialization error: {0}")]
|
|
Serde(#[from] serde_json::Error),
|
|
|
|
/// I/O operation error
|
|
#[error("I/O error: {0}")]
|
|
Io(#[from] std::io::Error),
|
|
|
|
/// Database error
|
|
#[error("Database error: {0}")]
|
|
Database(String),
|
|
|
|
/// Generic error for other cases
|
|
#[error("ETL error: {0}")]
|
|
Other(String),
|
|
}
|
|
|
|
impl EtlError {
|
|
/// Create a validation error
|
|
#[must_use]
|
|
pub fn validation(message: String) -> Self {
|
|
Self::Validation(message)
|
|
}
|
|
}
|
|
|
|
/// Result type for ETL operations
|
|
pub type Result<T> = std::result::Result<T, EtlError>;
|
|
|
|
/// Core data structure representing processed data in the ETL pipeline
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct DataRecord {
|
|
/// Unique identifier for the record
|
|
pub id: Uuid,
|
|
/// Event timestamp (when the data event occurred)
|
|
pub event_time: DateTime<Utc>,
|
|
/// Processing timestamp (when the record was processed)
|
|
pub process_time: DateTime<Utc>,
|
|
/// Data partition key for distributed processing
|
|
pub partition_key: Option<String>,
|
|
/// The actual data payload
|
|
pub data: DataPayload,
|
|
/// Metadata including lineage and quality information
|
|
pub metadata: RecordMetadata,
|
|
}
|
|
|
|
/// The actual data content of a record
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum DataPayload {
|
|
/// Structured data as key-value pairs
|
|
Structured(HashMap<String, DataValue>),
|
|
/// Semi-structured data (JSON, XML, etc.)
|
|
SemiStructured(serde_json::Value),
|
|
/// Raw binary data
|
|
Binary(Vec<u8>),
|
|
/// Text content
|
|
Text(String),
|
|
}
|
|
|
|
/// Individual data values within records
|
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
|
pub enum DataValue {
|
|
/// Null/missing value
|
|
Null,
|
|
/// Boolean value
|
|
Bool(bool),
|
|
/// Integer value
|
|
Int(i64),
|
|
/// Floating point value
|
|
Float(f64),
|
|
/// String value
|
|
String(String),
|
|
/// Array of values
|
|
Array(Vec<Self>),
|
|
/// Nested object
|
|
Object(HashMap<String, Self>),
|
|
/// Timestamp value
|
|
Timestamp(DateTime<Utc>),
|
|
/// Duration value
|
|
Duration(Duration),
|
|
/// Binary data
|
|
Binary(Vec<u8>),
|
|
}
|
|
|
|
/// Metadata associated with each data record
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct RecordMetadata {
|
|
/// Source system identifier
|
|
pub source: String,
|
|
/// Transformation lineage
|
|
pub lineage: Vec<TransformationTrace>,
|
|
/// Data quality scores
|
|
pub quality_scores: Option<QualityScores>,
|
|
/// Custom attributes
|
|
pub attributes: HashMap<String, String>,
|
|
/// Schema version
|
|
pub schema_version: Option<String>,
|
|
/// Checksum for integrity verification
|
|
pub checksum: Option<String>,
|
|
}
|
|
|
|
/// Trace of a transformation applied to the data
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct TransformationTrace {
|
|
/// Transformation identifier
|
|
pub transform_id: String,
|
|
/// Timestamp when transformation was applied
|
|
pub applied_at: DateTime<Utc>,
|
|
/// Transformation parameters
|
|
pub parameters: HashMap<String, String>,
|
|
/// Input schema hash
|
|
pub input_schema_hash: Option<String>,
|
|
/// Output schema hash
|
|
pub output_schema_hash: Option<String>,
|
|
}
|
|
|
|
/// Quality scores for data validation
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct QualityScores {
|
|
/// Overall quality score (0.0 to 1.0)
|
|
pub overall: f64,
|
|
/// Completeness score
|
|
pub completeness: f64,
|
|
/// Accuracy score
|
|
pub accuracy: f64,
|
|
/// Consistency score
|
|
pub consistency: f64,
|
|
/// Timeliness score
|
|
pub timeliness: f64,
|
|
/// Uniqueness score
|
|
pub uniqueness: f64,
|
|
}
|
|
|
|
impl DataValue {
|
|
/// Check if the value is null/missing
|
|
#[must_use]
|
|
pub fn is_null(&self) -> bool {
|
|
matches!(self, Self::Null)
|
|
}
|
|
|
|
/// Get the type name of the value
|
|
#[must_use]
|
|
pub fn type_name(&self) -> &'static str {
|
|
match self {
|
|
Self::Null => "null",
|
|
Self::Bool(_) => "boolean",
|
|
Self::Int(_) => "integer",
|
|
Self::Float(_) => "float",
|
|
Self::String(_) => "string",
|
|
Self::Array(_) => "array",
|
|
Self::Object(_) => "object",
|
|
Self::Timestamp(_) => "timestamp",
|
|
Self::Duration(_) => "duration",
|
|
Self::Binary(_) => "binary",
|
|
}
|
|
}
|
|
|
|
/// Convert to f64 if possible for numerical operations
|
|
#[must_use]
|
|
pub fn as_f64(&self) -> Option<f64> {
|
|
match self {
|
|
Self::Int(i) => Some(*i as f64),
|
|
Self::Float(f) => Some(*f),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Convert to string representation
|
|
#[must_use]
|
|
pub fn as_string(&self) -> Option<String> {
|
|
match self {
|
|
Self::String(s) => Some(s.clone()),
|
|
Self::Int(i) => Some(i.to_string()),
|
|
Self::Float(f) => Some(f.to_string()),
|
|
Self::Bool(b) => Some(b.to_string()),
|
|
_ => None,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<serde_json::Value> for DataValue {
|
|
fn from(value: serde_json::Value) -> Self {
|
|
match value {
|
|
serde_json::Value::Null => Self::Null,
|
|
serde_json::Value::Bool(b) => Self::Bool(b),
|
|
serde_json::Value::Number(n) => {
|
|
if let Some(i) = n.as_i64() {
|
|
Self::Int(i)
|
|
} else if let Some(f) = n.as_f64() {
|
|
Self::Float(f)
|
|
} else {
|
|
Self::Null
|
|
}
|
|
}
|
|
serde_json::Value::String(s) => Self::String(s),
|
|
serde_json::Value::Array(arr) => Self::Array(arr.into_iter().map(Self::from).collect()),
|
|
serde_json::Value::Object(obj) => {
|
|
Self::Object(obj.into_iter().map(|(k, v)| (k, Self::from(v))).collect())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Global configuration for the ETL system
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct EtlSystemConfig {
|
|
/// Enable DAG scheduling
|
|
pub dag_scheduling: bool,
|
|
/// Enable stream processing
|
|
pub stream_processing: bool,
|
|
/// Enable lineage tracking
|
|
pub lineage_tracking: bool,
|
|
/// Enable incremental processing
|
|
pub incremental_processing: bool,
|
|
/// Enable data quality monitoring
|
|
pub quality_monitoring: bool,
|
|
/// Maximum number of concurrent tasks
|
|
pub max_concurrent_tasks: usize,
|
|
/// Task timeout in milliseconds
|
|
pub task_timeout_ms: u64,
|
|
/// Enable performance metrics collection
|
|
pub collect_metrics: bool,
|
|
/// Enable alerting
|
|
pub enable_alerts: bool,
|
|
/// Checkpoint interval for state management
|
|
pub checkpoint_interval_ms: u64,
|
|
}
|
|
|
|
impl Default for EtlSystemConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
dag_scheduling: true,
|
|
stream_processing: true,
|
|
lineage_tracking: true,
|
|
incremental_processing: true,
|
|
quality_monitoring: true,
|
|
max_concurrent_tasks: 10,
|
|
task_timeout_ms: 300000, // 5 minutes
|
|
collect_metrics: true,
|
|
enable_alerts: true,
|
|
checkpoint_interval_ms: 60000, // 1 minute
|
|
}
|
|
}
|
|
}
|