import logging import os import time from pathlib import Path import redis # PyFlink imports from pyflink.table import EnvironmentSettings, StreamTableEnvironment, DataTypes from pyflink.table.udf import udf from pyflink.datastream import StreamExecutionEnvironment from pyflink.common.configuration import Configuration from pyflink.java_gateway import get_gateway # Configuration aws_region = "us-east-2" # Configuration class to read runtime properties class SimpleConfig: """Configuration for hooks comment signal service with CDK property group support""" def __init__(self): # Set default values first (for local development) 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: 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.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: logging.error(f"Error getting application properties: {e}") logging.info("Using default local development configuration") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class HooksCommentSignalService: """Service for processing hooks comment signals (comments, likes on comments)""" def __init__(self, config): self.config = config if self.config.env == "prod" or self.config.env == "staging": # Initialize Flink environment for production/staging current_dir = Path(__file__).resolve().parent pyflink_jar_path = current_dir / "lib/pyflink-dependencies.jar" 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.stream_env = StreamExecutionEnvironment.get_execution_environment() self.env_settings = ( EnvironmentSettings.new_instance() .with_configuration(conf) .in_streaming_mode() .build() ) self.table_env = StreamTableEnvironment.create( self.stream_env, self.env_settings ) # Add JARs to the table environment if os.path.exists(pyflink_jar_path): jar_url = f"file://{pyflink_jar_path}" print(f"Setting pipeline.jars to: {jar_url}") self.table_env.get_config().get_configuration().set_string( "pipeline.jars", jar_url ) self._configure_environment() self._register_udfs() else: # Initialize Flink environment for local development self.stream_env = StreamExecutionEnvironment.get_execution_environment() # Add connector JAR for Kinesis current_dir = os.path.dirname(os.path.abspath(__file__)) parent_dir = os.path.dirname(current_dir) kinesis_jar_path = os.path.join( parent_dir, "jar/flink-sql-connector-kinesis-5.0.0-1.20.jar" ) conf = Configuration() if os.path.exists(kinesis_jar_path): 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 ) self._configure_environment() self._register_udfs() def _configure_environment(self): configuration = self.table_env.get_config().get_configuration() # Exactly-once processing configuration.set_string("execution.checkpointing.mode", "EXACTLY_ONCE") configuration.set_string( "execution.checkpointing.externalized-checkpoint-retention", "RETAIN_ON_CANCELLATION", ) # State TTL configuration configuration.set_string("table.exec.state.ttl", "36h") configuration.set_string("table.exec.state.ttl.cleanup.strategy", "delete") # Mini-batch processing for better performance 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): # Store config values in local variables to avoid serialization issues redis_host = self.config.redis_host redis_port = self.config.redis_port @udf(result_type=DataTypes.BOOLEAN()) def process_hook_comment_event( user_id: str, hook_id: str, hook_status: str, recommendation_item_id: str ) -> bool: """ Process HookComment events and store signals in Redis. Args: user_id: The ID of the user who commented hook_id: The ID of the hook being commented on hook_status: The status of the hook recommendation_item_id: Optional recommendation item ID Returns: bool: True if the event was processed successfully, False otherwise """ try: redis_client = redis.Redis( host=redis_host, port=redis_port, decode_responses=True ) current_timestamp = time.time() # Track comments created by user on hooks comments_key = f"hooks_positive_signal_comment:{user_id}" # Add hook to user's commented hooks sorted set with timestamp as score redis_client.zadd(comments_key, {hook_id: current_timestamp}) # Keep only the most recent 100 entries in the sorted set # Remove all elements except the top 100 (highest scores/most recent) redis_client.zremrangebyrank(comments_key, 0, -101) print( f"Hook comment: User {user_id} commented on hook {hook_id} (status: {hook_status}, rec_item: {recommendation_item_id})" ) return True except Exception as e: print( f"Error processing hook comment event for user {user_id}, hook {hook_id}: {e}" ) return False self.table_env.create_temporary_system_function( "process_hook_comment_event", process_hook_comment_event ) logger.info("Comment signal UDFs registered successfully") def create_kinesis_source_table(self): """Create Kinesis source table for hook comment events""" ddl = f""" CREATE TABLE hook_comment_events ( name STRING, `timestamp` STRING, source STRING, user_id STRING, session_id STRING, properties ROW< hook_id STRING, hook_status STRING, recommendation_item_id STRING >, request_id STRING ) WITH ( 'connector' = 'kinesis', 'stream.arn' = '{self.config.stream_arn}', 'aws.credentials.provider' = 'AUTO', 'aws.region' = '{aws_region}', 'source.init.position' = 'LATEST', 'format' = 'json', 'source.reader.type' = 'EFO', 'source.efo.consumer.name' = 'hooks-comment-signal-consumer-{self.config.env}', 'source.efo.lifecycle' = 'SELF_MANAGED' ) """ self.table_env.execute_sql(ddl) print("Executing DDL:") print(ddl) def run_pipeline(self): try: self.create_kinesis_source_table() job = self.process_comment_events() job.wait() except Exception as e: logger.error(f"Error in pipeline execution: {e}") raise def process_comment_events(self): # Create sink table for processing results create_sink_sql = """ CREATE TABLE hook_comment_events_sink ( user_id STRING, hook_id STRING, hook_status STRING, recommendation_item_id STRING, success BOOLEAN ) WITH ( 'connector' = 'print', 'print-identifier' = 'HOOK-COMMENT-EVENTS' ) """ self.table_env.execute_sql(create_sink_sql) # Process HookComment events comment_sql = """ INSERT INTO hook_comment_events_sink SELECT user_id, properties.hook_id, properties.hook_status, properties.recommendation_item_id, process_hook_comment_event( user_id, properties.hook_id, properties.hook_status, properties.recommendation_item_id ) as success FROM hook_comment_events WHERE name = 'HookComment' AND user_id IS NOT NULL AND properties.hook_id IS NOT NULL AND properties.hook_status IS NOT NULL """ print("Processing hook comment events...") comment_job = self.table_env.execute_sql(comment_sql) return comment_job def main(): """ Main entry point for the Hooks Comment Signal Service. This service processes real-time hook comment events from a Kinesis stream and tracks user engagement signals related to comments and comment interactions. """ try: config = SimpleConfig() logger.info("Starting Hooks Comment Signal Service...") logger.info(f"Environment: {config.env}") logger.info(f"Stream ARN: {config.stream_arn}") logger.info(f"Redis: {config.redis_host}:{config.redis_port}") service = HooksCommentSignalService(config) logger.info("Service initialized successfully") service.run_pipeline() except KeyboardInterrupt: logger.info("Service interrupted by user") except Exception as e: logger.error(f"Service failed: {e}") raise if __name__ == "__main__": main()