Realtime Compute for Apache Flink:Job performance FAQ

Enable miniBatch to improve throughput

miniBatch reduces how often Flink reads state data by buffering records and processing them in micro-batches. This improves throughput and reduces data output, at the cost of slightly higher latency. Micro-batch processing does not apply to scenarios that require extremely low latency, but in data aggregation scenarios it is strongly recommended to improve system performance.

miniBatch triggers micro-batch processing based on event messages inserted at the source at a specified interval.

To enable miniBatch, add the following configuration to the Other Configuration field in the Parameters section on the Configuration tab of the deployment details page:

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s

Enable LocalGlobal to resolve common data hotspot issues

LocalGlobal splits aggregation into two phases — local and global — similar to the combine and reduce phases in MapReduce. In the local phase, Flink aggregates a micro-batch of data cached at each upstream node and produces an accumulator value. In the global phase, the accumulators are merged into the final result. By filtering skewed data early, LocalGlobal reduces data hotspot issues in the global aggregation phase.

LocalGlobal is suitable for deployments that use common aggregate functions such as SUM, COUNT, MAX, MIN, and AVG.

LocalGlobal is enabled by default, but it requires miniBatch to be enabled first. In addition, AggregateFunction must be used to merge data.

To verify that LocalGlobal is active, check whether a GlobalGroupAggregate or LocalGroupAggregate node appears in the final topology.

Enable PartialFinal to resolve data hotspot issues with COUNT DISTINCT

LocalGlobal is less effective for COUNT DISTINCT because local aggregation cannot remove duplicate distinct keys, which causes large amounts of data to accumulate in the global aggregation phase. PartialFinal addresses this by automatically scattering data and dividing the aggregation into two phases.

PartialFinal is suitable when COUNT DISTINCT aggregation cannot meet your performance requirements.

PartialFinal is disabled by default. To enable it, add the following configuration to the Other Configuration field:

table.optimizer.distinct-agg.split.enabled: true

To verify that PartialFinal is active, check whether one-layer aggregation changes to two-layer aggregation in the final topology.

Use AGG WITH FILTER instead of AGG WITH CASE WHEN for COUNT DISTINCT

When computing COUNT DISTINCT across multiple conditions on the same field, use AGG WITH FILTER instead of AGG WITH CASE WHEN. The SQL optimizer recognizes that all COUNT DISTINCT expressions operate on the same field and uses a single shared state instance instead of one per expression — reducing both state read/write operations and overall resource usage. In performance tests, the AGG WITH FILTER syntax doubles the deployment performance compared to AGG WITH CASE WHEN.

Original statement:

COUNT(distinct visitor_id) as UV1, COUNT(distinct case when is_wireless='y' then visitor_id else null end) as UV2

Optimized statement:

COUNT(distinct visitor_id) as UV1, COUNT(distinct visitor_id) filter (where is_wireless='y') as UV2

TopN algorithms

Flink selects the TopN algorithm based on the input stream type:

• AppendRank: Used for static input streams, such as Log Service data sources.

• UpdateFastRank: The optimal algorithm for dynamic input streams (aggregated or joined data). Use this algorithm when possible.

• RetractRank: The fallback algorithm for dynamic input streams. If it doesn't meet your performance requirements, switch to UpdateFastRank.

The algorithm name appears in the corresponding node name in the job topology.

Switching from RetractRank to UpdateFastRank

To use UpdateFastRank, the following conditions must be met:

• The input stream is a dynamic stream.

• The input stream contains primary key information — for example, the GROUP BY clause aggregates on the primary key.

• The ORDER BY field or function (such as COUNT, COUNT DISTINCT, or SUM with positive values) updates monotonically in the opposite direction of sorting.

For ORDER BY SUM DESC, add a filter to guarantee positive SUM values:

insert into print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt -- Exclude rownum from output to reduce the volume of data written to the result table.
FROM (
    SELECT
      *,
      ROW_NUMBER() OVER (
        PARTITION BY cate_id,
        stat_date -- Include stat_date to prevent out-of-order data when state TTL expires.
        ORDER BY pay_ord_amt DESC
      ) as rownum -- Sort by the aggregated SUM value.
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          -- SUM returns positive values, so the result is monotonically increasing.
          -- This allows TopN to use the optimized algorithm for top 100 records.
          sum(total_fee) filter (where total_fee >= 0) as pay_ord_amt
        FROM random_test
        WHERE total_fee >= 0
        GROUP BY cate_name, seller_id, stat_date, cate_id
      ) a
    ) WHERE rownum <= 100;

