Powering Real-Time AI with SQL: Declarative Feature Engineering & Inference

In the evolution of modern data architectures, two massive waves are currently colliding: real-time stream processing and artificial intelligence (AI).

Historically, these two domains lived in separate silos. AI models were trained on historical data in batches and served via microservices, while streaming pipelines focused on moving and transforming telemetry data at scale. But today, applications like real-time fraud detection, dynamic pricing, and autonomous user engagement require models to act on information the sub-second it is generated.

However, a major bottleneck remains: the feature gap. AI models are inherently stateless and expect structured features, whereas raw streaming data is typically sparse and nested. To bridge this gap, teams need a fast, declarative way to perform real-time feature engineering.

The solution? Flink SQL inside Flinkflow.

🛑 The Real-Time AI Bottleneck: The Feature Gap

Integrating live data streams with AI models isn’t as simple as making an API call inside a map function. Raw events are rarely descriptive enough for a model to make a smart prediction.

For example, if you feed a model a raw click event:

{"userId": "usr_101", "productId": "prod_889", "action": "click", "timestamp": "2026-06-20T12:00:00Z"}

The model has no context. To determine whether the user is highly engaged or likely to purchase, the model needs features:

• How many times has this user clicked in the last 5 minutes?
• What is their most-viewed product category today?
• How does their click volume compare to their historical average?

Calculating these rolling windows, slide intervals, and stateful aggregations on massive streams at sub-second latencies requires complex, low-level streaming code (Java/Scala) or heavy, slow batch feature stores.

📊 Why SQL is the Perfect Engine for Real-Time Features

SQL is the native language of data manipulation. When it comes to real-time feature engineering, SQL excels where other paradigms struggle:

1. Declarative Windowing: SQL makes temporal aggregations (TUMBLE, HOP, and SESSION windows) simple and human-readable.
2. Stream-Table Joins: You can easily join high-velocity clickstreams with lookup tables (such as user profiles and product catalogs) to add demographic or pricing context.
3. Familiarity: Data scientists, analysts, and ML engineers already write features in SQL. Using SQL in production means no translation layers or language barriers between research and deployment.

🆕 Flink SQL + Flinkflow: Preparation Meets Inference

With Flinkflow’s polyglot pipeline model, you no longer have to choose between a pure SQL setup and a pure Python environment. Flinkflow embeds Flink SQL as a first-class type: sql step inside its declarative YAML DSL.

This enables a clean, optimized separation of concerns:

• The SQL Step handles high-throughput filtering, windowing, and rolling feature calculations.
• The Agent/Python Step consumes those processed feature vectors and executes model inference (e.g., using GPT-4o, Claude, or a local PyTorch model).

🏗️ Hands-On: Real-Time Personalization Pipeline

Let’s look at a concrete example. We want to ingest a raw stream of user click events, calculate click statistics over a sliding 5-minute window using Flink SQL, and pass these features to an AI agent to generate a personalized product recommendation.

Here is how easily this is declared in a single Flinkflow pipeline:

name: "Real-time AI Recommendation Engine"
parallelism: 2

steps:
  # Step 1: Ingest raw clickstream events from Kafka
  - type: source
    name: user-clicks
    properties:
      topic: "analytics.clicks"
      bootstrapServers: "kafka:9092"

  # Step 2: Use Flink SQL to calculate rolling features (Category Affinity)
  - type: sql
    name: user-feature-generator
    properties:
      schema.userId: "string"
      schema.category: "string"
      schema.clickTime: "timestamp"
      watermark.column: "clickTime"
      watermark.delay: "5"
      query: |
        SELECT 
          userId,
          category,
          COUNT(*) AS category_click_count_5m
        FROM TABLE(TUMBLE(TABLE input, DESCRIPTOR(clickTime), INTERVAL '5' MINUTE))
        GROUP BY userId, category, window_start, window_end

  # Step 3: Pass SQL features to LLM for a personalized recommendation
  - type: agent
    name: recommendation-agent
    inputs: [user-feature-generator]
    properties:
      provider: "openai"
      model: "gpt-4o"
      apiKey: "${OPENAI_API_KEY}"
      temperature: 0.2
      prompt: |
        Based on the user's real-time browsing behavior:
        User ID: {{userId}}
        Top Category Browsed: {{category}}
        Number of Clicks in Category (Last 5m): {{category_click_count_5m}}

        Generate a personalized product recommendation copy (max 1 sentence) tailored to this specific category.

  # Step 4: Output recommendations to a push notification service
  - type: sink
    name: push-notifications-kafka
    properties:
      topic: "notifications.recommendations"
      bootstrapServers: "kafka:9092"

🛠️ The Architecture Under the Hood

1. In-Memory Aggregations via RocksDB

Flink SQL uses Flink’s native State Backends (like RocksDB) to track the state of the 5-minute sliding windows. The window data is cached locally on Flink’s task managers. When the window closes, Flink emits the feature set instantly, avoiding external database lookups.

2. Async wait for Inference

AI inference—especially via external LLMs—takes time. Flinkflow prevents network calls from causing backpressure by using Async I/O. It executes concurrent LLM requests asynchronously, ensuring the pipeline maintains high-throughput event processing even when API latency spikes.

3. Load-Time Schema Validation

Flinkflow validates your SQL step schema, watermark column types, and input fields before running the pipeline. This ensures that the outputs of your SQL features match the expectations of your downstream model prompts.

🚀 Common Use Cases

• Financial Fraud Detection: Aggregate transaction counts and amounts over a sliding 10-minute window with SQL, then run an XGBoost, ONNX Runtime, or Flink ML model step to block suspicious accounts.
• Predictive Maintenance: Calculate temperature and vibration averages over sliding device windows with SQL, then trigger anomaly detection models to alert operators.
• Dynamic E-Commerce Pricing: Join product inventory, viewer demand, and competitor pricing tables in SQL to recalculate optimal prices in real-time.

🏁 Get Started with Flinkflow

Ready to power your AI models with real-time SQL feature stores?

• Explore the Code: Head over to the Flinkflow GitHub Repository.
• Read the Docs: Read our complete guide to joins, lookups, and SQL steps.
• Discuss: Connect with our engineers on Zulip to discuss stream enrichment patterns!