Build a Real-Time E-Commerce Analytics API from Kafka - II

In the previous blog post, we built a complete real-time metrics API from Kafka in 15 minutes. We covered:

• Connecting to Kafka and ingesting order events
• Enriching data with dimension tables and PostgreSQL
• Creating Materialized Views for fast aggregations
• Building multiple API endpoints for real-time metrics

Now let's extend that foundation with production ready features: exporting data back to Kafka, connecting to BI tools, comprehensive monitoring and optimization strategies.

What We'll Cover

In this Part 2 guide, we'll add:

1. Export to Kafka - Send processed data back to Kafka for event driven architectures
2. BI Tool Integration - Connect to Tableau, Power BI, Grafana and more
3. Monitoring and Optimization - Track performance, identify bottlenecks and optimize
4. Schema Evolution and Branching - Safely evolve schemas using branches and FORWARD_QUERY
5. Common Patterns and Extensions - Real-world patterns for scaling your pipeline

Prerequisites

This guide assumes you've completed Part 1 and have:

• A working Kafka connector ingesting order events
• Enriched orders Materialized View
• Multiple API endpoints deployed
• Access to your Tinybird workspace

Part 1: Export Data from Your Analytics API Back to Kafka

Exporting processed data back to Kafka enables event driven architectures. Other services can consume your analytics data for downstream processing.

Common use cases include:

• Triggering downstream processes based on metrics
• Feeding data to other analytics systems
• Creating event streams for real-time dashboards
• Integrating with microservices that need aggregated data

Create a Kafka Sink Connection

First, create a connection for exporting to Kafka. You can reuse your existing Kafka connection or create a new one for a different cluster:

tb connection create kafka

When prompted, provide:

• Connection name: kafka_sink (or reuse kafka_ecommerce)
• Bootstrap servers: Your Kafka broker addresses
• Authentication: Your Kafka credentials

Or create the connection file manually:

TYPE kafka
KAFKA_BOOTSTRAP_SERVERS pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
KAFKA_SECURITY_PROTOCOL SASL_SSL
KAFKA_SASL_MECHANISM PLAIN
KAFKA_KEY {{ tb_secret("KAFKA_KEY") }}
KAFKA_SECRET {{ tb_secret("KAFKA_SECRET") }}

Export Hourly Revenue Summaries

Let's create a Sink that exports hourly revenue summaries to Kafka:

NODE hourly_revenue_summary
SQL >
    SELECT
        toStartOfHour(timestamp) as hour,
        product_category,
        customer_country,
        sum(order_total) as total_revenue,
        count() as order_count,
        count(DISTINCT customer_id) as unique_customers,
        avg(order_total) as avg_order_value
    FROM enriched_orders
    WHERE timestamp >= now() - INTERVAL 1 HOUR
      AND timestamp < now()
    GROUP BY hour, product_category, customer_country

TYPE sink
EXPORT_CONNECTION_NAME kafka_sink
EXPORT_KAFKA_TOPIC revenue_summaries
EXPORT_SCHEDULE "0 * * * *"  -- Every hour at minute 0

This Sink:

• Aggregates revenue metrics by hour, category and country
• Exports to the revenue_summaries Kafka topic
• Runs every hour automatically
• Other services can consume these summaries for real-time processing

Export Real-Time Alerts

Create a Sink that exports high value orders for fraud detection or special handling:

NODE high_value_orders
SQL >
    SELECT
        order_id,
        customer_id,
        customer_name,
        product_name,
        order_total,
        timestamp,
        payment_method,
        customer_country
    FROM enriched_orders
    WHERE order_total > {{ Decimal(1000, 1000, description='Minimum order value threshold') }}
      AND timestamp >= now() - INTERVAL 5 MINUTE

TYPE sink
EXPORT_CONNECTION_NAME kafka_sink
EXPORT_KAFKA_TOPIC high_value_orders
EXPORT_SCHEDULE "*/5 * * * *"  -- Every 5 minutes