TopN optimization methods

Omit rownum from TopN output

Don't include rownum in the final SELECT of a TopN query. We recommend that you sort the results when they are finally displayed in the frontend. This significantly reduces the amount of data that is to be written to the result table.

For more details, see Top-N.

Increase the TopN cache size

TopN uses an in-state cache to reduce disk access. The cache hit ratio is calculated as:

cache_hit = cache_size * parallelism / top_n / partition_key_num

For example, with Top100, a cache of 10,000 records, parallelism of 50, and 100,000 PARTITION BY keys: 10,000 × 50 / 100 / 100,000 = 5%. A low cache hit ratio means frequent disk state access, which significantly degrades performance.

If the number of PARTITION BY keys is large, increase the cache size and heap memory of TopN:

table.exec.rank.topn-cache-size: 200000

With a cache size of 200,000, the hit ratio for the example above becomes 200,000 × 50 / 100 / 100,000 = 100%.

For instructions on adjusting heap memory, see Configure job deployments.

Include a time field in PARTITION BY

Add a time field such as Day to the PARTITION BY clause for daily rankings. Without it, state data expiry (time-to-live, TTL) can cause TopN results to go out of order.

Syntax

Flink SQL does not have native deduplication syntax. Use ROW_NUMBER() with an OVER clause to keep the first or last record per key. Deduplication is implemented as a special case of TopN.

SELECT *
FROM (
   SELECT *,
    ROW_NUMBER() OVER (PARTITION BY col1[, col2..]
     ORDER BY timeAttributeCol [asc|desc]) AS rownum
   FROM table_name)
WHERE rownum = 1

The query has two levels:

1. Use ROW_NUMBER() to rank records by the time attribute:

   • proctime: Flink deduplicates based on processing time. Results may vary across reruns.

   • rowtime: Flink deduplicates based on event time. Results are deterministic.

2. Keep only the record where rownum = 1 for each key.

Deduplicate Keep FirstRow

Keep FirstRow retains the first record per key and discards subsequent duplicates. State data stores only the primary key, making state access very efficient.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) as rowNum
  FROM T
)
WHERE rowNum = 1

This example deduplicates table T on column b, keeping the first record based on processing time. Call PROCTIME() directly to use processing time without declaring a proctime attribute.

Deduplicate Keep LastRow

Keep LastRow retains the most recent record per key. It slightly outperforms the LAST_VALUE function.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) as rowNum
  FROM T
)
WHERE rowNum = 1

This example deduplicates table T on columns b and d, keeping the latest record based on event time.

Enable LocalGlobal to resolve common data hotspot issues

LocalGlobal splits aggregation into two phases — local and global — similar to the combine and reduce phases in MapReduce. In the local phase, Flink aggregates a micro-batch of data cached at each upstream node and produces an accumulator value. In the global phase, the accumulators are merged into the final result. By filtering skewed data early, LocalGlobal reduces data hotspot pressure in the global aggregation phase.

LocalGlobal is suitable for deployments using common aggregate functions such as SUM, COUNT, MAX, MIN, and AVG.

LocalGlobal is enabled by default, but requires miniBatch to be enabled first. AggregateFunction must also be used to merge data.

To verify that LocalGlobal is active, check whether a GlobalGroupAggregate or LocalGroupAggregate node appears in the final topology.

Enable PartialFinal to resolve data hotspot issues with COUNT DISTINCT

LocalGlobal is less effective for COUNT DISTINCT because local aggregation cannot eliminate duplicate distinct keys. PartialFinal addresses this by scattering data automatically and splitting aggregation into two phases.

PartialFinal is suitable when COUNT DISTINCT aggregation performance doesn't meet requirements.

PartialFinal is disabled by default. Enable it with:

table.optimizer.distinct-agg.split.enabled: true

To verify, check whether one-layer aggregation changes to two-layer aggregation in the final topology.