Performance Tuning

    In this page, we will introduce some useful optimization options and the internals of streaming aggregation which will bring great improvement in some cases.

    By default, group aggregation operators process input records one by one, i.e., (1) read accumulator from state, (2) accumulate/retract record to the accumulator, (3) write accumulator back to state, (4) the next record will do the process again from (1). This processing pattern may increase the overhead of StateBackend (especially for RocksDB StateBackend). Besides, data skew, which is very common in production, will worsen the problem and make it easy for the jobs to be under backpressure situations.

    The core idea of mini-batch aggregation is caching a bundle of inputs in a buffer inside of the aggregation operator. When the bundle of inputs is triggered to process, only one operation per key to access state is needed. This can significantly reduce the state overhead and get a better throughput. However, this may increase some latency because it buffers some records instead of processing them in an instant. This is a trade-off between throughput and latency.

    The following figure explains how the mini-batch aggregation reduces state operations.

    MiniBatch optimization is disabled by default for group aggregation. In order to enable this optimization, you should set options , table.exec.mini-batch.allow-latency and table.exec.mini-batch.size. Please see configuration page for more details.

    The following examples show how to enable these options.

    Java

    Scala

    1. // instantiate table environment
    2. val tEnv: TableEnvironment = ...
    4. // access flink configuration
    5. val configuration = tEnv.getConfig().getConfiguration()
    6. // set low-level key-value options
    7. configuration.setString("table.exec.mini-batch.enabled", "true") // enable mini-batch optimization
    8. configuration.setString("table.exec.mini-batch.allow-latency", "5 s") // use 5 seconds to buffer input records
    9. configuration.setString("table.exec.mini-batch.size", "5000") // the maximum number of records can be buffered by each aggregate operator task
    1. # instantiate table environment
    2. t_env = ...
    4. # access flink configuration
    5. configuration = t_env.get_config().get_configuration();
    6. # set low-level key-value options
    7. configuration.set_string("table.exec.mini-batch.enabled", "true"); # enable mini-batch optimization
    8. configuration.set_string("table.exec.mini-batch.allow-latency", "5 s"); # use 5 seconds to buffer input records
    9. configuration.set_string("table.exec.mini-batch.size", "5000"); # the maximum number of records can be buffered by each aggregate operator task

    Local-Global is proposed to solve data skew problem by dividing a group aggregation into two stages, that is doing local aggregation in upstream firstly, and followed by global aggregation in downstream, which is similar to Combine + Reduce pattern in MapReduce. For example, considering the following SQL:

    1. SELECT color, sum(id)
    2. GROUP BY color

    It is possible that the records in the data stream are skewed, thus some instances of aggregation operator have to process much more records than others, which leads to hotspot. The local aggregation can help to accumulate a certain amount of inputs which have the same key into a single accumulator. The global aggregation will only receive the reduced accumulators instead of large number of raw inputs. This can significantly reduce the network shuffle and the cost of state access. The number of inputs accumulated by local aggregation every time is based on mini-batch interval. It means local-global aggregation depends on mini-batch optimization is enabled.

    The following figure shows how the local-global aggregation improve performance.

    Performance Tuning - 图2

    The following examples show how to enable the local-global aggregation.

    Java

    Scala

    1. // instantiate table environment
    2. val tEnv: TableEnvironment = ...
    4. // access flink configuration
    5. val configuration = tEnv.getConfig().getConfiguration()
    7. configuration.setString("table.exec.mini-batch.enabled", "true") // local-global aggregation depends on mini-batch is enabled
    8. configuration.setString("table.exec.mini-batch.allow-latency", "5 s")
    9. configuration.setString("table.exec.mini-batch.size", "5000")
    10. configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE") // enable two-phase, i.e. local-global aggregation

    Python

    1. # instantiate table environment
    2. t_env = ...
    4. # access flink configuration
    5. configuration = t_env.get_config().get_configuration();
    6. # set low-level key-value options
    7. configuration.set_string("table.exec.mini-batch.enabled", "true"); # local-global aggregation depends on mini-batch is enabled
    8. configuration.set_string("table.exec.mini-batch.allow-latency", "5 s");
    9. configuration.set_string("table.exec.mini-batch.size", "5000");
    10. configuration.set_string("table.optimizer.agg-phase-strategy", "TWO_PHASE"); # enable two-phase, i.e. local-global aggregation

    Local-Global optimization is effective to eliminate data skew for general aggregation, such as SUM, COUNT, MAX, MIN, AVG. But its performance is not satisfactory when dealing with distinct aggregation.

    For example, if we want to analyse how many unique users logined today. We may have the following query:

    1. SELECT day, COUNT(DISTINCT user_id)
    2. FROM T

    COUNT DISTINCT is not good at reducing records if the value of distinct key (i.e. user_id) is sparse. Even if local-global optimization is enabled, it doesn’t help much. Because the accumulator still contain almost all the raw records, and the global aggregation will be the bottleneck (most of the heavy accumulators are processed by one task, i.e. on the same day).

    After split distinct aggregate, the above query will be rewritten into the following query automatically:

    The following figure shows how the split distinct aggregation improve performance (assuming color represents days, and letter represents user_id).

    NOTE: Above is the simplest example which can benefit from this optimization. Besides that, Flink supports to split more complex aggregation queries, for example, more than one distinct aggregates with different distinct key (e.g. COUNT(DISTINCT a), SUM(DISTINCT b)), works with other non-distinct aggregates (e.g. SUM, MAX, MIN, COUNT).

    The following examples show how to enable the split distinct aggregation optimization.

    Java

    1. // instantiate table environment
    4. tEnv.getConfig()        // access high-level configuration
    5.   .getConfiguration()   // set low-level key-value options
    6.   .setString("table.optimizer.distinct-agg.split.enabled", "true");  // enable distinct agg split

    Scala

    1. // instantiate table environment
    2. val tEnv: TableEnvironment = ...
    4. tEnv.getConfig         // access high-level configuration
    5.   .getConfiguration    // set low-level key-value options
    6.   .setString("table.optimizer.distinct-agg.split.enabled", "true")  // enable distinct agg split

    Python

    1. # instantiate table environment
    2. t_env = ...
    4. t_env.get_config()        # access high-level configuration
    5.   .get_configuration()    # set low-level key-value options
    6.   .set_string("table.optimizer.distinct-agg.split.enabled", "true"); # enable distinct agg split

    In some cases, user may need to calculate the number of UV (unique visitor) from different dimensions, e.g. UV from Android, UV from iPhone, UV from Web and the total UV. Many users will choose CASE WHEN to support this, for example:

    However, it is recommended to use FILTER syntax instead of CASE WHEN in this case. Because FILTER is more compliant with the SQL standard and will get much more performance improvement. FILTER is a modifier used on an aggregate function to limit the values used in an aggregation. Replace the above example with FILTER modifier as following:

    1. SELECT
    2.  day,
    3.  COUNT(DISTINCT user_id) AS total_uv,
    4.  COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('android', 'iphone')) AS app_uv,
    5.  COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('wap', 'other')) AS web_uv
    6. GROUP BY day