Deploy and Monitor Sinks

Deploy your Sinks:

tb --cloud deploy

Monitor Sink operations:

SELECT
    timestamp,
    pipe_name,
    status,
    elapsed_time,
    rows_written,
    error
FROM tinybird.jobs_log
WHERE job_type = 'sink'
  AND timestamp > now() - INTERVAL 24 hour
ORDER BY timestamp DESC

For Kafka-specific metrics:

SELECT
    timestamp,
    pipe_name,
    topic,
    messages_sent,
    bytes_sent,
    error
FROM tinybird.sinks_ops_log
WHERE timestamp > now() - INTERVAL 24 hour
ORDER BY timestamp DESC

Part 2: Connect Your Analytics API to BI Tools

Tinybird provides a ClickHouse® HTTP interface that enables you to connect BI tools and SQL clients directly to your Data Sources. This interface provides a standardized way to query your data using SQL, making it easy to integrate with your existing analytics workflow.

Connect BI Tools Using the ClickHouse HTTP Interface

Tinybird is compatible with the ClickHouse HTTP interface, enabling you to connect BI tools and SQL clients directly to your Data Sources. You can use tools like Grafana, Tableau, Power BI, Metabase and Superset to create dashboards and visualizations.

To connect, use these connection parameters:

• Protocol: HTTPS
• Host: clickhouse.<REGION>.tinybird.co (replace <REGION> with your workspace region)
• Port: 443
• SSL/TLS: Required
• Username: <WORKSPACE_NAME> (optional, for identification)
• Password: <TOKEN> (your Tinybird Auth Token)

The ClickHouse interface exposes your workspace Data Sources, Service Data Sources and system tables, allowing you to query them directly with SQL.

For detailed connection instructions and a complete list of compatible tools, see the ClickHouse Interface documentation.

Part 3: Monitor and Optimize Your Kafka Analytics Pipeline

Monitoring your Kafka connector is essential for maintaining a healthy pipeline. Tinybird provides the tinybird.kafka_ops_log Service Data Source for comprehensive monitoring of your Kafka integration.

What You Can Monitor

The kafka_ops_log Service Data Source provides visibility into:

• Consumer lag: Track how far behind your consumer is from the latest messages in each partition
• Throughput: Monitor message processing rates, bytes processed and ingestion success rates
• Errors and warnings: Identify deserialization failures, connectivity issues and Materialized View errors
• Partition performance: Analyze lag distribution and processing rates across partitions
• Connector health: Get comprehensive health summaries and identify inactive connectors

The connector's serverless architecture automatically scales consumers based on load and kafka_ops_log provides visibility into this autoscaling behavior along with error handling and debugging information.

Kafka Meta Columns

In addition to monitoring via kafka_ops_log, each Data Source connected to Kafka includes Kafka meta columns that store metadata from Kafka messages:

• __topic: The Kafka topic the message was read from
• __partition: The partition number
• __offset: The message offset within the partition
• __timestamp: The timestamp of the Kafka message
• __key: The message key (if present)

These columns can be used to track message ordering, detect gaps, analyze distribution and correlate with connector operations.

Comprehensive Monitoring Guide

For detailed monitoring queries, alerting setup, optimization strategies and best practices, see the Kafka monitoring guide. The guide includes:

• SQL queries for monitoring consumer lag, throughput and errors
• Alerting queries for high lag, error rates and processing stalls
• Partition-level analysis and performance comparison
• Optimization strategies for schema parsing, Materialized Views and partition distribution
• Integration with monitoring tools like Grafana, Datadog and PagerDuty

Part 4: Evolve Your Analytics API Schema Safely with Branches

As your application evolves, you'll need to add new fields, change data types, or modify your schema.

Tinybird's branching feature lets you test schema changes safely with production data before deploying to production.

Using Branches for Schema Evolution

