Media Streaming Analytics Pipeline

-- Streaming Events Fact Table
CREATE TABLE fact_stream_events (
    event_id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    content_id BIGINT NOT NULL,
    event_timestamp TIMESTAMP(3) NOT NULL,
    event_type VARCHAR(50) NOT NULL,
    position_seconds INT,
    quality_level VARCHAR(20),
    device_type VARCHAR(50),
    network_quality VARCHAR(20),
    session_id VARCHAR(100),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_user_content_time (user_id, content_id, event_timestamp),
    INDEX idx_event_type (event_type),
    INDEX idx_session (session_id)
);

-- Content Dimension Table
CREATE TABLE dim_content (
    content_id BIGINT PRIMARY KEY AUTO_INCREMENT,
    content_title VARCHAR(255) NOT NULL,
    content_type VARCHAR(50),
    genre_primary VARCHAR(100),
    genre_secondary JSON,
    release_date DATE,
    duration_minutes INT,
    language VARCHAR(10),
    age_rating VARCHAR(10),
    cast_members JSON,
    director VARCHAR(255)
);

-- User Dimension Table
CREATE TABLE dim_users (
    user_id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_name VARCHAR(255),
    age_group VARCHAR(20),
    location_country VARCHAR(100),
    subscription_type VARCHAR(50),
    join_date DATE,
    preferences JSON
);

// Real-time Streaming Analytics with Flink
@Slf4j
public class StreamingAnalyticsProcessor {
    
    private final StreamExecutionEnvironment env;
    private final KafkaSource<StreamEvent> source;
    
    public StreamingAnalyticsProcessor() {
        this.env = StreamExecutionEnvironment.getExecutionEnvironment();
        this.source = KafkaSource.<StreamEvent>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("streaming-events")
            .setGroupId("analytics-processor")
            .setStartingOffsets(OffsetsInitializer.latest())
            .build();
    }
    
    public void processStream() {
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Streaming Events")
            .map(new EventDeserializer())
            .keyBy(StreamEvent::getUserId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .aggregate(new UserEngagementAggregator())
            .process(new RecommendationProcessor())
            .addSink(new RedisSink());
    }
}

// User Engagement Aggregator
public class UserEngagementAggregator implements AggregateFunction<StreamEvent, UserEngagement, UserEngagement> {
    
    @Override
    public UserEngagement createAccumulator() {
        return new UserEngagement();
    }
    
    @Override
    public UserEngagement add(StreamEvent event, UserEngagement accumulator) {
        accumulator.addEvent(event);
        return accumulator;
    }
    
    @Override
    public UserEngagement getResult(UserEngagement accumulator) {
        return accumulator;
    }
    
    @Override
    public UserEngagement merge(UserEngagement a, UserEngagement b) {
        return a.merge(b);
    }
}

# Content Recommendation Engine
import numpy as np
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import redis
import json

class ContentRecommendationEngine:
    def __init__(self, redis_client):
        self.redis_client = redis_client
        self.tfidf = TfidfVectorizer(max_features=1000)
        self.content_features = None
        
    def train_model(self, content_data):
        """Train the recommendation model"""
        # Prepare content features
        content_text = content_data['title'] + ' ' + content_data['genre'] + ' ' + content_data['description']
        
        # TF-IDF vectorization
        self.content_features = self.tfidf.fit_transform(content_text)
        
        # Store in Redis for real-time access
        self._store_features()
    
    def get_recommendations(self, user_id, n_recommendations=10):
        """Get personalized recommendations for a user"""
        # Get user preferences and viewing history
        user_profile = self._get_user_profile(user_id)
        
        # Calculate similarity scores
        user_vector = self._create_user_vector(user_profile)
        similarities = cosine_similarity(user_vector, self.content_features)
        
        # Get top recommendations
        top_indices = np.argsort(similarities[0])[-n_recommendations:][::-1]
        
        return [self.content_data.iloc[i] for i in top_indices]
    
    def update_user_preferences(self, user_id, content_id, rating):
        """Update user preferences based on new interactions"""
        user_key = f"user:{user_id}:preferences"
        
        # Get current preferences
        current_prefs = self.redis_client.get(user_key)
        if current_prefs:
            prefs = json.loads(current_prefs)
        else:
            prefs = {'ratings': {}, 'genres': {}, 'actors': {}}
        
        # Update preferences
        prefs['ratings'][content_id] = rating
        
        # Store updated preferences
        self.redis_client.setex(user_key, 3600, json.dumps(prefs))  # 1 hour TTL
    
    def _store_features(self):
        """Store model features in Redis for real-time access"""
        feature_key = "content:features"
        self.redis_client.setex(feature_key, 3600, self.content_features.tobytes())
    
    def _get_user_profile(self, user_id):
        """Retrieve user profile from Redis"""
        user_key = f"user:{user_id}:profile"
        profile = self.redis_client.get(user_key)
        return json.loads(profile) if profile else {}
    
    def _create_user_vector(self, user_profile):
        """Create user feature vector for similarity calculation"""
        # Implementation depends on user profile structure
        # This is a simplified example
        return np.zeros((1, self.content_features.shape[1]))