Shulou(Shulou.com)06/03 Report--

Author: Zhang Guanghui

This article will show you the whole process of how Byte Jump migrates Storm from Jstorm to Flink and its subsequent plans. You can learn about the background of the introduction of Flink by byte jump companies and the process of building Flink clusters. How does Byte Jump be compatible with previous Jstorm assignments and build a task management platform based on Flink? This article will unveil these mysteries for you one by one.

The content of this paper is as follows:

The background of introducing Flink

The Construction process of Flink Cluster

The background of constructing flow Management platform and introducing Flink

The following picture shows the business scenario of a byte-beating company.

Cdn.xitu.io/2019/4/25/16a5370e3c99fefa?w=942&h=486&f=jpeg&s=102343 ">

First of all, there are advertising, AB testing, push and data warehouse services in the application layer. Then, in the process of using J storm, a layer of templates is added, which is mainly applied to the computing model of storm, and the language is python. So there is a relatively abstract schema in the middle, running on top of the J storm computing engine at the bottom.

Byte Jump has many J-storm clusters, and in July 17, before planning to migrate to Flink, the J storm cluster was about the size shown in the figure below, with about 5000 machines at that time.

Next, let's introduce the whole process of migrating Flink. First of all, let's introduce in detail how J-Storm was used at that time.

Above is an example of word count: on the left is a directory structure under resources, where the Spout/Bolt logic is written by some python scripts. Then there is a topology_online.yaml configuration file on the outermost layer.

What is this configuration file for? Is to connect all the Spout and Bolt together to form a directed independent graph, that is, the DAG graph. This is the entire directory structure when using J storm, which is used by most users. On the right is the logic of Spout and Bolt, which is actually abstracted out of a function in which the business function is written, and then the tuple_batch, that is, the upstream data, is used to do some computational logic.

The following details of the configuration file information, in fact, we have the entire topology topology information, such as the name of the job, how many resources the job requires, how many work number. Will there be individual spout and Bolt configuration information, such as consumed topic or some concurrency?

In addition to this information, there is also the flow of the entire data stream, such as the output of spout, the message of output messsage, and so on. Finally, there is the entire shuffle logic from Spout to Bolt. This is the whole way we used Jstorm before. Finally, the contents of the entire directory structure will be parsed, the topology of the entire storm will be built according to the configuration file, and then submitted to the cluster to run.

What is the problem with using Jstorm clustering? The first problem, because we were writing code using python, there is no memory isolation for the whole cluster, and there is no memory limit between job and work. For example, in the actual process, we often encounter a user who may have a problem writing the code so that a work may take up 70 gigabytes of memory and take up 1x3 of the machine's memory. The second problem is that there is no expanded management between business teams, and there is no clue about the budget and review. At that time, we all ran in a large cluster, and then individual businesses ran separately in some small clusters, but every time we were short of resources, and we could not sort out the budget.

The third problem is that there are too many clusters, and the operation and maintenance platform is not very good, and they all rely on people for operation and maintenance. It is basically out of control when there are too many clusters at this time.

The fourth problem is that some of the code we write in python has poor performance. But it is also difficult for us to promote this Java on the basis of Storm, because some of our colleagues actually do not approve of Java, because he thinks that the development speed of java is too slow.

At that time, we wanted to solve the above problems. One idea was to put Jstorm on top of yarn and directly make Jstorm compatible with yarn. Later, because I knew that Ali was using Flink, I went to investigate Flink and found some advantages of Flink, so I wanted to try to solve the existing problems with Flink.

The first problem with Flink can be solved successfully, because the Flink job runs on top of yarn, which solves the problem of memory isolation. Then Yarn also supports queues, and we can divide queues according to business, so that our problem of expanding budget review is solved. We do not need to operate and maintain a cluster of our own, because we have yarn to manage our resources, which also saves the members of operation and maintenance. On this basis, we can also do some physical isolation queues, in fact, the physical isolation queues are also experiencing problems. Because the physical isolation queue only says that the machine is isolated, but the cabinet is not isolated and the network bandwidth is not isolated, so even physical isolation queues now encounter, for example, when sharing cabinets with offline jobs. The exit bandwidth of this cabinet is full. In order to solve these problems, we may want to do QOS on this offline cluster to solve this problem.

Flink is actually compatible with Storm, for example, previous history jobs can be migrated, and there is no need to maintain two computing engines. Flink supports some high-priority API such as SQL and windowing features including checkpoint. The demand for exactly-once in our headline business is not particularly strong.

These are the advantages of Flink, so we decided to migrate from J storm to Flink.

The Construction process of Flink Cluster

In the process of migration, the first thing is to set up the Flink cluster. At first, it must be the pursuit of stability, such as isolating offline yarn clusters, and then isolating name node and name space on Hdfs without relying on HDFS. Then we sort out the jobs on the original storm, which belong to different businesses, and then map them to different queues, and finally isolate some special queues. These are a few points we considered when preparing for this Fink cluster.

Let's consider how Flink is compatible with J storm and then migrate it.

