Over the past few years, Nexmark has been widely utilized for benchmarking stream processing. However, the existing runner cannot accurately evaluate the characteristics of stream computing. Unlike batch processing, stream processing handles an unlimited amount of data and maintains intermediate results in an internal state. The state is crucial for stateful stream processing. The state size is controlled by a time-to-live mechanism, and the overall throughput of the stream job stabilizes as the state size becomes steady. The most meaningful performance indicators should be collected after stabilization. The current benchmark runner runs each query from an empty cold start and collects metrics during the entire job run on a limited dataset. This approach is more akin to a batch streaming benchmark and does not accurately benchmark stateful stream computing due to the following reasons:

1. Performance Degradation During Benchmark: As the job runs, the state initially fits in memory and then spills to disk, causing performance to drop greatly. In stream processing for unbounded data, the stabilized (or relatively stabilized) performance over a long run is more meaningful, while the degrading performance should not dominate the benchmark results.
2. Lack of Fault Tolerance Evaluation: Fault tolerance is crucial for unlimited data processing. Checkpoints consume CPU time or IO bandwidth to snapshot the current state, which may affect the processing performance itself. Fast checkpointing and recovery are expected to minimize the impact of service outages. However current runner does not evaluate the checkpointing and recovery.
3. CPU Efficiency (Throughput per Core Time Unit) is Not the Best Indicator: In most scenarios, throughput under limited resources is more meaningful. If we only consider CPU consumption per record, serial processing without any framework overhead would achieve the best score, but it seems pointless.

To address these issues, we plan to make changes to the current Nexmark runner and setup. The entire pipeline for each query will consist of two phases:

• Initialization: Execute 75% of the dataset to fill the internal state.
• Benchmark: Process the last 25% of the dataset. The state size is relatively consistent, ideally only 20% above the baseline size, which is similar to the production environment where the TTL controls the state size. Benchmark results will be generated based on the collected metrics of this phase.

The default dataset size is also proposed to be enlarged. We have chosen 200 million as the new default size. In this case, several queries running in Flink will reach about 3GB state size, which is an IO-intensive scenario.
We also propose another indicator for benchmark results. From the user's perspective, the most relevant metric is the cost associated with processing a fixed size of data. Our primary focus is on sustained performance once the process has stabilized. Since machine resources typically remain constant during steady processing, cost can be represented as resource usage over time. Therefore, the recommended scoring formula will be:

$Score = \frac{Record Count}{Price} = \frac{Record Count}{Resource\times time} = \frac{Throughput}{Resource}$

Since the resources are not comparable across different setups, we will provide a recommended configuration for Nexmark as we have done previously. This way, throughput will be the sole variable influencing the score. In a fixed setup, higher throughput indicates better performance of the stream processing system. Additionally, we will take the checkpoint and recovery time as another dimension of evaluation indicators, indicating the speed of fault tolerance. We also plan to set the checkpoint interval to 30 seconds, particularly for Flink. This interval is a balanced compromise to account for both fault tolerance needs and performance. Such a frequency makes the overhead associated with checkpoints, which is an essential aspect of the process, adequately reflected in the final results.

To sum up, we propose to apply the following change to current runner:

• Introduce two phase for each query: Initialization and Benchmark phase, processing 75% and 25% of the dataset respectively. A checkpoint and a restore are performed between the two phases.
• Enlarge the default dataset to 200 million.
• The checkpoint interval is set to at most 30 seconds.
• Take throughput as the main performance indicator in a fixed setup.
• Evaluate the checkpoint and recovery time as secondary indicators.

Thus we believe the new Nexmark runner will evaluate the streaming processing in a better way.