Branches allow you to create ephemeral environments to test changes without affecting production. This is perfect for testing schema evolution.

Create a branch:

tb branch create schema_evolution_test

Start development in the branch:

tb --branch=schema_evolution_test dev

This starts a development session in your branch. Keep this terminal running while you work.

Example: Adding a New Column

Let's say your Kafka messages now include a discount_code field and you want to add it to your schema.

1. Update the Kafka Data Source schema:

SCHEMA >
    `order_id` String `json:$.order_id`,
    `customer_id` String `json:$.customer_id`,
    `product_id` Int32 `json:$.product_id`,
    `quantity` Int32 `json:$.quantity`,
    `price` Decimal(10, 2) `json:$.price`,
    `order_total` Decimal(10, 2) `json:$.order_total`,
    `timestamp` DateTime `json:$.timestamp`,
    `payment_method` LowCardinality(String) `json:$.payment_method`,
    `shipping_address` String `json:$.shipping_address`,
    `discount_code` Nullable(LowCardinality(String)) `json:$.discount_code`,  -- New field
    `data` String `json:$`

ENGINE MergeTree
ENGINE_PARTITION_KEY toYYYYMM(timestamp)
ENGINE_SORTING_KEY timestamp, order_id

KAFKA_CONNECTION_NAME kafka_ecommerce
KAFKA_TOPIC orders
KAFKA_GROUP_ID {{ tb_secret("KAFKA_GROUP_ID", "ecommerce_consumer") }}
KAFKA_AUTO_OFFSET_RESET latest

FORWARD_QUERY >
    SELECT 
        order_id,
        customer_id,
        product_id,
        quantity,
        price,
        order_total,
        timestamp,
        payment_method,
        shipping_address,
        NULL as discount_code,  -- Default value for existing rows
        data
    FROM orders_kafka

Key points:

• We make discount_code Nullable since old messages won't have this field
• The FORWARD_QUERY provides a default value (NULL) for existing rows
• New messages from Kafka will include the discount_code field automatically

2. Update the Enrichment Materialized View:

First, update the datasource file to include the new field and FORWARD_QUERY:

SCHEMA >
    `order_id` String,
    `customer_id` String,
    `product_id` String,
    `quantity` Int32,
    `price` Decimal(10, 2),
    `order_total` Decimal(10, 2),
    `timestamp` DateTime,
    `payment_method` LowCardinality(String),
    `shipping_address` String,
    `discount_code` Nullable(LowCardinality(String)),
    `product_name` String,
    `product_category` LowCardinality(String),
    `product_brand` String,
    `product_base_price` Decimal(10, 2),
    `customer_name` String,
    `customer_country` LowCardinality(String),
    `customer_segment` LowCardinality(String),
    `discount_amount` Decimal(10, 2)

ENGINE MergeTree
ENGINE_PARTITION_KEY toYYYYMM(timestamp)
ENGINE_SORTING_KEY timestamp, order_id

FORWARD_QUERY >
    SELECT 
        order_id,
        customer_id,
        product_id,
        quantity,
        price,
        order_total,
        timestamp,
        payment_method,
        shipping_address,
        NULL as discount_code,  -- Default for existing enriched data
        product_name,
        product_category,
        product_brand,
        product_base_price,
        customer_name,
        customer_country,
        customer_segment,
        discount_amount
    FROM enriched_orders

Then update the materialized view pipe:

NODE enriched_orders
SQL >
    SELECT
        o.order_id,
        o.customer_id,
        o.product_id,
        o.quantity,
        o.price,
        o.order_total,
        o.timestamp,
        o.payment_method,
        o.shipping_address,
        o.discount_code,  -- Include new field
        -- Enrich with product data
        p.product_name,
        p.category as product_category,
        p.brand as product_brand,
        p.base_price as product_base_price,
        -- Enrich with customer data
        c.customer_name,
        c.country as customer_country,
        c.customer_segment,
        -- Calculate derived fields
        o.order_total - (o.quantity * p.base_price) as discount_amount
    FROM orders_kafka o
    LEFT JOIN products FINAL p ON o.product_id = p.product_id
    LEFT JOIN customers FINAL c ON o.customer_id = c.customer_id

