Stream Processing

1
2

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9249

1
2
3
4

scrape_configs:
  - job_name: 'flink'
    static_configs:
      - targets: ['jobmanager:9249', 'taskmanager:9249']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

loki:
  image: grafana/loki:latest
  ports:
    - "3100:3100"

promtail:
  image: grafana/promtail:latest
  volumes:
    - /var/log:/var/log
    - ./promtail-config.yaml:/etc/promtail/promtail.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

server:
  http_listen_port: 9080

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: flink_logs
    static_configs:
      - targets:
          - localhost
        labels:
          job: flink
          __path__: /flink/log/*.log

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Jaeger 설정
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(agent_host_name="jaeger", agent_port=6831)
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(jaeger_exporter))

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("process_event"):
    # Flink 내부 연산 로직 처리
    process_event()

1
2
3
4
5

jaeger:
  image: jaegertracing/all-in-one:latest
  ports:
    - "16686:16686"  # UI
    - "6831:6831/udp" # Agent Port

1
2
3
4
5
6
7
8

alerting:
  alertmanagers:
    - static_configs:
        - targets:
            - 'alertmanager:9093'

rule_files:
  - 'alert_rules.yml'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

groups:
  - name: flink_alerts
    rules:
      - alert: HighLatency
        expr: histogram_quantile(0.95, rate(taskmanager_job_task_operator_latency_ms_bucket[5m])) > 500
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "High latency detected in Flink pipeline"

1
2
3
4
5
6

alertmanager:
  image: prom/alertmanager
  ports:
    - "9093:9093"
  volumes:
    - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

apiVersion: apps/v1
kind: Deployment
metadata:
  name: flink-jobmanager
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink-jobmanager
  template:
    metadata:
      labels:
        app: flink-jobmanager
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9249"
    spec:
      containers:
        - name: jobmanager
          image: flink:1.17
          ports:
            - containerPort: 8081
            - containerPort: 9249
          env:
            - name: JOB_MANAGER_RPC_ADDRESS
              value: flink-jobmanager
          volumeMounts:
            - name: config-volume
              mountPath: /opt/flink/conf
      volumes:
        - name: config-volume
          configMap:
            name: flink-config
---
apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
spec:
  selector:
    app: flink-jobmanager
  ports:
    - name: web
      port: 8081
    - name: metrics
      port: 9249

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: flink-servicemonitor
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app: flink-jobmanager
  endpoints:
    - port: metrics
      interval: 15s
  namespaceSelector:
    matchNames:
      - default

1

helm install kube-prometheus-stack prometheus-community/kube-prometheus-stack

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

resource = Resource(attributes={
    "service.name": "flink-stream-job",
    "service.namespace": "streaming"
})

provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

otlp_exporter = OTLPSpanExporter(endpoint="otel-collector:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(otlp_exporter))

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("window_processing"):
    # Flink 연산자 또는 사용자 정의 함수 처리
    process_event()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

######  OpenTelemetry Collector 구성 (OTEL → Jaeger)

**Collector ConfigMap**

```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
data:
  collector-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
          http:

    exporters:
      jaeger:
        endpoint: jaeger-collector.default.svc.cluster.local:14250
        tls:
          insecure: true

    service:
      pipelines:
        traces:
          receivers: [otlp]
          exporters: [jaeger]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-collector
spec:
  replicas: 1
  selector:
    matchLabels:
      app: otel-collector
  template:
    metadata:
      labels:
        app: otel-collector
    spec:
      containers:
        - name: otel-collector
          image: otel/opentelemetry-collector:latest
          args: ["--config=/conf/collector-config.yaml"]
          ports:
            - containerPort: 4317
            - containerPort: 55681
          volumeMounts:
            - name: config
              mountPath: /conf
      volumes:
        - name: config
          configMap:
            name: otel-collector-config

1
2

helm repo add open-telemetry https://open-telemetry.github.io/opentelemetry-helm-charts
helm install otel open-telemetry/opentelemetry-collector

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: promtail
spec:
  selector:
    matchLabels:
      name: promtail
  template:
    metadata:
      labels:
        name: promtail
    spec:
      containers:
        - name: promtail
          image: grafana/promtail:2.9.2
          args:
            - -config.file=/etc/promtail/promtail.yaml
          volumeMounts:
            - name: logs
              mountPath: /var/log
            - name: config
              mountPath: /etc/promtail
      volumes:
        - name: logs
          hostPath:
            path: /var/log
        - name: config
          configMap:
            name: promtail-config

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

apiVersion: v1
kind: ConfigMap
metadata:
  name: promtail-config
data:
  promtail.yaml: |
    server:
      http_listen_port: 9080
    positions:
      filename: /tmp/positions.yaml
    clients:
      - url: http://loki:3100/loki/api/v1/push
    scrape_configs:
      - job_name: flink_logs
        static_configs:
          - targets:
              - localhost
            labels:
              job: flink
              __path__: /flink/log/*.log

1
2

helm repo add grafana https://grafana.github.io/helm-charts
helm install loki grafana/loki-stack

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

apiVersion: v1
kind: ConfigMap
metadata:
  name: alertmanager-config
data:
  alertmanager.yml: |
    global:
      resolve_timeout: 5m
    route:
      receiver: slack-notifications
    receivers:
      - name: slack-notifications
        slack_configs:
          - api_url: https://hooks.slack.com/services/XXXXXX
            channel: "#flink-alerts"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer
import json
from datetime import datetime
import logging

class FraudDetectionProcessor:
    """실시간 사기 탐지 스트림 처리기"""
    
    def __init__(self):
        # Flink 실행 환경 설정
        self.env = StreamExecutionEnvironment.get_execution_environment()
        self.env.set_parallelism(4)  # 병렬 처리 수 설정
        
        # 체크포인팅 활성화 (장애 복구용)
        self.env.enable_checkpointing(10000)  # 10초마다 체크포인트
        
        # 테이블 환경 설정
        self.table_env = StreamTableEnvironment.create(self.env)
        
        # 사기 탐지 규칙 저장소 (상태 관리)
        self.fraud_patterns = {}
        
    def create_kafka_source(self):
        """카프카에서 거래 이벤트 스트림 생성"""
        properties = {
            'bootstrap.servers': 'localhost:9092',
            'group.id': 'fraud-detection-group',
            'auto.offset.reset': 'latest'
        }
        
        # 카프카 컨슈머 생성 (거래 이벤트 토픽에서 데이터 수신)
        kafka_consumer = FlinkKafkaConsumer(
            topics='transaction-events',
            deserialization_schema=SimpleStringSchema(),
            properties=properties
        )
        
        return self.env.add_source(kafka_consumer)
    
    def parse_transaction_event(self, event_json):
        """거래 이벤트 JSON 파싱 및 구조화"""
        try:
            event = json.loads(event_json)
            return {
                'transaction_id': event.get('transaction_id'),
                'user_id': event.get('user_id'),
                'amount': float(event.get('amount', 0)),
                'location': event.get('location'),
                'timestamp': datetime.fromisoformat(event.get('timestamp')),
                'card_number': event.get('card_number'),
                'merchant_category': event.get('merchant_category')
            }
        except Exception as e:
            logging.error(f"이벤트 파싱 오류: {e}")
            return None
    
    def detect_velocity_fraud(self, transaction):
        """거래 빈도 기반 사기 탐지"""
        user_id = transaction['user_id']
        current_time = transaction['timestamp']
        
        # 사용자별 최근 거래 이력 관리 (상태 저장)
        if user_id not in self.fraud_patterns:
            self.fraud_patterns[user_id] = []
        
        user_transactions = self.fraud_patterns[user_id]
        
        # 최근 5분 내 거래 수 계산
        recent_transactions = [
            t for t in user_transactions 
            if (current_time - t['timestamp']).total_seconds() < 300
        ]
        
        # 5분 내 5회 이상 거래 시 의심
        if len(recent_transactions) >= 5:
            return True, "높은 거래 빈도 감지"
        
        # 현재 거래를 이력에 추가
        user_transactions.append(transaction)
        
        # 메모리 최적화: 1시간 이상 된 거래 제거
        self.fraud_patterns[user_id] = [
            t for t in user_transactions 
            if (current_time - t['timestamp']).total_seconds() < 3600
        ]
        
        return False, "정상"
    
    def detect_location_fraud(self, transaction):
        """위치 기반 사기 탐지"""
        user_id = transaction['user_id']
        current_location = transaction['location']
        current_time = transaction['timestamp']
        
        if user_id not in self.fraud_patterns:
            return False, "신규 사용자"
        
        user_transactions = self.fraud_patterns[user_id]
        
        # 최근 1시간 내 거래 중 다른 위치에서의 거래 확인
        for prev_transaction in user_transactions:
            time_diff = (current_time - prev_transaction['timestamp']).total_seconds()
            
            # 1시간 내 거래 중 위치가 다른 경우
            if time_diff < 3600 and prev_transaction['location'] != current_location:
                # 물리적으로 불가능한 이동 시간 계산 (간단한 예시)
                if self.calculate_travel_time(prev_transaction['location'], current_location) > time_diff:
                    return True, f"불가능한 위치 이동: {prev_transaction['location']} -> {current_location}"
        
        return False, "정상"
    
    def calculate_travel_time(self, location1, location2):
        """두 위치 간 최소 이동 시간 계산 (간단한 예시)"""
        # 실제로는 지리적 거리와 교통수단을 고려한 복잡한 계산 필요
        if location1 != location2:
            return 1800  # 30분 최소 이동 시간 가정
        return 0
    
    def detect_amount_fraud(self, transaction):
        """거래 금액 기반 사기 탐지"""
        user_id = transaction['user_id']
        amount = transaction['amount']
        
        # 사용자별 평균 거래 금액 계산
        if user_id not in self.fraud_patterns:
            return False, "신규 사용자"
        
        user_transactions = self.fraud_patterns[user_id]
        
        if len(user_transactions) < 5:
            return False, "거래 이력 부족"
        
        # 최근 30일 평균 거래 금액 계산
        recent_amounts = [t['amount'] for t in user_transactions[-30:]]
        avg_amount = sum(recent_amounts) / len(recent_amounts)
        
        # 평균의 5배 이상이면 의심
        if amount > avg_amount * 5:
            return True, f"비정상적 고액 거래: {amount} (평균: {avg_amount:f})"
        
        return False, "정상"
    
    def calculate_risk_score(self, fraud_results):
        """여러 탐지 결과를 종합하여 위험 점수 계산"""
        risk_score = 0
        risk_factors = []
        
        for is_fraud, reason in fraud_results:
            if is_fraud:
                if "빈도" in reason:
                    risk_score += 40
                elif "위치" in reason:
                    risk_score += 35
                elif "고액" in reason:
                    risk_score += 25
                risk_factors.append(reason)
        
        return min(risk_score, 100), risk_factors
    
    def process_transaction(self, transaction_event):
        """거래 이벤트 처리 메인 로직"""
        transaction = self.parse_transaction_event(transaction_event)
        
        if not transaction:
            return None
        
        # 다양한 사기 탐지 규칙 적용
        fraud_checks = [
            self.detect_velocity_fraud(transaction),
            self.detect_location_fraud(transaction),
            self.detect_amount_fraud(transaction)
        ]
        
        # 위험 점수 계산
        risk_score, risk_factors = self.calculate_risk_score(fraud_checks)
        
        # 결과 생성
        result = {
            'transaction_id': transaction['transaction_id'],
            'user_id': transaction['user_id'],
            'risk_score': risk_score,
            'risk_factors': risk_factors,
            'timestamp': transaction['timestamp'].isoformat(),
            'action': 'BLOCK' if risk_score > 70 else 'ALLOW'
        }
        
        return json.dumps(result)
    
    def create_kafka_sink(self):
        """처리 결과를 카프카로 전송"""
        properties = {
            'bootstrap.servers': 'localhost:9092'
        }
        
        # 사기 탐지 결과를 별도 토픽으로 전송
        kafka_producer = FlinkKafkaProducer(
            topic='fraud-detection-results',
            serialization_schema=SimpleStringSchema(),
            producer_config=properties
        )
        
        return kafka_producer
    
    def run(self):
        """스트림 처리 파이프라인 실행"""
        try:
            # 데이터 소스 생성
            transaction_stream = self.create_kafka_source()
            
            # 거래 이벤트 처리 및 사기 탐지
            fraud_detection_stream = transaction_stream.map(
                self.process_transaction,
                output_type=Types.STRING()
            ).filter(lambda x: x is not None)  # None 결과 제외
            
            # 결과를 카프카로 전송
            fraud_detection_stream.add_sink(self.create_kafka_sink())
            
            # 실행 시작
            self.env.execute("Real-time Fraud Detection")
            
        except Exception as e:
            logging.error(f"스트림 처리 오류: {e}")
            raise

# 사용 예시
if __name__ == "__main__":
    # 로깅 설정
    logging.basicConfig(level=logging.INFO)
    
    # 사기 탐지 프로세서 생성 및 실행
    fraud_detector = FraudDetectionProcessor()
    
    print("실시간 사기 탐지 시스템 시작…")
    fraud_detector.run()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# Faust 기반 실시간 거래 이상 탐지 예시
import faust

# 거래 데이터 모델 정의
class Transaction(faust.Record):
    user_id: str
    amount: float
    timestamp: float

app = faust.App('fraud-detection', broker='kafka://localhost:9092')
transactions = app.topic('transactions', value_type=Transaction)

# 이상 거래 탐지 함수
@app.agent(transactions)
async def detect_fraud(transactions):
    async for tx in transactions:
        # 예시: 1000 이상 거래는 이상 거래로 간주
        if tx.amount > 1000:
            print(f"이상 거래 감지: {tx.user_id}, 금액: {tx.amount}")
            # 알림, DB 저장 등 추가 처리 가능

if __name__ == '__main__':
    app.main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# PyFlink 예시: Kafka에서 실시간 거래 데이터 수신 후 이상 거래 탐지
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors import FlinkKafkaConsumer

# 1. 실행 환경 설정
env = StreamExecutionEnvironment.get_execution_environment()

# 2. Kafka에서 데이터 읽기
kafka_props = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'fraud-detection-group'
}

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

# 3. 스트림 데이터 수신
stream = env.add_source(consumer)

# 4. 거래 데이터를 기준으로 필터링 - 예: 1000달러 이상이면 의심 거래
suspicious = stream.filter(lambda tx: float(tx.split(",")[2]) > 1000)

# 5. 콘솔 출력 (또는 다른 Sink 사용 가능)
suspicious.print()

# 6. 실행
env.execute("Fraud Detection Stream Job")