Real‑Time Dashboard Stack (RocketMQ + Flink + Elasticsearch + Kibana): Part 2 – Development Environment Setup

Introduction – Real‑time business monitoring such as e‑commerce dashboards requires a streaming architecture. The combination of RocketMQ, Flink, Elasticsearch and Kibana can provide low‑latency aggregation and visualization.

Series plan – Part 1 prepares the basic environment (this article), Part 2 builds the data pipeline, and Part 3 designs the dashboard.

1. Simulating production data

A sample MySQL order table oms_order is created with fields for order ID, order number, user ID, monetary amounts (using DECIMAL), status, payment type, source type, receiver information, logistics, timestamps, and logical delete flag. The DDL is provided in a CREATE TABLE statement.

A Bash script mock_order.sh generates mock order messages and sends them to RocketMQ. The script defines configuration variables (NameServer address, topic, mqadmin path), random data sources for users, products, receivers, phones, and cities, and then enters an infinite loop that:

Selects random user, receiver, phone, and city JSON.

Parses city JSON with awk to extract province, city, region, and detail.

Generates a globally unique order number using a timestamp and random suffix.

Builds one or two random product items, calculates total amount, applies a simple promotion (‑50 when total > 1000), and assembles a JSON payload.

Compresses the JSON to a single line and sends it to RocketMQ using mqadmin sendMessage with the order number as the key.

Logs success or failure and sleeps 1–3 seconds before the next iteration.

2. Developing the Flink job for the dashboard

The Flink job consists of three parts:

Source : A RocketMQ virtual table maps JSON messages to structured fields ( order_sn, pay_amount, receiver_province, receiver_city, create_time, etc.). Watermarks are defined on a timestamp derived from create_time.

Transform : Real‑time aggregation uses a daily tumbling window ( TUMBLE) to compute cumulative metrics such as total order count and total GMV per province and city.

Sink : An Elasticsearch table is created with an upsert mode and a composite primary key ( stat_date, province, city). The upsert ensures that each day's statistics are overwritten with the latest values.

The Java implementation ( OrderDashboardJob) performs the following steps:

package com.qinyadan.system;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class OrderDashboardJob {
    public static void main(String[] args) throws Exception {
        // 1. Initialize streaming environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        // Enable checkpointing for exactly‑once semantics
        env.enableCheckpointing(5000);
        // 2. Create Table environment
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 3. Register RocketMQ source table
        tableEnv.executeSql(
            "CREATE TABLE rocketmq_order_source (
" +
            "    `order_sn` STRING,
" +
            "    `pay_amount` DECIMAL(12, 2),
" +
            "    `receiver_province` STRING,
" +
            "    `receiver_city` STRING,
" +
            "    `create_time` STRING,
" +
            "    `ts` AS TO_TIMESTAMP(`create_time`, 'yyyy-MM-dd HH:mm:ss'),
" +
            "    WATERMARK FOR `ts` AS `ts` - INTERVAL '5' SECOND
" +
            ") WITH (
" +
            "    'connector' = 'rocketmq',
" +
            "    'topic' = 'order_topic',
" +
            "    'consumerGroup' = 'flink_order_group',
" +
            "    'nameServerAddress' = '127.0.0.1:9876',
" +
            "    'scanStartupMode' = 'latest',
" +
            "    'format' = 'json'
" +
            ")");
        // 4. Register Elasticsearch sink table
        tableEnv.executeSql(
            "CREATE TABLE es_order_dashboard_sink (
" +
            "    `stat_date` STRING,
" +
            "    `province` STRING,
" +
            "    `city` STRING,
" +
            "    `total_order_count` BIGINT,
" +
            "    `total_gmv` DECIMAL(16, 2),
" +
            "    PRIMARY KEY (`stat_date`, `province`, `city`) NOT ENFORCED
" +
            ") WITH (
" +
            "    'connector' = 'elasticsearch',
" +
            "    'hosts' = 'http://127.0.0.1:9200',
" +
            "    'index' = 'order_dashboard_city',
" +
            "    'format' = 'json'
" +
            ")");
        // 5. Define transformation and insert
        String insertSql =
            "INSERT INTO es_order_dashboard_sink
" +
            "SELECT
" +
            "    DATE_FORMAT(TUMBLE_START(ts, INTERVAL '1' DAY), 'yyyy-MM-dd') AS stat_date,
" +
            "    receiver_province AS province,
" +
            "    receiver_city AS city,
" +
            "    COUNT(order_sn) AS total_order_count,
" +
            "    SUM(pay_amount) AS total_gmv
" +
            "FROM rocketmq_order_source
" +
            "GROUP BY TUMBLE(ts, INTERVAL '1' DAY), receiver_province, receiver_city";
        tableEnv.executeSql(insertSql);
    }
}

Notes on Elasticsearch upsert

Defining PRIMARY KEY (stat_date, province, city) makes Flink generate a combined hash that is used as the document _id in Elasticsearch.

When new order streams arrive or retract streams update the window, Elasticsearch overwrites the existing document with the same _id, achieving perfect real‑time updates for the dashboard.

3. Visualising the stored data

The article shows three steps for building the dashboard:

Register the business data source table (image shown).

Compute business metrics using window functions or multi‑stream joins (image shown).

Persist results to sinks such as Redis for low‑latency queries, intermediate MQ‑backed tables, or Elasticsearch/MySQL for durable storage (image shown).

Current experimental setup sends data to RocketMQ first, then batches writes to Elasticsearch for dashboard display.

The article concludes with a note that the author’s experimental workflow first pushes data to RocketMQ and later batches it into Elasticsearch for dashboard consumption.