TYPE materialized
DATASOURCE enriched_orders

3. Update Materialized Views that use the new field:

If you want to aggregate by discount_code, update your revenue metrics Materialized View. First, update the datasource file:

SCHEMA >
    `hour` DateTime,
    `product_category` LowCardinality(String),
    `customer_country` LowCardinality(String),
    `discount_code` Nullable(LowCardinality(String)),
    `order_count` UInt64,
    `total_revenue` Decimal(10, 2),
    `avg_order_value` Decimal(10, 2),
    `total_units_sold` UInt32

ENGINE SummingMergeTree
ENGINE_PARTITION_KEY toYYYYMM(hour)
ENGINE_SORTING_KEY hour, product_category, customer_country, discount_code

FORWARD_QUERY >
    SELECT 
        hour,
        product_category,
        customer_country,
        NULL as discount_code,  -- Default for existing aggregated data
        order_count,
        total_revenue,
        avg_order_value,
        total_units_sold
    FROM revenue_metrics

Then update the materialized view pipe:

NODE revenue_by_hour
SQL >
    SELECT
        toStartOfHour(timestamp) as hour,
        product_category,
        customer_country,
        discount_code,  -- New grouping dimension
        count() as order_count,
        sum(order_total) as total_revenue,
        avg(order_total) as avg_order_value,
        sum(quantity) as total_units_sold
    FROM enriched_orders
    GROUP BY hour, product_category, customer_country, discount_code

TYPE materialized
DATASOURCE revenue_metrics

4. Update API Endpoints:

Add the new field to your endpoints:

TOKEN "metrics_api_token" READ

DESCRIPTION >
    Real-time revenue metrics by time window, category, country and discount code

NODE filtered_metrics
SQL >
    %
    SELECT *
    FROM revenue_metrics
    WHERE hour >= toDateTime({{ String(start_time, '2025-01-27 00:00:00', description='Start time') }})
      AND hour <= toDateTime({{ String(end_time, '2025-01-27 23:59:59', description='End time') }})
      {\% if defined(product_category) %}
      AND product_category = {{ String(product_category, description='Filter by product category') }}
      {\% endif %}
      {\% if defined(customer_country) %}
      AND customer_country = {{ String(customer_country, description='Filter by country') }}
      {\% endif %}
      {\% if defined(discount_code) %}
      AND discount_code = {{ String(discount_code, description='Filter by discount code') }}
      {\% endif %}

NODE aggregated
SQL >
    SELECT
        sum(order_count) as total_orders,
        sum(total_revenue) as total_revenue,
        avg(avg_order_value) as avg_order_value,
        sum(total_units_sold) as total_units_sold,
        min(hour) as period_start,
        max(hour) as period_end,
        discount_code  -- Include in results
    FROM filtered_metrics
    GROUP BY product_category, customer_country, discount_code
    ORDER BY total_revenue DESC

Testing Schema Changes in a Branch

1. Deploy to the branch:

While tb --branch=schema_evolution_test dev is running, your changes are automatically applied to the branch. You can also deploy explicitly:

tb --branch=schema_evolution_test deploy

2. Test with production data:

Branches can use production data. When creating the branch, use the --last-partition flag to bring the last partition of production data:

tb branch create schema_evolution_test --last-partition

3. Validate the changes:

# Test queries in the branch
tb --branch=schema_evolution_test sql "SELECT discount_code, count(*) FROM enriched_orders WHERE discount_code IS NOT NULL GROUP BY discount_code"

# Test endpoints
tb --branch=schema_evolution_test endpoint data api_revenue_metrics

4. Check deployment before promoting:

