Flink Meetup - Real-Time User Behavior Analytics with Flink

This is a talk I gave at last year’s Beijing Flink Meetup. The video is available here. It covers GrowingIO’s work and lessons learned while moving its real-time stack from Spark to Flink.

The presentation included GrowingIO-specific terminology and use cases, so I have removed some nontechnical slides here.

The slide introduces GrowingIO’s business. GrowingIO uses automatic event capture to collect user behavior comprehensively, then helps companies analyze data and improve products through big-data tools and visualizations. This brings not only enormous data volumes, but also cost and efficiency challenges for data engineers.

This simplified 1.0 architecture uses Kafka as the source and Spark Streaming as the engine. Redis stores user and event counts, taking advantage of HyperLogLog. The left side shows charts from the GrowingIO platform.

Most 1.0 data supported only basic counting and monitoring, without analytical functions such as dimensional breakdowns. We wanted, for example, to observe campaign results live and adjust spending across channels, or reorder homepage product recommendations and remove unpopular items to increase overall orders.

Besides unmet product requirements, Spark Streaming had limitations for truly real-time applications. We encountered several problems:

• Excess resource consumption: We used ten-second batches for near-real-time latency, but input volumes varied. Small batches left resources idle, while large batches introduced delays. Idle capacity should process the next arriving data. Flink’s continuous streaming avoids fixed intervals and substantially smooths latency peaks.
• Execution model: Spark Streaming submits a batch after every interval, with synchronous map/reduce stages. Repeated scheduling and execution-plan generation add unnecessary overhead.
• Limited processing control: Different customers define real time differently, but Spark Streaming cannot easily offer different latency targets. Flink triggers customize output frequencies by data, improving flexibility while saving resources.

Version 2.0 uses Apache Flink. Sources include Kafka and a command stream, which mainly:

• Immediately updates computation logic when users modify charts.
• Controls trigger frequency and commands.

The Process stage performs ETL on source data. Custom triggers and windows then process it before writing results to HBase aggregate tables.

Watermark calculation is affected by source skew. If Kafka data is skewed, operators may receive very different event times. We use the 99th percentile of collected event times, ordered from smallest to largest, as the watermark reference.

We found that watermarks were not checkpointed. Restoring could therefore accept late data incorrectly and modify historical results that should remain fixed. See FLINK-5601. Our solution makes WatermarkOperator stateful, uses UnionState to recover the minimum watermark, and passes it downstream during open initialization. We submitted this solution to the community.

Savepoints and checkpoints persist state by serializing objects. Recovery therefore requires the new program’s objects to be compatible with the old serialized objects. Sometimes recovery points are unavailable or unusable:

• First startup: No persisted state exists.
• State-related classes change, such as a window aggregator’s data structure.
• Savepoint files are damaged or accidentally deleted, or disks fail.
• The job’s MaxParallelism changes, which Flink validates during recovery.

We investigated three approaches:

1. Use ConnectedStream.

Connect restored state with the live stream. During startup, send state records into windows to initialize WindowState, and do similarly for other operators. This complicates processing: one source is empty most of the time, and the program must handle aggregate state and live detailed records very differently.

2. Use bravo.

This community project uses Flink’s batch API to read savepoint/checkpoint files and deserialize them with logic matching Flink’s source. It currently supports only RocksDBStateBackend. I tried extending it with HeapStateBackend serialization and deserialization, but its code structure made that harder than expected.

3. Use QueryableState.

Flink’s experimental QueryableState feature reads specified state from an operator. We extended this read mechanism with writes, allowing APIs to inject state into running operators. The tradeoff is thread safety: state is not inherently thread-safe, and coordinating access introduces performance overhead.

Looking Ahead

Data generally becomes less valuable with age: earlier access makes it more useful. Behavior analytics will extend beyond dimensional analysis and audience selection, with increasingly complex strategies for specific scenarios. We plan to explore CEP. Oracle already defines CEP SQL syntax, and Calcite supports much of it in parsing and logical planning. Flink CEP still falls short of our needs:

• Insufficiently expressive syntax
• No simultaneous matching of multiple rules
• No dynamic rule changes
• Excessive memory consumption

Our next step is to address these issues in Flink CEP and build a behavior-analysis engine that enables new real-time analytical tools.