Implementing distributed Flink SQL gateway based on Kyuubi

#sql #opensource

Apache Kyuubi [1] is a distributed multi-tenant SQL gateway. Its main function is to accept SQL submitted by users through protocols such as JDBC/REST and route it to the SQL engine managed by it for execution according to the multi-tenant isolation policy. In the latest Kyuubi 1.8 version, Kyuubi Flink Engine migrates to the Flink SQL Gateway (hereinafter referred to as FSG) API and supports the Flink Application mode, which allows us to quickly deploy a production-ready distributed Flink SQL gateway with Kyuubi.

Why you need Kyuubi

I believe the first question that many readers will think of is, Flink already provides SQL Gateway, why do we need to introduce Kyuubi? In fact, the key answer lies in Kyuubi’s description: one is distributed and the other is multi-tenant, the two complement each other. The gateway load capacity needs to be expanded horizontally, so it will naturally evolve to distributed; in a distributed environment, the affinity of sessions needs to be ensured, so the need for multi-tenant routing naturally arises; and multi-tenants have higher requirements for isolation, so inversely It will also promote the development of distributed architecture.

The combination of SQL Gateway and SQL Client can be used out of the box and help us quickly run through the demo. However, in an enterprise production environment, a single instance is often difficult to meet user requests. In addition, different business users often have different Flink versions, built-in dependencies, and resource configurations. demand, which resulted in us having to maintain multiple SQL Gateway instances. What’s even more troublesome is that an instance may be shared among multiple small users, or it may be exclusive to a large user. There is a many-to-many mapping relationship between users and SQL Gateway instances. In addition, high availability and load balancing are also difficult problems that have to be faced. Various problems have caused the operation and maintenance management cost of SQL Gateway to remain high, and the emergence of Kyuubi has solved this problem to a great extent.

Basic principles of Kyuubi

Kyuubi mainly has three modules: Client, Server and Engine.

Kyuubi Client is relatively simple. There are out-of-the-box clients like kyuubi-beeline, and you can also choose open tools or protocols like DBeaver or JDBC/REST.
Kyuubi Server is the core module and its main responsibilities are:
Provide Client with Frontend including multiple protocols and accept Connction.
Manage the Engine lifecycle and route user requests to the appropriate Engine.
Manage Session/Operation status.
Kyuubi Engine is the module that carries the actual computing load. It is responsible for translating Kyuubi Server requests into engine native operations and supports multiple computing engines such as Spark / Flink / Trino.

Many readers may have discovered that the positioning of FSG is similar to Kyuubi’s Flink Engine (in fact, Kyuubi Flink Engine does embed an FSG), and the abstraction layer of Kyuubi Server is the key to solving distributed and multi-tenant problems.

In the Kyuubi architecture, the communication between Client and Server, Server and Engine all have service discovery and load balancing, which means that different modules are very loosely coupled. Client can connect to any Kyuubi Server in the same Namespace, and Kyuubi Server can also access any Engine under the same Namespace. For stateful short connections in Batch scenarios, Kyuubi will persist relevant states to the database and ensure that the connection can always fall to the same server through forwarding between servers. This design gives Kyuubi excellent horizontal expansion capabilities.

As mentioned above, routing of multi-tenant requests is the core function of Kyuubi Server. Kyuubi Server provides different levels of Engine Share Level to meet the isolation needs of multi-tenants, and selects the Engine corresponding to the request based on the Engine Share Level.

In addition, the life cycle of Engine is automatically controlled by Server. If there is currently no suitable Engine, the Server will start an Engine; and if an Engine is idle for more than a certain period of time, the Engine will automatically shut down to save resources.

Additional benefits

In addition to the above-mentioned basic advantages, Kyuubi also has additional benefits compared to using FSG directly. There are short-term differences caused by different development progress, and there are also long-term differences that may exist due to different project positioning.

Application deployment mode

Kyuubi supports Flink on YARN Application mode in the latest 1.8 version, and FSG’s Application mode is still in the discussion stage (see FLIP-316 [2]). It is worth noting that the two implementations of the Application pattern are not the same in the long run.

To understand the differences behind it, first briefly review the basics of Flink SQL. When we execute a Flink SQL, we will go through the following stages:

Parsing: The SQL submitted by the user will first be parsed into a logical execution plan by Parser.
Optimization: The logical execution plan is optimized by Planner Optimizer and a physical execution plan will be generated.
Compile: The physical execution plan is then used to generate Java code (DataStream Transformation) through Planner CodeGen code to build JobGraph
Execution: Submit JobGraph to JobManager, which applies TaskManager to execute the job

For Flink Session mode or Per-Job mode, the first three steps are completed in the TableEnvironment of Flink Client, and the compiled JobGraph can be serialized and contains all job information, so it is very suitable for dividing SQL gateway and Flink cluster. boundaries. In other words, after the gateway completes SQL parsing, optimization, and compilation, it can REST or submit the JobGraph to distributed storage such as HDFS/S3.

However, the Application pattern poses new architectural challenges. The construction of JobGraph in Application mode must be done on the JobManager side, which means compilation also needs to be done on the JobManager side. To do this, we need to serialize the physical execution plan generated in step 2 and upload it to the JobManager, and this serialized object is Json Plan [3].

Json Plan was introduced in Flink 1.14. It is a serialized representation of Flink’s physical execution plan ExecNodeGraph. It was originally designed for cross-version upgrades of Flink SQL. Now it is also very convenient for communication between FSG and JobManager out of the box. FLIP-316, FSG’s proposal to support Application mode, is designed based on Json Plan.

However, Json Plan currently has certain limitations. The biggest limitation is that Json Plan only supports INSERT statements and STATEMENT SET statements in Streaming mode. This results in the Application mode being temporarily unable to support Batch mode and SELECT, limiting data exploration scenarios.

In contrast, the Application mode of Kyuubi Flink Engine adopts an architecture similar to the Spark Cluster mode. SQL parsing, optimization, compilation and execution are completed directly in the TableEnvironment on the JobManager side. Job information does not need to be transferred across processes, so it is not restricted by Json Plan. Through Kyuubi, we can use Flink to do arbitrary data exploration just like using Spark.

Observability

In enterprise-level applications, observability is undoubtedly an important basic requirement. Kyuubi provides a rich set of Metrics and Reporters, as well as a Web UI embedded in Kyuubi Server, allowing us to keep track of the gateway’s load at any time.

In terms of metrics, Kyuubi Metrics can be divided into several categories:

Connection Metric: Monitor the number of Connections in different states
Operation Metric: Monitor the number of Operations in different states and their execution time
Thread pool Metric: Monitor whether the Kyuubi Server service thread pool is sufficient
Engine Metric: Monitor the number of Engines in different states of different users
RPC call Metric: Monitor the frequency and time of different RPC calls between Kyuubi Server and Engine

In terms of reports, Kyuubi provides the most popular Prometheus Reporter and JMX Reporter. If you need log-based metrics, you can also choose SLF4J Reporter, JSON Reporter or Console Reporter.

Future outlook

From Spark-specific to multi-engine support, Kyuubi is moving step by step towards the goal of “a standard solution for SQL Gateway”. In terms of Flink Engine support, Kyuubi will further focus on complementing K8s and camouflage authentication capabilities in the future to improve users’ experience of using Flink SQL in different environments.

About the writer

Paul Lin, Apache Kyuubi Committer, currently works in NetEase Games, Technology Center Data and Platform Services Department, responsible for Flink basic components and Streamfly real-time computing platform. Before joining Kyuubi, he was mainly active in the Flink community.