tb --cloud deploy --check

This validates your deployment without actually deploying, catching potential issues early.

5. Deploy to production:

Once validated, deploy to cloud:

tb --cloud deploy

Tinybird automatically:

• Creates a staging deployment
• Runs the FORWARD_QUERY to migrate existing data
• Validates the migration
• Promotes to production if successful
• Rolls back if there are errors

When to Use FORWARD_QUERY

Use FORWARD_QUERY when:

• Adding new columns: Provide default values for existing rows
• Changing data types: Transform existing data to the new type
• Removing columns: Use SELECT * EXCEPT 'column_name'
• Renaming columns: Map old column names to new ones

Example: Changing a data type:

SCHEMA >
    `order_id` String `json:$.order_id`,
    `quantity` Int32 `json:$.quantity`,
    `price` Decimal(10, 2) `json:$.price`,  -- Changed from Float64
    `order_total` Decimal(10, 2) `json:$.order_total`,
    `timestamp` DateTime `json:$.timestamp`,
    ...

ENGINE MergeTree
ENGINE_PARTITION_KEY toYYYYMM(timestamp)
ENGINE_SORTING_KEY timestamp, order_id

FORWARD_QUERY >
    SELECT 
        * EXCEPT(price),
        toDecimal32(price, 2) as price
    FROM orders_kafka

Example: Adding a calculated field:

SCHEMA >
    `order_id` String `json:$.order_id`,
    `quantity` Int32 `json:$.quantity`,
    `price` Decimal(10, 2) `json:$.price`,
    `order_total` Decimal(10, 2) `json:$.order_total`,
    `discount_percentage` Float32,  -- New calculated field
    `timestamp` DateTime `json:$.timestamp`,
    ...

ENGINE MergeTree
ENGINE_PARTITION_KEY toYYYYMM(timestamp)
ENGINE_SORTING_KEY timestamp, order_id

FORWARD_QUERY >
    SELECT 
        *,
        CASE 
            WHEN order_total > 0 THEN (1 - (order_total / (quantity * price))) * 100
            ELSE 0
        END as discount_percentage
    FROM orders_kafka

Best Practices for Schema Evolution

1. Always test in a branch or locally first: Use branches to validate schema changes with production data, or test locally with sample data
2. Use --check before deploying: Run tb --cloud deploy --check to catch issues early
3. Use Nullable or DEFAULT for new fields: Old messages won't have new fields. Use Nullable() types if NULL has semantic meaning, or use DEFAULT values with FORWARD_QUERY to provide defaults for existing data
4. Provide sensible defaults: Use FORWARD_QUERY to provide default values for existing data when using DEFAULT instead of Nullable
5. Update downstream components: Remember to update Materialized Views and endpoints that use the changed schema
6. Monitor after deployment: Check datasources_ops_log for backfill progress and any errors

For more details, see the schema evolution guide and branches documentation.

Part 5: Advanced Patterns for Scaling Your Analytics API

Pattern 1: Incremental Dimension Updates

For large dimension tables that change frequently, use incremental updates:

NODE customers_incremental
SQL >
    SELECT *
    FROM postgresql(
        {{ tb_secret('pg_host') }} || ':' || {{ tb_secret('pg_port') }},
        {{ tb_secret('pg_database') }},
        'customers',
        {{ tb_secret('pg_username') }},
        {{ tb_secret('pg_password') }}
    )
    WHERE updated_at > (
        SELECT max(updated_at) FROM customers
    )

TYPE COPY
TARGET_DATASOURCE customers
COPY_MODE append
COPY_SCHEDULE "*/15 * * * *"  -- Every 15 minutes

Pattern 2: Multiple Kafka Topics

Consume from multiple Kafka topics by creating separate Data Sources:

KAFKA_TOPIC orders

KAFKA_TOPIC payments

KAFKA_TOPIC shipping

Then join them in Materialized Views or queries as needed.

