siqueiraa/kaflowsql repository overview

# That's it. Your entire streaming pipeline.
name: user_personalization
query: |
  SELECT 
    a.user_id,
    a.action,
    a.timestamp,
    p.subscription_tier,
    p.full_name
  FROM user_actions a
  LEFT JOIN customer_profiles p ON a.user_id = p.user_id

output:
  topic: personalized_actions
  format: avro
  key: user_id

name: fraud_detection_prep
query: |
  SELECT 
    t.transaction_id,
    t.amount,
    u.risk_score,
    u.account_age_days,
    m.merchant_category,
    m.country
  FROM transactions t
  LEFT JOIN users u ON t.user_id = u.user_id
  LEFT JOIN merchants m ON t.merchant_id = m.merchant_id
  WHERE t.amount > 100

output:
  topic: fraud_detection_input
  format: json
  key: transaction_id

emission:
  type: smart
  require: [risk_score, merchant_category]

name: personalization_engine
query: |
  SELECT 
    c.user_id,
    c.page_view,
    c.timestamp,
    u.subscription_tier,
    p.category,
    p.price,
    h.last_purchase_date
  FROM clickstreams c
  LEFT JOIN user_profiles u ON c.user_id = u.user_id
  LEFT JOIN products p ON c.product_id = p.product_id
  LEFT JOIN purchase_history h ON c.user_id = h.user_id

output:
  topic: personalized_recommendations
  format: avro
  key: user_id

state:
  events:
    default_ttl: 1h
    overrides:
      clickstreams: 30m        # High-volume, short retention
      purchase_history: 24h    # Purchase patterns change slowly
  dimensions:
    - user_profiles            # User demographics rarely change
    - products                # Product catalog is mostly static

name: sensor_enrichment
query: |
  SELECT 
    s.device_id,
    s.sensor_value,
    s.timestamp,
    d.location,
    d.device_type,
    c.threshold_config
  FROM sensor_readings s
  LEFT JOIN devices d ON s.device_id = d.device_id
  LEFT JOIN device_config c ON s.device_id = c.device_id

output:
  topic: enriched_sensor_data
  format: avro
  key: device_id

emission:
  type: immediate  # High throughput, partial data acceptable
  
state:
  events:
    default_ttl: 5m  # High-volume, short retention
  dimensions:
    - devices
    - device_config

# Create your pipeline
cat > pipelines/my_pipeline.yaml << EOF
name: user_enrichment
query: |
  SELECT 
    e.user_id,
    e.event_type,
    e.timestamp,
    p.subscription_tier,
    p.country
  FROM user_events e
  LEFT JOIN user_profiles p ON e.user_id = p.user_id

output:
  topic: enriched_events
  format: json
  key: user_id

emission:
  type: smart
  require: [subscription_tier]

state:
  events:
    default_ttl: 1h          # Business-appropriate retention
    overrides:
      user_events: 2h        # Longer for user journey tracking  
  dimensions:
    - user_profiles          # Static reference data never expires
EOF

# Configure Kafka connection
cat > config.yaml << EOF
kafka:
  brokers: ["your-kafka:9092"]
  schemaRegistry: "http://your-registry:8081"
  useAvro: true

state:
  rocksdb:
    path: /data/rocksdb
EOF

# Run KaflowSQL
docker run --rm \
  -v $(pwd)/pipelines:/app/pipelines:ro \
  -v $(pwd)/config.yaml:/app/config.yaml:ro \
  siqueiraa/kaflowsql:latest

# Download latest release
wget https://github.com/siqueiraa/KaflowSQL/releases/latest/download/kaflowsql-linux-amd64.tar.gz
tar -xzf kaflowsql-linux-amd64.tar.gz

# Configure and run
./engine config.yaml pipelines/

name: customer_360_windowed
window: 24h

query: |
  SELECT 
    e.user_id,
    e.event_type,
    e.timestamp,
    p.email,
    p.tier,
    t.last_purchase_amount,
    s.support_tickets
  FROM events e
  LEFT JOIN profiles p ON e.user_id = p.user_id
  LEFT JOIN transactions t ON e.user_id = t.user_id  
  LEFT JOIN support s ON e.user_id = s.user_id

state:
  events:
    default_ttl: 1h           # Most events are short-lived
    overrides:
      events: 2h              # User interactions need longer correlation
      transactions: 24h       # Financial data patterns evolve slowly
      support: 7d             # Support tickets have long resolution cycles
  dimensions:
    - profiles                # User profiles are slow-changing reference data

emission:
  type: smart
  require: [email, tier]      # Wait for essential profile data

# Required: Pipeline identification
name: my_pipeline
window: 1h                     # Optional: Global TTL override

# Required: SQL query defining the joins
query: |
  SELECT columns...
  FROM topic1 alias1
  LEFT JOIN topic2 alias2 ON alias1.key = alias2.key
  WHERE conditions...

# Required: Output configuration
output:
  topic: output_topic_name     # Target Kafka topic
  format: json                 # json | avro
  key: field_name             # Kafka message key field
  schema:                     # Optional: Avro schema definition
    name: MySchema
    fields: [...]

# Optional: Emission strategy
emission:
  type: smart                 # smart | immediate
  require:                    # Fields required before emission
    - field1
    - field2

# Optional: State management
state:
  events:
    default_ttl: 1h          # Default TTL for event topics
    overrides:               # Per-topic TTL overrides
      topic_name: 24h
  dimensions:                # Slow-changing dimension topics (no TTL)
    - lookup_topic1
    - lookup_topic2

kafka:
  brokers: ["broker1:9092", "broker2:9092"]
  schemaRegistry: "http://registry:8081"
  useAvro: true
  consumer:
    group: "kaflowsql"
    autoOffsetReset: "earliest"

state:
  ttl: 1h
  rocksdb:
    path: "/data/rocksdb"
    checkpoint:
      enabled: true
      interval: 5m
      s3:
        enabled: true
        bucket: "my-kaflowsql-state"
        region: "us-east-1"
        accessKey: "your-access-key"
        secretKey: "your-secret-key"

logging:
  level: "info"
  format: "json"

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kaflowsql
spec:
  replicas: 3
  selector:
    matchLabels:
      app: kaflowsql
  template:
    metadata:
      labels:
        app: kaflowsql
    spec:
      containers:
      - name: kaflowsql
        image: siqueiraa/kaflowsql:latest
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        volumeMounts:
        - name: pipelines
          mountPath: /app/pipelines
        - name: config
          mountPath: /app/config.yaml
          subPath: config.yaml
        - name: data
          mountPath: /data
      volumes:
      - name: pipelines
        configMap:
          name: kaflowsql-pipelines
      - name: config
        configMap:
          name: kaflowsql-config
      - name: data
        persistentVolumeClaim:
          claimName: kaflowsql-data