Processamento de Fluxo Contínuo com Estado: Detecção de Fraudes em Tempo Real usando Amazon MSK e Apache Flink

resource "aws_msk_cluster" "transaction_stream" {
cluster_name = "enterprise-transaction-stream"
kafka_version = "3.5.1"
number_of_broker_nodes = 3

broker_node_group_info {
instance_type = "kafka.m5.large"
ebs_volume_size = 1000
client_subnets = [var.subnet_az1, var.subnet_az2, var.subnet_az3]
security_groups = [aws_security_group.msk_internal.id]
}

client_authentication {
sasl {
iam = true
}
unauthenticated = false
}

encryption_info {
encryption_in_transit {
client_broker = "TLS"
in_cluster = true
}
}

tags = {
Domain = "FraudDetection"
Tier = "Ingestion"
}
}

import os
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.watermark_strategy import WatermarkStrategy
from pyflink.datastream.window import SlidingEventTimeWindows
from pyflink.common.time import Time
import json

def process_fraud_detection():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

# Configuração de conectores Kafka com autenticação IAM omitida por brevidade
properties = {
'bootstrap.servers': os.environ.get('MSK_BOOTSTRAP_SERVERS'),
'group.id': 'fraud-detector-group'
}

consumer = FlinkKafkaConsumer(
topics='raw-transactions',
deserialization_schema=SimpleStringSchema(),
properties=properties
)

stream = env.add_source(consumer)

def parse_and_extract_amount(event_str):
data = json.loads(event_str)
return (data['account_id'], float(data['amount']), data['timestamp'])

parsed_stream = stream.map(parse_and_extract_amount)

# Agrupamento e processamento de janela temporal com estado
alerts = parsed_stream \
.assign_timestamps_and_watermarks(
WatermarkStrategy.for_monotonous_timestamps()
) \
.key_by(lambda x: x[0]) \
.window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(10))) \
.reduce(lambda a, b: (a[0], a[1] + b[1], max(a[2], b[2]))) \
.filter(lambda x: x[1] > 10000.0) \
.map(lambda x: json.dumps({"account_id": x[0], "alert_reason": "High Velocity Volume", "total": x[1]}))

producer = FlinkKafkaProducer(
topic='fraud-alerts',
serialization_schema=SimpleStringSchema(),
producer_config=properties
)

alerts.add_sink(producer)
env.execute("RealTimeFraudDetection")

if __name__ == '__main__':
process_fraud_detection()

resource "aws_kinesisanalyticsv2_application" "fraud_detector_app" {
name = "FraudDetectionEngine"
runtime_environment = "FLINK-1_18"
service_execution_role = aws_iam_role.flink_execution_role.arn

application_configuration {
application_code_configuration {
code_content {
s3_content_location {
bucket_arn = aws_s3_bucket.flink_code.arn
file_key = "fraud_detection.zip"
}
}
code_content_type = "ZIPFILE"
}

environment_properties {
property_group {
property_group_id = "KafkaConfig"
property_map = {
MSK_BOOTSTRAP_SERVERS = aws_msk_cluster.transaction_stream.bootstrap_brokers_sasl_iam
}
}
}

flink_application_configuration {
checkpoint_configuration {
configuration_type = "CUSTOM"
checkpointing_enabled = true
checkpoint_interval = 60000
min_pause_between_checkpoints = 5000
}

monitoring_configuration {
configuration_type = "CUSTOM"
log_level = "INFO"
metrics_level = "TASK"
}

parallelism_configuration {
configuration_type = "CUSTOM"
parallelism = 4
parallelism_per_kpu = 1
auto_scaling_enabled = true
}
}
}
}