Pattern 3: Real-Time Alerts

Create an endpoint that triggers alerts based on thresholds:

TOKEN "metrics_api_token" READ

DESCRIPTION >
    Alert when revenue drops below threshold

NODE low_revenue_check
SQL >
    %
    SELECT
        toStartOfMinute(timestamp) as minute,
        sum(order_total) as revenue
    FROM enriched_orders
    WHERE timestamp >= now() - INTERVAL 5 MINUTE
    GROUP BY minute
    HAVING revenue < {{ Decimal(1000, 1000, description='Revenue threshold') }}

TYPE endpoint

Call this endpoint periodically and trigger alerts when revenue drops below the threshold.

Pattern 4: Time Windowed Aggregations

Create Materialized Views for different time windows. First, create the datasource file:

SCHEMA >
    `date` Date,
    `product_category` LowCardinality(String),
    `customer_country` LowCardinality(String),
    `total_revenue` Decimal(10, 2),
    `order_count` UInt64

ENGINE SummingMergeTree
ENGINE_PARTITION_KEY toYYYYMM(date)
ENGINE_SORTING_KEY date, product_category, customer_country

Then create the materialized view pipe:

NODE revenue_by_day
SQL >
    SELECT
        toStartOfDay(timestamp) as date,
        product_category,
        customer_country,
        sum(order_total) as total_revenue,
        count() as order_count
    FROM enriched_orders
    GROUP BY date, product_category, customer_country

TYPE materialized
DATASOURCE revenue_by_day

Pattern 5: Customer Lifetime Value

Calculate and store customer lifetime value. First, create the datasource file:

SCHEMA >
    `customer_id` String,
    `customer_name` String,
    `customer_country` LowCardinality(String),
    `total_orders` UInt64,
    `lifetime_value` Decimal(10, 2),
    `avg_order_value` Decimal(10, 2),
    `first_order_date` DateTime,
    `last_order_date` DateTime,
    `customer_lifetime_days` UInt32

ENGINE ReplacingMergeTree
ENGINE_SORTING_KEY customer_id

Then create the materialized view pipe:

NODE customer_ltv
SQL >
    SELECT
        customer_id,
        customer_name,
        customer_country,
        count() as total_orders,
        sum(order_total) as lifetime_value,
        avg(order_total) as avg_order_value,
        min(timestamp) as first_order_date,
        max(timestamp) as last_order_date,
        dateDiff('day', min(timestamp), max(timestamp)) as customer_lifetime_days
    FROM enriched_orders
    GROUP BY customer_id, customer_name, customer_country

TYPE materialized
DATASOURCE customer_ltv

Pattern 6: Top N Queries

Create Materialized Views optimized for top N queries. First, create the datasource file:

SCHEMA >
    `hour` DateTime,
    `product_id` String,
    `product_name` String,
    `product_category` LowCardinality(String),
    `order_count` UInt64,
    `units_sold` UInt32,
    `revenue` Decimal(10, 2)

ENGINE SummingMergeTree
ENGINE_PARTITION_KEY toYYYYMM(hour)
ENGINE_SORTING_KEY hour, product_id

Then create the materialized view pipe:

NODE top_products_hourly
SQL >
    SELECT
        toStartOfHour(timestamp) as hour,
        product_id,
        product_name,
        product_category,
        count() as order_count,
        sum(quantity) as units_sold,
        sum(order_total) as revenue
    FROM enriched_orders
    GROUP BY hour, product_id, product_name, product_category

TYPE materialized
DATASOURCE top_products

Then query for top products efficiently:

SELECT
    product_id,
    product_name,
    sum(revenue) as total_revenue
FROM top_products
WHERE hour >= now() - INTERVAL 7 DAY
GROUP BY product_id, product_name
ORDER BY total_revenue DESC
LIMIT 10

Next Steps

You've now extended your real-time metrics API with production-ready features:

