Master

mlflow/Dockerfile

@@ -1,12 +1,30 @@
-FROM python:3.10.16-slim-bookworm
-WORKDIR /app
-COPY requirements.txt /app/requirements.txt
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libpq-dev \
+  && rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache-dir \
+    mlflow==2.14.1 \
+    psycopg2-binary==2.9.9 \
+    gunicorn==21.2.0
+
+ENV BACKEND_STORE_URI="postgresql+psycopg2://admin:admin@postgres:5432/mlflow" \
+    ARTIFACT_ROOT="/mlflow/artifacts"
+
+EXPOSE 5000
+
+VOLUME ["/mlflow/artifacts"]
+
+CMD mlflow server \
+    --host 0.0.0.0 \
+    --port 5000 \
+    --backend-store-uri "$BACKEND_STORE_URI" \
+    --artifacts-destination "$ARTIFACT_ROOT"
 
-# Install dependencies
-RUN pip install -r /app/requirements.txt
 
-# Copy the rest of the application
-COPY . .
 
-# Set the command to run your application
-EXPOSE 5500

mlflow/requirements.txt

@@ -1,4 +1,3 @@
-cryptography
-boto3
-mlflow
-psycopg2-binary
+mlflow==2.14.1
+psycopg2-binary==2.9.9
+gunicorn==21.2.0

producer/Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+RUN pip install --no-cache-dir kafka-python==2.0.2 numpy==1.26.4 faker==24.9.0 python-dateutil==2.9.0.post0
+
+COPY services/producer/app /app/app
+
+ENV KAFKA_BOOTSTRAP_SERVERS=kafka:9092 \
+    KAFKA_TOPIC=transactions \
+    PRODUCE_RATE_PER_SEC=2
+
+CMD python -u app/producer.py
+
+
+

producer/main.py

@@ -0,0 +1,51 @@
+import json
+import os
+import random
+import time
+from datetime import datetime
+
+from dateutil.tz import tzutc
+from faker import Faker
+from kafka import KafkaProducer
+KAFKA_BOOTSTRAP_SERVERS=os.getenv("KAFKA_BOOTSTRAP_SERVERS","kafka:9092")
+KAFKA_TOPIC=os.getenv("KAFKA_TOPIC", "transactions")
+RATE=float(os.getenv("PRODUCE_RATE_PER_SEC", "2"))
+
+fake = Faker()
+
+def generate_transaction()->dict:
+    amount = round(random.lognormvariate(3, 1), 2)  # skewed positive amounts
+
+    # add some injected fraudulent patterns
+    if random.random()<0.05:
+        amount*=random.uniform(5,20)
+
+    return {
+        "transaction_id": fake.uuid4(),
+        "amount": float(amount),
+        "features":{
+            "num_items":int(max(1,random.gauss(2,1))),
+            "merchant_risk": float(random.random()),
+            "hour": float(datetime.now(tz=tzutc()).hour),
+        },
+        "event_time": datetime.utcnow().isoformat()+'Z',
+    }
+
+def main()->None:
+    producer=None
+    while producer is None:
+        try:
+            producer=KafkaProducer(
+                bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
+                value_serializer=lambda v: json.dumps(v).encode('utf-8'),
+                linger_ms=50,
+            )
+        except Exception:
+            time.sleep(2)
+    interval = 1.0 / RATE if RATE > 0 else 0.5
+    while True:
+        payload=generate_transaction()
+        producer.send(KAFKA_TOPIC,value=payload)
+        time.sleep(interval)
+if __name__=="__main__":
+    main()