import base64 import json import logging import os from typing import Dict, List from datetime import datetime, timedelta # PyFlink imports from pyflink.table import EnvironmentSettings, TableEnvironment, StreamTableEnvironment, DataTypes from pyflink.table.udf import udf from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import Row from pyflink.common.configuration import Configuration import redis logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class CollaborativeFilteringService: """Service for collaborative filtering recommendations""" def __init__(self, kinesis_stream, stream_arn, redis_host, redis_port, config): self.kinesis_stream = kinesis_stream self.stream_arn = stream_arn self.redis_host = redis_host self.redis_port = redis_port self.config = config from pathlib import Path if self.config.env == "prod" or self.config.env == "staging": # Initialize Flink environment for cloud deployment current_dir = Path(__file__).resolve().parent pyflink_jar_path = current_dir / "lib/pyflink-dependencies.jar" self.stream_env = StreamExecutionEnvironment.get_execution_environment() conf = Configuration() if pyflink_jar_path.exists(): conf.set_string("pipeline.jars", f"file://{pyflink_jar_path}") print(f"Setting pipeline.jars to: file://{pyflink_jar_path}") else: print(f"⚠️ Warning: {pyflink_jar_path} not found") self.env_settings = ( EnvironmentSettings .new_instance() .with_configuration(conf) .in_streaming_mode() .build() ) self.table_env = StreamTableEnvironment.create(self.stream_env, self.env_settings) # Configure environment and register UDFs self._configure_environment() self._register_udfs() else: # Initialize Flink environment for local development current_dir = Path(__file__).resolve().parent.parent # Go up one level from app-collaborative-filtering kinesis_jar_path = current_dir / "jar/flink-sql-connector-kinesis-5.0.0-1.20.jar" self.stream_env = StreamExecutionEnvironment.get_execution_environment() conf = Configuration() if kinesis_jar_path.exists(): conf.set_string("pipeline.jars", f"file://{kinesis_jar_path}") print(f"Loaded Kinesis connector from: {kinesis_jar_path}") else: print(f"⚠️ Warning: Kinesis JAR not found at {kinesis_jar_path}") self.env_settings = ( EnvironmentSettings .new_instance() .with_configuration(conf) .in_streaming_mode() .build() ) self.table_env = StreamTableEnvironment.create(self.stream_env, self.env_settings) # Configure environment and register UDFs self._configure_environment() self._register_udfs() # Initialize Redis client self.redis_client = redis.Redis( host=redis_host, port=redis_port, decode_responses=True, db=0 ) def _configure_environment(self): """Configure Flink environment for collaborative filtering pipeline""" configuration = self.table_env.get_config().get_configuration() # Local development optimizations if self.config.env == "dev": # Increase network buffers for local development configuration.set_string("taskmanager.memory.network.fraction", "0.3") # Reduce parallelism for local development configuration.set_string("parallelism.default", "1") # Increase task manager memory configuration.set_string("taskmanager.memory.process.size", "2gb") # 1. Set State TTL to 24 hours (automatically removes old state) configuration.set_string("table.exec.state.ttl", "36h") # 2. Enable idle state cleanup (removes unused state) configuration.set_string("table.exec.state.ttl.cleanup.strategy", "delete") # optional delete optimization #configuration.set_string("table.exec.state.ttl.cleanup.interval", "1h") configuration.set_string("table.exec.mini-batch.enabled", "true") configuration.set_string("table.exec.mini-batch.allow-latency", "5s") configuration.set_string("table.exec.mini-batch.size", "5000") def _register_udfs(self): """Register UDFs for collaborative filtering using timestamps""" @udf(result_type=DataTypes.ROW([ DataTypes.FIELD("userId", DataTypes.STRING()), DataTypes.FIELD("songId", DataTypes.STRING()), DataTypes.FIELD("actionName", DataTypes.STRING()), DataTypes.FIELD("event_timestamp", DataTypes.STRING()) ])) def parse_request_body(request_body: str): """Parse JSON request body into structured data""" try: payload = base64.b64decode(request_body) json_value = json.loads(payload) ts_str = json_value.get('timestamp') event_timestamp = datetime.fromisoformat(ts_str.replace('Z', '+00:00')) if ts_str else None actionName = json_value.get('properties').get('actionName') if json_value.get('properties', {}).get('context', {}).get('likeStatus') is True: actionName = 'like' elif json_value.get('properties', {}).get('context', {}).get('likeStatus') is False: actionName = 'dislike' else: pass return Row( json_value.get('properties').get('userId'), json_value.get('properties').get('principal_object_value'), actionName, ts_str ) except Exception as e: return None # UDF to write recommendations to ALL users when someone likes a hook # Add this to your _register_udfs() method redis_host_cfg = self.redis_host redis_port_cfg = self.redis_port @udf(result_type=DataTypes.BOOLEAN()) def write_cooccurrence_to_redis(hook1: str, recommended_hook_ids_str: str) -> bool: """Write co-occurrence recommendations to Redis""" try: import redis redis_client = redis.Redis(host=redis_host_cfg, port=int(redis_port_cfg), decode_responses=True) rec_key = f"hook_recommendations:{hook1}" redis_client.set(rec_key, recommended_hook_ids_str) redis_client.expire(rec_key, 86400) # TTL 24 hours return True except Exception as e: print(f"Error writing co-occurrence for hook {hook1}: {e}") logger.error(f"Error writing co-occurrence for hook {hook1}: {e}") return False # Register the UDF self.table_env.create_temporary_system_function("write_cooccurrence_to_redis", write_cooccurrence_to_redis) # Register functions self.table_env.create_temporary_system_function("parse_request_body", parse_request_body) logger.info("UDFs registered successfully (timestamp-based recommendations)") def create_kinesis_source_table(self): """Create Kinesis source table for hook events""" # Use the correct schema based on actual data format ddl = f""" CREATE TABLE hook_events ( `name` STRING, `timestamp` STRING, `user_id` STRING, `userId` STRING, `source` STRING, `session_id` STRING, `properties` ROW< `hook_id` STRING, `hookId` STRING, `hookPlayDuration` DOUBLE, `hook_status` STRING, `recommendation_item_id` STRING >, `request_id` STRING, `_datadog` ROW<`x-datadog-trace-id` STRING>, `proc_time` AS PROCTIME() ) WITH ( 'connector' = 'kinesis', 'stream.arn' = '{self.stream_arn}', 'aws.credentials.provider' = 'AUTO', 'aws.region' = 'us-east-2', 'source.init.position' = 'LATEST', 'format' = 'json', 'source.reader.type' = 'EFO', 'source.efo.consumer.name' = 'collaborative-filtering-consumer-{self.config.env}', 'source.efo.lifecycle' = 'SELF_MANAGED' ) """ self.table_env.execute_sql(ddl) print("Executing DDL:") print(ddl) # Create a parsed view using the UDF parsed_view_sql = """ CREATE VIEW hook_events_view AS SELECT COALESCE(user_id, userId, 'NULL') as user_id, COALESCE(properties.hook_id, properties.hookId, 'NULL') as hook_id, name as event_type, `timestamp` as event_timestamp, `proc_time` as proc_time FROM hook_events WHERE name IN ('HookLike', 'HookUndoLike') AND (user_id IS NOT NULL OR userId IS NOT NULL) AND (properties.hook_id IS NOT NULL OR properties.hookId IS NOT NULL) AND proc_time >= CURRENT_TIMESTAMP - INTERVAL '24' HOUR """ self.table_env.execute_sql(parsed_view_sql) print("Executing parsed view SQL:") print(parsed_view_sql) logger.info("Kinesis source table created") def create_redis_sink(self): """Create Redis sink for storing collaborative filtering recommendations""" redis_sink_sql = """ CREATE TABLE user_recommendations_sink ( `temp_value` string ) WITH ( 'connector' = 'print' ) """ self.table_env.execute_sql(redis_sink_sql) logger.info("Redis sink created") def run_pipeline(self): """Run the collaborative filtering pipeline with automatic view recreation and 24-hour data retention""" try: # Create source table self.create_kinesis_source_table() job = self.create_direct_cf() # Continue with pipeline job.wait() except Exception as e: logger.error(f"Error in pipeline execution: {e}") raise e def create_direct_cf(self): """Direct collaborative filtering with top 100 recent users logic""" # Step 2: Create user-hook matrix (who liked what) user_hook_matrix_sql = """ CREATE VIEW user_hook_matrix AS SELECT user_id, hook_id, event_timestamp FROM ( SELECT user_id, hook_id, event_timestamp, event_type, ROW_NUMBER() OVER ( PARTITION BY user_id, hook_id ORDER BY event_timestamp DESC ) AS row_num FROM hook_events_view WHERE user_id IS NOT NULL AND hook_id IS NOT NULL AND event_type IN ('HookLike', 'HookUndoLike') AND proc_time >= CURRENT_TIMESTAMP - INTERVAL '24' HOUR ) WHERE row_num = 1 AND event_type = 'HookLike' """ self.table_env.execute_sql(user_hook_matrix_sql) # Step 3: Find first 20 users who liked the same hook most recently top_100_recent_users_sql = """ CREATE VIEW top_100_recent_users AS SELECT * FROM ( SELECT a.user_id as target_user_id, a.hook_id as liked_hook_id, b.user_id as similar_user_id, b.event_timestamp, ROW_NUMBER() OVER ( PARTITION BY a.user_id, a.hook_id ORDER BY b.event_timestamp DESC ) as rn FROM user_hook_matrix a JOIN user_hook_matrix b ON a.hook_id = b.hook_id WHERE a.user_id <> b.user_id ) WHERE rn <= 20 """ self.table_env.execute_sql(top_100_recent_users_sql) # Step 4: Get recommendations from these top 100 users direct_recommendations_sql = """ CREATE VIEW direct_recommendations AS SELECT target_user_id, LISTAGG(CAST(recommended_hook_id AS STRING), ',') AS recommended_hooks, COUNT(*) AS num_recommendations FROM ( SELECT DISTINCT target_user_id, recommended_hook_id, latest_event_timestamp FROM ( SELECT target_user_id, recommended_hook_id, latest_event_timestamp, ROW_NUMBER() OVER ( PARTITION BY target_user_id ORDER BY latest_event_timestamp DESC ) AS rn FROM ( SELECT target_user_id, recommended_hook_id, MAX(event_timestamp) AS latest_event_timestamp FROM ( -- Branch A: Recommendations for target users SELECT a.target_user_id, b.hook_id AS recommended_hook_id, b.event_timestamp FROM top_100_recent_users a JOIN user_hook_matrix b ON a.similar_user_id = b.user_id WHERE b.hook_id <> a.liked_hook_id AND NOT EXISTS ( SELECT 1 FROM user_hook_matrix c WHERE c.user_id = a.target_user_id AND c.hook_id = b.hook_id ) UNION ALL -- Branch B: Recommendations for similar users SELECT a.similar_user_id AS target_user_id, b.hook_id AS recommended_hook_id, b.event_timestamp FROM top_100_recent_users a JOIN user_hook_matrix b ON a.target_user_id = b.user_id WHERE b.hook_id <> a.liked_hook_id AND NOT EXISTS ( SELECT 1 FROM user_hook_matrix c WHERE c.user_id = a.similar_user_id AND c.hook_id = b.hook_id ) ) candidates GROUP BY target_user_id, recommended_hook_id ) collapsed ) ranked WHERE rn <= 20 ) topk GROUP BY target_user_id; """ self.table_env.execute_sql(direct_recommendations_sql) # Step 5: Create sink for individual recommendations create_sink_sql = """ CREATE TABLE efficient_cf_sink ( target_user_id STRING, recommended_hooks STRING, success BOOLEAN ) WITH ( 'connector' = 'print', 'print-identifier' = 'EFFICIENT-CF' ) """ self.table_env.execute_sql(create_sink_sql) insert_sql = """ INSERT INTO efficient_cf_sink SELECT target_user_id, recommended_hooks, write_cooccurrence_to_redis(target_user_id, recommended_hooks) as success FROM direct_recommendations """ print("Starting direct collaborative filtering with top 100 users...") job = self.table_env.execute_sql(insert_sql) return job # Simplified configuration class SimpleConfig: """Configuration for collaborative filtering service with CDK property group support""" def __init__(self): # Set default values first (for local development) self.kinesis_stream = "rec-events-stream" self.stream_arn = "arn:aws:kinesis:us-east-2:590183763515:stream/rec-events-stream" self.redis_host = "localhost" self.redis_port = 6379 self.env = "dev" # Try to get runtime properties (for cloud deployment) try: from pyflink.java_gateway import get_gateway gateway = get_gateway() # Get application properties using KinesisAnalyticsRuntime j_kinesis_analytics_runtime = gateway.jvm.com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime application_properties = j_kinesis_analytics_runtime.getApplicationProperties() # Get the 'rec.config' property group (as defined in CDK) rec_config_properties = application_properties.get('rec.config') if rec_config_properties: # Override defaults with runtime properties self.kinesis_stream = rec_config_properties.getProperty('stream.name') or self.kinesis_stream self.stream_arn = rec_config_properties.getProperty('stream.arn') or self.stream_arn self.redis_host = rec_config_properties.getProperty('redis.host') or self.redis_host self.redis_port = int(rec_config_properties.getProperty('redis.port') or str(self.redis_port)) self.env = rec_config_properties.getProperty('env.stage') or self.env except Exception as e: logger.error(f"Error getting application properties: {e}") logger.info("Using default local development configuration") self.hooks_per_user = 50 self.state_ttl_hours = 24 def main(): """Main entry point for timestamp-based recommendation service""" config = SimpleConfig() try: # Create service service = CollaborativeFilteringService( kinesis_stream=config.kinesis_stream, stream_arn=config.stream_arn, redis_host=config.redis_host, redis_port=config.redis_port, config=config ) # Test Redis connection service.redis_client.ping() logger.info("Redis connection successful") # Run pipeline service.run_pipeline() except Exception as e: logger.error(f"Service failed: {e}") raise if __name__ == "__main__": # Uncomment to generate test data # generate_simple_test_data() # Run the service main()