• Export to Kafka - Send processed data to other services
• BI Tool Integration - Connect to Tableau, Power BI, Grafana and more
• Comprehensive Monitoring - Track performance and identify issues
• Optimization Strategies - Improve throughput and reduce lag
• Common Patterns - Real-world patterns for scaling

Continue Building

Here are some ways to further extend your pipeline:

1. Add more Sinks - Export to S3, GCS, or other Kafka topics
2. Create custom dashboards - Build dashboards using your BI tools
3. Set up automated alerts - Use monitoring queries to trigger alerts
4. Scale your pipeline - Add more topics, enrichments and endpoints
5. Optimize performance - Fine-tune Materialized Views and queries

Resources

• Kafka Connector Documentation - Complete setup guide
• Kafka Sink Documentation - Export data to Kafka
• Monitoring Kafka Connectors - Comprehensive monitoring guide
• Materialized Views - Performance optimization
• API Endpoints - Endpoint configuration

About Tinybird for Real-Time Analytics APIs from Kafka

Tinybird is a serverless platform for building real-time analytics APIs from Kafka and other data sources.

It provides:

• Serverless Kafka connectors with automatic consumer scaling
• ClickHouse powered analytics with sub second query latency
• Automatic API generation from SQL queries
• Built in monitoring through Service Data Sources
• Schema evolution tools with branches and FORWARD_QUERY

Tinybird handles infrastructure management, scaling and monitoring so you can focus on building analytics features.

The platform automatically scales consumers based on load, provides comprehensive monitoring through kafka_ops_log and enables safe schema evolution without downtime.

Frequently Asked Questions (FAQs)

How do you export data from an analytics API back to Kafka?

To export data from your analytics API back to Kafka, use Tinybird Sinks. Create a Sink pipe that queries your Data Sources and exports the results to a Kafka topic.

Sinks can run on a schedule (e.g., hourly) or be triggered manually. The exported data can then be consumed by other services in your event driven architecture.

Can you connect BI tools to a real-time analytics API from Kafka?

Yes, you can connect BI tools to your real-time analytics API from Kafka using Tinybird's ClickHouse HTTP interface. Tools like Grafana, Tableau, Power BI, Metabase and Superset can connect directly to your Data Sources using standard ClickHouse connection parameters. This enables real-time dashboards and visualizations without additional infrastructure.

How do you monitor a Kafka analytics pipeline?

Monitor your Kafka analytics pipeline using Tinybird's tinybird.kafka_ops_log Service Data Source. This provides visibility into consumer lag, throughput, errors and partition performance.

You can query this data to track metrics, set up alerts and optimize your pipeline. For comprehensive monitoring queries and best practices, see the Kafka monitoring guide.

How do you evolve schemas in a real-time analytics API?

Evolve schemas safely using Tinybird's branching feature and FORWARD_QUERY. Create a branch to test schema changes with production data, update your datasource schema and use FORWARD_QUERY to provide default values for existing data.

This allows you to add new columns, change data types, or modify schemas without breaking production. Always test in a branch or locally before deploying.

What are the best practices for scaling a Kafka analytics API?

Best practices for scaling include:

• Monitor consumer lag and throughput regularly
• Use Materialized Views for pre-aggregated metrics
• Optimize schemas with explicit column definitions
• Use appropriate partition keys for even distribution
• Set up alerts for high lag and error rates
• Leverage Tinybird's automatic consumer scaling

How fast is a real-time analytics API from Kafka?

A real-time analytics API from Kafka can serve data within seconds of events arriving in Kafka. With proper optimization using Materialized Views and efficient schemas, query latency can be sub-100ms even with high throughput. The exact latency depends on your processing pipeline and query complexity.

Start Building

Ready to build your own real-time analytics pipeline? Sign up for Tinybird and start building today. The free Build plan includes everything you need to get started.

If you have questions or want to share what you've built, join our Slack community for support and inspiration.