We used version 1.32 of Flink at that time, because Flink has Flink-storm project, which can convert Storm jobs into Flink jobs, so we use these technologies for reference to implement a Flink- jstorm. Equivalent to transforming the topology of a J storm into a Flink job. It is not enough to do this, because we have a series of peripheral tools to align. For example, when submitting a job before, it is submitted through a script to ask the user to mask some other parameters. If we use flink, we also need to build such a script, then submit the Flink Job, and finally stop flink Job. The third point is to build flink job peripheral tools, automatically register alarms, such as consumption delay alarms, auto-register the Dashboard and some log service, all of which are aligned with the original service.

After the alignment, we need to build a migration script, and the most difficult part of the migration is the resource configuration. Because of how many resources Storm uses and how to match Storm, for migrating users, if they do it for the first time, they certainly don't understand these things. So we write such a script to help users generate the corresponding resource usage in its Flink cluster. When the work is done, we begin to migrate. Up to now, the overall migration has been completed, and about ten jobs have not been migrated yet. Now the size of the cluster has reached more than 6000.

We have some other optimizations during the migration. For example, J storm can support the restart of task and work dimensions, and Flink is not particularly good. We have made some optimizations in this respect to achieve a single task and single tm granularity restart, so as to solve part of the job restart due to task restart of the entire job.

Build a flow management platform

After the migration, we built a streaming management platform. The purpose of this platform is to solve some problems encountered in the actual process, such as the whole cluster can not determine which jobs are running on it, and can not notify specific users, and some user jobs do not know which jobs they have submitted. When we build streaming jobs, our goal is actually the same as other management platforms. for example, we provide some interface operations, and then provide a version management, which is to facilitate users to upgrade and rollback operations. we also provide an one-stop tool for checking problems: aggregating all the information users need on one page. Prevent users from constantly jumping around and avoid switching between different systems. There are some historical records before whether I ran on yarn or storm, one of my homework was kill, in fact, I do not know. In response to this problem, we provide some historical operational records of some goals.

When designing this management platform, we considered that providing such a front-end management platform may only be aimed at part of the products within the company, and other products have also made their own front-end. They can use a template to generate a storm task according to their own logic. Based on this, we abstract the entire management platform into two layers: the top layer is actually equivalent to a user-oriented or similar front-end product. The middle layer is actually a scheduling task similar to submitting a job. This layer is only responsible for picking up the task, then stopping the task, managing the life cycle, and pulling the job back up due to failure. This is what the middle tier TSS layer does.

In this way, we can dock to all front-end platforms. TSS communication through a RPC completely shields all underlying services from the underlying logic of interaction such as Filnk and Yarn as well as HDFS.

Next, it is relatively easy for the user to write a job, and the process is as follows:

The first step is for users to generate their own job template. We use the script rack provided by maven to generate the schema of some jobs. After this job is executed, it will help you prepare some porm files, as well as some routine components like kafkasource, and then you can fill in your main logic directly in this template. Because one of the most common problems we encounter when writing Java programs is the problem of package conflicts. Therefore, the porm file helps users to exclude some jar packets that may conflict, so that the probability of packet conflicts becomes smaller and smaller.

Our test job is basically tested in IDEA or local mode, and a script is provided to submit the job to the stage environment. Register the job on the platform before submitting it, and then add some configuration information.

The following is a code version management interface:

After submitting the entire assignment, it is shown in the following figure:

After submitting a job, the user may want to see how the job is running. We can show the user the running status of his job in four ways.

The first is Flink UI, which is officially available to UI users. The second is Dashboard, where we show the task dimension in the job, the network buffer between QPS and task, and these important information come together to create a Dashboard, which may be more convenient when looking up problems. The third is the error log, in fact, like everyone's idea, put a distributed log together and then write it to ES. The fourth is to make a tool for Jobtrace, that is, we match some common exceptions in Flink, and then directly give the user a guide to the use of wiki, telling users that, for example, your job OM needs to expand memory. As long as there is something wrong with the user's job, we match all known exceptions to the user.

Here is the kibana of ES:

This is the function of our Jobtrace, we match these common exceptions of Flink, each exception actually corresponds to a wiki and then let users solve their own problems.

Finally, I would like to share our short-term plan, the previous one is basically finished and tends to be stable, but now we have encountered some new problems. For example, the issue of resource utilization, because when a user submits a job, the user puts up a resource at will because he is not particularly sensitive to resources. Maybe he needs two CPU, but he mentions four CPU. We want to be able to monitor how many resources he needs through a tool, and then notify yarn to reset the resource. Is to dynamically adjust job resources and automatically reset them.

The second problem is to optimize the job restart speed. Many of our businesses monitor the stability of its business according to the indicators of flow calculation. if a job is restarted at the upstream, a group of people at the bottom receive a report that there are some problems online. The reason is that one of the upstream jobs is restarted. We want to minimize the restart interval or reboot seamlessly, which is a problem that needs to be explored in the next stage.

The fourth point: Flink SQL has just been launched, may need some energy to promote.

Finally, we hope to abstract more pattern operation models here, because we have some requirements such as kafka2ES,kafka2hdfs, can we abstract them into a schema, and then provide some services.

The above is the main content I share this time. I would like to thank the organizers and participants of Flink and our colleagues, because the above sharing was done by me and my colleagues.

For more information, please visit the Apache Flink Chinese Community website.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.