HOW
TO BUILD A
LAKEHOUSE

helm repo add minio https://charts.min.io/
helm repo update

helm upgrade --install minio minio/minio \
  --namespace data \
  --set mode=standalone \
  --set rootUser=minioadmin,rootPassword=minioadmin \
  --set persistence.size=500Gi \
  --set resources.requests.memory=1Gi

apiVersion: batch/v1
kind: Job
metadata:
  name: minio-bootstrap
  namespace: data
spec:
  template:
    spec:
      restartPolicy: OnFailure
      containers:
      - name: mc
        image: minio/mc:latest
        env:
        - name: MINIO_ROOT_USER
          value: minioadmin
        - name: MINIO_ROOT_PASSWORD
          value: minioadmin
        command: ["sh","-c"]
        args:
        - |
          set -e
          mc alias set local http://minio.data.svc.cluster.local:9000 \$MINIO_ROOT_USER \$MINIO_ROOT_PASSWORD
          for b in bronze silver gold airbyte iceberg mlflow; do 
            mc mb -p local/\$b || true
          done
          mc policy set public local/bronze || true

helm repo add trino https://trinodb.github.io/charts/
helm upgrade --install trino trino/trino -n data \
  --set server.workers=2 \
  --set service.type=ClusterIP

apiVersion: v1
kind: ConfigMap
metadata:
  name: trino-catalogs
  namespace: data
  labels: { app: trino }
data:
  iceberg.properties: |
    connector.name=iceberg
    catalog.warehouse=s3://iceberg
    iceberg.catalog.type=nessie
    iceberg.nessie.uri=http://nessie.data.svc.cluster.local:19120/api/v2
    fs.native-s3.enabled=true
    s3.endpoint=http://minio.data.svc.cluster.local:9000
    s3.path-style-access=true
    s3.aws-access-key=minioadmin
    s3.aws-secret-key=minioadmin
  clickhouse.properties: |
    connector.name=clickhouse
    clickhouse.http-port=8123
    connection-url=jdbc:clickhouse://clickhouse-clickhouse.data.svc.cluster.local:8123
  postgres.properties: |
    connector.name=postgresql
    connection-url=jdbc:postgresql://postgres-postgresql.data.svc.cluster.local:5432/postgres
    connection-user=postgres
    connection-password=postgres

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  name: bronze-to-silver-iceberg
  namespace: data
spec:
  entrypoint: etl
  templates:
  - name: etl
    steps:
    - - name: bronze-to-silver
        template: spark-sql
  - name: spark-sql
    container:
      image: apache/spark:3.5.1
      command: ["/bin/sh","-lc"]
      args:
      - |
        spark-sql \
          --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,org.apache.hadoop:hadoop-aws:3.3.4,software.amazon.awssdk:bundle:2.20.156 \
          --conf spark.sql.catalog.iceberg=org.apache.iceberg.spark.SparkCatalog \
          --conf spark.sql.catalog.iceberg.catalog-impl=org.apache.iceberg.nessie.NessieCatalog \
          --conf spark.sql.catalog.iceberg.uri=http://nessie.data.svc.cluster.local:19120/api/v2 \
          --conf spark.sql.catalog.iceberg.warehouse=s3a://iceberg \
          --conf spark.sql.catalog.iceberg.ref=main \
          --conf spark.hadoop.fs.s3a.endpoint=http://minio.data.svc.cluster.local:9000 \
          --conf spark.hadoop.fs.s3a.path.style.access=true \
          --conf spark.hadoop.fs.s3a.access.key=minioadmin \
          --conf spark.hadoop.fs.s3a.secret.key=minioadmin <<'SQL'
        CREATE SCHEMA IF NOT EXISTS iceberg.silver;
        CREATE TABLE IF NOT EXISTS iceberg.silver.nyc_trips (
          vendor_id string,
          tpep_pickup_timestamp timestamp,
          tpep_dropoff_timestamp timestamp,
          passenger_count int,
          trip_distance double,
          fare_amount double
        )
        USING iceberg
        PARTITIONED BY (days(tpep_pickup_timestamp));

        CREATE OR REPLACE TEMPORARY VIEW bronze_csv
        USING csv
        OPTIONS (
          paths 's3a://bronze/nyc_trips/*.csv',
          header 'true',
          inferSchema 'true'
        );

        INSERT INTO iceberg.silver.nyc_trips
        SELECT vendor_id,
               cast(tpep_pickup_timestamp as timestamp) as tpep_pickup_timestamp,
               cast(tpep_dropoff_timestamp as timestamp) as tpep_dropoff_timestamp,
               cast(passenger_count as int) as passenger_count,
               cast(trip_distance as double) as trip_distance,
               cast(fare_amount as double) as fare_amount
        FROM bronze_csv;
SQL

apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm
  namespace: ml
spec:
  replicas: 1
  selector:
    matchLabels: { app: vllm }
  template:
    metadata:
      labels: { app: vllm }
    spec:
      containers:
      - name: vllm
        image: vllm/vllm-openai:latest
        args:
        - "--model=/models/Qwen2.5-7B-Instruct"
        - "--max-num-seqs=64"
        - "--dtype=float16"
        ports:
        - containerPort: 8000
        resources:
          limits:
            nvidia.com/gpu: 1
        volumeMounts:
        - name: models
          mountPath: /models
      volumes:
      - name: models
        persistentVolumeClaim:
          claimName: models-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: vllm
  namespace: ml
spec:
  selector: { app: vllm }
  ports:
  - port: 8000
    targetPort: 8000

-- Connect to Trino
trino --server trino.data.svc.cluster.local:8080 \
      --catalog iceberg --schema silver

-- Query partitioned data with time travel
SELECT date(tpep_pickup_timestamp) d, count(*) 
FROM nyc_trips 
WHERE tpep_pickup_timestamp >= TIMESTAMP '2024-01-01'
GROUP BY 1 
ORDER BY 1 DESC 
LIMIT 10;

-- Create gold aggregation
CREATE SCHEMA IF NOT EXISTS iceberg.gold;
CREATE TABLE IF NOT EXISTS iceberg.gold.daily_trips AS
SELECT date(tpep_pickup_timestamp) d, count(*) trips
FROM iceberg.silver.nyc_trips
GROUP BY 1;

curl -s https://llm.example.local/v1/chat/completions \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "model":"Qwen2.5-7B-Instruct",
    "messages":[{"role":"user","content":"Summarize yesterday'''s lakehouse loads."}],
    "temperature":0.2
  }'