Adopting Flink SQL: Implementation and Operational Experience

The Azar Matching Dev Team reviewed adopting Flink SQL to reduce the complexity and operational burden of a legacy Flink app that used 96 CPUs. The core match event join had already been separated into its own Flink app, so the most realistic direction was to move the subsequent conditional event publishing and Redis flag storage to an SQL-based approach.

Alternatives included merging everything back into a single Flink App, splitting into multiple apps, or using Flink SQL, and the team chose Flink SQL for productivity and operational efficiency. The rationale broke down into three main points.

• High Availability: Checkpoint and Savepoint, JobManager HA, and TaskManager redistribution enabled handling failures.
• Advanced streaming features: JOIN, UNION, window, event time, and watermark could all be handled via SQL.
• Extensibility: With UDF and Custom Connector, they built a Redis Connector directly, and supplemented ARRAY intersection, which the built-ins lacked.

Among the alternatives considered, ksqlDB had drawbacks in stateful processing: failover carried a heavy changelog replay burden, and since replicas perform the same computation, resource usage could double. Spark Structured Streaming had a good ecosystem and scalability, but its micro-batch nature introduced record-level latency, and with the team having almost no Spark experience, it was hard to choose given the need for a Custom Sink.

For the operating environment, they configured a Session mode Flink Cluster on top of Kubernetes. Unlike the existing Application mode, this structure submits jobs to an already-running cluster, so they added settings like high-availability.type: kubernetes, high-availability.storageDir: s3://..., and kubernetes.service-account, and had the JobManager use jobmanager.rpc.address=$(POD_IP) so each instance uses a different address, achieving HA.

For query deployment, instead of a dedicated UI, they used a GitOps approach combining GitHub Actions and the Flink SQL Gateway API. Job-specific folders and SQL files are managed in the repository, and a Python-written action reads the corresponding SQL and calls the Gateway REST API, making implementation and testing simple.

During operations, they also documented several representative failure patterns.

• JobManager / TaskManager fail: TaskManagers were sometimes restarted due to Kubernetes QoS policy, but work continued as tasks were redistributed to other TaskManagers.
• Malformed data ingestion: json.ignore-parse-errors was used to ignore JSON parsing errors, and JSON_VALUE was protected with DEFAULT ... ON ERROR.
• Insufficient resources: When TaskManager CPU exceeded 100% or memory ran low, they responded by adjusting resources and parallelism.
• Some jobs fail after cluster restart: Fixed an issue where timeout and retry settings were too short, causing retries to end too quickly.
• Query condition changes: Savepoint restoration is valid for simple condition changes, but if window conditions change, state compatibility breaks, and building a dedicated app directly may be better.

Monitoring was set up using Flink metric. numRunningJobs was used to quickly detect abnormal termination, and metrics like taskmanager.cpu.load, taskmanager.memory.used, busyTimeMsPerSecond, and Kafka's records-lag-max were used to track load and latency.

The appendix includes a HOP window example that receives login events from Kafka, aggregates the number of logins in the past 1 minute every 10 seconds, and outputs them back to Kafka. This example uses WATERMARK, HOP_START, and GROUP BY HOP(...) to show that a stateful streaming app can be written using only Flink SQL.