CN111767092B - Job Execution Method, Device, System, and Computer-Readable Storage Medium - Google Patents

Abstract

The invention relates to the technical field of financial science and technology, and discloses a job execution method, a job execution device, a job execution system and a computer readable storage medium. The job execution method includes: when an execution request of Spark job is received, acquiring a version number, dynamic configuration parameters and Spark job codes of a target Spark engine according to the execution request; determining deployment catalog information and version loading rules of a target Spark engine according to the version number; acquiring static configuration parameters according to the deployment catalog information, and initializing a target Spark engine by using the dynamic configuration parameters and the static configuration parameters according to the version loading rules so as to start the target Spark engine; the Spark job code is submitted to the target Spark engine to execute the job. The invention can realize that Spark operation of multiple versions is supported in one Linkis service cluster at the same time, and reduces operation and maintenance cost.

Description

Job Execution Method, Device, System, and Computer-Readable Storage Medium

technical field

The present invention relates to the technical field of financial technology (Fintech), in particular to a job execution method, device, system and computer-readable storage medium.

Background technique

With the development of computer technology, more and more technologies are applied in the financial field. The traditional financial industry is gradually transforming into financial technology (Fintech). Spark technology is no exception. However, due to the security and real-time requirements of the financial industry, it is also Higher requirements are put forward for Spark technology.

In existing big data cluster applications, the big data processing environment in financial institutions (such as banks) is relatively centralized, and the data is also very concentrated and the amount of data is also very large. Due to the centralized processing of data, many big data application components are also deployed centrally, and the versions of these components are updated frequently, with several versions updated every year. engine).

Each version update of Spark will bring many new features, but sometimes it is not well compatible with the old version of the job, and there are often a large number of users in an environment, some users need to use the new version of Spark, and some Many users need an old version of Spark, but the current Linkis (a data middleware that connects multiple computing and storage engines) service clusters can only support one version of Spark jobs, and cannot integrate multiple versions of Spark to run at the same time. Therefore, in order to meet the multi-version requirements of these users, it is usually necessary to deploy multiple sets of Linkis clusters, which requires a large number of machines to deploy different versions of Spark Driver. At the same time, user jobs also need to be switched between different environments. A set of Linkis increases the cost and difficulty of operation and maintenance, and it is also difficult for users to maintain the use of each environment.

Contents of the invention

The main purpose of the present invention is to provide a job execution method, device, system, and computer-readable storage medium, aiming at simultaneously supporting multiple versions of Spark job operation in a Linkis service cluster and reducing operation and maintenance costs.

In order to achieve the above object, the present invention provides a job execution method, the job execution method includes:

When receiving the execution request of the Spark job, obtain the version number, dynamic configuration parameters and the Spark job code of the target Spark engine according to the execution request;

Determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

Acquire static configuration parameters according to the deployment directory information, use the dynamic configuration parameters and the static configuration parameters to initialize the target Spark engine according to the version loading rules, to start the target Spark engine;

Submit the Spark job code to the target Spark engine to execute the job.

Optionally, before the step of determining the deployment directory information and version loading rules of the target Spark engine according to the version number, it also includes:

Obtain the user identification corresponding to the execution request, and detect whether there is an idle Spark engine corresponding to the user identification and the version number;

If it does not exist, then execute the step: determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

If it exists, submit the Spark job code to the idle Spark engine to execute the job.

Optionally, before the step of determining the deployment directory information and version loading rules of the target Spark engine according to the version number, it also includes:

Acquiring the user ID corresponding to the execution request, and judging whether the user is in the preset grayscale list according to the user ID;

If the user is not in the default grayscale list, then perform the step of: determining the deployment directory information and version loading rules of the target Spark engine according to the version number;

If the user is in the preset grayscale list, create a grayscale Spark engine, and submit the Spark job code to the grayscale Spark engine to execute the job.

Optionally, the job execution method further includes:

In the initialization process, determine the target calling method according to the version number and the preset abstraction layer interface;

The file package on which the target Spark engine depends is loaded under the directory corresponding to the deployment directory information according to the target calling method.

Optionally, before the step of submitting the Spark job code to the target Spark engine to execute the job, it also includes:

Modify the Spark job code according to the version number;

The step of submitting the Spark job code to the target Spark engine to execute the job includes:

Submit the modified Spark job code to the target Spark engine to execute the job.

Optionally, the step of submitting the Spark job code to the target Spark engine to execute the job includes:

Submit the Spark job code to the driver node of the target Spark engine;

Converting the Spark job code through the driver node to obtain a Spark task;

The Spark task is assigned to the executor nodes deployed on the Yarn cluster to execute the job.

Optionally, before the step of converting the Spark job code through the driver node to obtain the Spark task, it also includes:

In the initialization process, when the Scala interpreter is created in the driver node of the target Spark engine, the class loader of the main thread is injected into the Scala interpreter, so that the class loader of the main thread becomes the Scala interpreter The parent of the class loader, and make the Scala interpreter create a corresponding class loader according to the parent's class loader;

The step of converting the Spark job code through the driver node to obtain the Spark task includes:

The Spark job code is transformed by the class loader of the Scala interpreter created in the driver node to obtain the Spark task;

After the step of assigning the Spark task to the executor node deployed on the Yarn cluster to execute the job, it also includes:

When receiving the serialized execution result returned by the executor node based on the Spark task, modify the class loader of the current thread of the target Spark engine to the class loader of the Scala interpreter to pass the Scala The class loader of the interpreter deserializes the serialized execution result.

In addition, in order to achieve the above purpose, the present invention also provides a job execution device, the job execution device includes:

The first obtaining module is used to obtain the version number, dynamic configuration parameters and Spark job code of the target Spark engine according to the execution request when receiving the execution request of the Spark job;

The first determining module is used to determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

An engine initialization module, configured to obtain static configuration parameters according to the deployment directory information, use the dynamic configuration parameters and the static configuration parameters to initialize the target Spark engine according to the version loading rules, to start the target Spark engine;

A job execution module, configured to submit the Spark job code to the target Spark engine to execute the job.

In addition, in order to achieve the above object, the present invention also provides a job execution system, the job execution system includes: a memory, a processor, and a job execution program stored in the memory and operable on the processor, the When the job execution program is executed by the processor, the above steps of the job execution method are realized.

In addition, in order to achieve the above object, the present invention also provides a computer-readable storage medium, on which a job execution program is stored, and when the job execution program is executed by a processor, the above-mentioned job execution can be realized. method steps.

The present invention provides a job execution method, device, system, and computer-readable storage medium. When receiving an execution request of a Spark job, the version number, dynamic configuration parameters, and Spark job code of the target Spark engine are obtained according to the execution request; and then, Determine the deployment directory information and version loading rules of the target Spark engine according to the version number; obtain static configuration parameters according to the deployment directory information, and use the dynamic configuration parameters and static configuration parameters to initialize the target Spark engine according to the version loading rules to start the target Spark engine; Then submit the Spark job code to the target Spark engine to execute the job. In the present invention, by pre-installing and deploying dependent jab packages that are prone to conflicts in different versions of Spark, when receiving an execution request for a Spark job, obtain the version number in the request to determine the corresponding deployment directory information, and obtain the corresponding The startup parameters (including dynamic configuration parameters and static configuration parameters), and then initialize the target version of the Spark engine according to the startup parameters to start the target version of the Spark engine, which can realize the dynamic loading of different versions of Spark dependent jar packages, and avoid multiple The version jar conflict problem makes it possible for multiple versions of Spark to be executed in parallel under the same Linkis cluster. Therefore, the present invention only needs to deploy a set of Linkis service clusters, and different versions of Spark engines can be created in the Linkis service clusters to support the operation of multi-version Spark jobs. Compared with the prior art by deploying multiple sets of Linkis clusters By deploying different versions of the Spark engine, the present invention can greatly reduce machine resources and operation and maintenance costs.

Description of drawings

FIG. 1 is a schematic diagram of the device structure of the hardware operating environment involved in the solution of the embodiment of the present invention;

FIG. 2 is a schematic flow chart of the first embodiment of the job execution method of the present invention;

3 is a schematic diagram of the system architecture of the job execution system of the present invention;

Fig. 4 is a schematic diagram of the functional modules of the first embodiment of the job execution device of the present invention.

The realization of the purpose of the present invention, functional characteristics and advantages will be further described in conjunction with the embodiments and with reference to the accompanying drawings.

Detailed ways

It should be understood that the specific embodiments described here are only used to explain the present invention, not to limit the present invention.

Referring to FIG. 1 , FIG. 1 is a schematic diagram of the device structure of the hardware operating environment involved in the solution of the embodiment of the present invention.

The job execution device in the embodiment of the present invention may be a server, or a terminal device such as a PC (Personal Computer, personal computer), a tablet computer, or a portable computer.

As shown in FIG. 1 , the job execution device may include: a processor 1001 , such as a CPU, a communication bus 1002 , a user interface 1003 , a network interface 1004 , and a memory 1005 . Wherein, the communication bus 1002 is used to realize connection and communication between these components. The user interface 1003 may include a display screen (Display), an input unit such as a keyboard (Keyboard), and the optional user interface 1003 may also include a standard wired interface and a wireless interface. Optionally, the network interface 1004 may include a standard wired interface and a wireless interface (such as a Wi-Fi interface). The memory 1005 can be a high-speed RAM memory, or a stable memory (non-volatile memory), such as a disk memory. Optionally, the memory 1005 may also be a storage device independent of the aforementioned processor 1001 .

Those skilled in the art can understand that the structure of the job execution equipment shown in Figure 1 does not constitute a limitation on the job execution equipment, and may include more or less components than those shown in the illustration, or combine certain components, or different components layout.

As shown in FIG. 1 , the memory 1005 as a computer storage medium may include an operating system, a network communication module, and a job execution program.

In the terminal shown in Figure 1, the network interface 1004 is mainly used to connect to the background server and perform data communication with the background server; the user interface 1003 is mainly used to connect to the client and perform data communication with the client; and the processor 1001 can be used for The job execution program stored in the memory 1005 is called, and each step of the following job execution method is executed.

Based on the above hardware structure, various embodiments of the job execution method of the present invention are proposed.

The invention provides a job execution method.

Referring to FIG. 2 , FIG. 2 is a schematic flowchart of a first embodiment of a job execution method according to the present invention.

In this embodiment, the job execution method includes:

Step S10, when receiving the execution request of the Spark job, obtain the version number, dynamic configuration parameters and Spark job code of the target Spark engine according to the execution request;

The job execution method of this embodiment is applied to a job execution system. As shown in FIG. 3 , the job execution system includes a Linkis cluster service and a Yarn cluster, wherein the Linkis cluster service includes an external service portal and an engine management service, and the external service portal includes Including Job job management module and engine manager. Specifically, compared to the existing job execution system, the engine management service has been transformed, adding a dynamic loading scheme for multi-version spark dependent jars, and dynamic adjustment of the class loader to solve the parallel execution of spark jobs of different versions question. In addition, a version identifier is added to the execution request parameters of the Spark job, and an engine manager is added correspondingly, which is used to manage Spark engine services of different versions and different users, and is also used to create or select an engine according to the version number.

Among them, the Linkis cluster service is a multi-computing and storage engine, which provides a unified Restful (a design style and development method for network applications) interface, and submits and executes SQL (Structured Query Language), PySpark (Spark API for Python developers), HiveQL (HiveQuery Language, a query language for data warehouse tools), Scala (a multi-paradigm programming language) and other script data middleware. The job management module in the external service portal is used to receive the execution request of the Spark job, and is also used to automatically adjust the error code of the Spark job. The engine manager is used to manage Spark engine services of different versions and users, and is also used to create or select an engine based on the version number. The engine management service is used to manage the process creation, status tracking, and process destruction of the Spark Context (Spark context content). The engine management service includes one or more Spark engines, and multiple Spark engines can be executed in parallel. The Spark engine passes RPC (Remote Procedure Call, remote procedure call) communicates with the external service portal. Yarn cluster provides a framework for job scheduling and cluster resource management in big data platforms.

In this embodiment, when the external service portal receives the execution request of the Spark job, it obtains the version number of the target Spark engine, the dynamic configuration parameters and the Spark job code according to the execution request. Among them, the dynamic configuration parameters include the execution number of executor nodes (Executor), the number of CPUs (Central Processing Unit, central processing unit), the size of memory, and the number of CPUs of driver nodes (Driver), the size of memory or one or more Multiple.

Step S20, determining the deployment directory information and version loading rules of the target Spark engine according to the version number;

Then, determine the deployment directory information and version loading rules of the target Spark engine according to the version number. Among them, the deployment directory information includes root directory deployment information and configuration directory deployment information, and the version loading rules include loading rules for different Spark versions dependent libraries.

It should be noted that during implementation, the root directory of the Spark service needs to be established in advance on the machine where the engine management service is located, and then the Spark_home directory and the Spark_conf_dir directory are created in the root directory according to different Spark version numbers. When installing and deploying Spark, Deploy according to this path rule. Specifically, set subdirectories of each version in the Spark_home directory and Spark_conf_dir directory, and install Spark to the corresponding subdirectories accordingly. It should be noted that during installation, the dependent jab packages that are prone to conflicts in different versions of Spark need to be deployed separately. Among them, the Spark_home directory is the root directory, which is the installation directory of Spark. The Spark_conf_dir directory includes static configuration parameters of different versions of Spark, such as application property parameters, operating environment parameters, runtime behavior parameters, and network parameters.

In addition, it should be noted that before installing and deploying different versions of Spark, in order to support multiple versions of the Spark engine itself, you can first compare the API interfaces of different versions of the Spark package to obtain the difference API. The API compiles different versions of the Spark package, and then deploys the compiled Spark package to the corresponding directory. That is, the Spark package is compiled first, and the changed parts of the API (Application Programming Interface, application program interface) are separated, and then the compiled Spark package is installed and deployed. For example, for APIs related to Vector (vector), we can separate two projects, each project is adapted to the corresponding version, and then release the jar package, and in Maven (Maven, project object model) according to the Profile (user configuration file ) mechanism to introduce different adaptation packages according to the Spark version.

Step S30, obtaining static configuration parameters according to the deployment directory information, using the dynamic configuration parameters and the static configuration parameters to initialize the target Spark engine according to the version loading rules, so as to start the target Spark engine;

Then, obtain the static configuration parameters according to the deployment directory information, specifically, obtain the static configuration parameters in the Spark_conf_dir directory, such as application attribute parameters, operating environment parameters, runtime behavior parameters, network parameters, etc. Furthermore, the target Spark engine is initialized with startup parameters (including dynamic configuration parameters and static configuration parameters) according to the version loading rule, so as to start the target Spark engine. That is, the dynamic configuration parameters and the static configuration parameters are correspondingly filled into the configuration parameters of the configuration file of the target Spark engine for initialization.

In addition, it should be noted that during the initialization process, the archives (parameter values) used by pySpark (the API provided by Spark for Python developers) also need to be uploaded to the corresponding hdfs (Hadoop Distributed File System, distributed file system) according to different versions. file system) directory, it is convenient for the executor (executor node) of the Spark engine to download these third-party files from the correct location to solve the Jar problem that the Executor depends on when starting.

Further, the job execution method also includes:

Step a1, during the initialization process, determine the target calling method according to the version number and the preset abstraction layer interface;

Step a2, according to the target calling method, load the file package that the target Spark engine depends on under the directory corresponding to the deployment directory information.

Furthermore, since the Spark engine service of Linkis has some dependencies on Spark at runtime, these dependencies are currently placed in a lib directory in the root directory in the form of jar packages. When there are different Spark versions, these jars will conflict. Therefore, these dependent jars need to be extracted in advance, not all loaded when the application starts, but according to different engine versions, in the Spark engine service Add a layer of abstract interface definition to realize multi-version dependent abstraction layer, make the underlying package that resolves the jar conflict into a version module (that is, the dependent file package below), and only load the specified version module when creating the engine to avoid multi-version jar conflict problem.

Specifically, during the initialization process, the target invocation method is determined according to the version number and the preset abstraction layer interface, and then, according to the target invocation method, the file package that the target Spark engine depends on under the directory corresponding to the deployment directory information is loaded, so that no Conflicting dependencies.

Step S40, submit the Spark job code to the target Spark engine to execute the job.

Finally, submit the Spark job code to the target Spark engine to execute the job. Specifically, the Spark job code is submitted to the driver node (Driver) of the target Spark engine, and then the Spark job code is converted by the Driver to obtain a Spark task (task). The specific conversion process can refer to the prior art; finally, the Spark tasks are assigned to executor nodes (Executor) deployed on the Yarn cluster to execute jobs.

The embodiment of the present invention provides a kind of job execution method, when receiving the execution request of Spark job, obtain the version number of target Spark engine, dynamic configuration parameter and Spark job code according to execution request; Then, determine the version number of target Spark engine according to version number Deploy directory information and version loading rules; obtain static configuration parameters according to the deployment directory information, use dynamic configuration parameters and static configuration parameters to initialize the target Spark engine according to the version loading rules to start the target Spark engine; and then submit the Spark job code to the target Spark engine to execute jobs. In the embodiment of the present invention, by pre-installing and deploying dependent jab packages that are prone to conflicts in different versions of Spark, when receiving an execution request for a Spark job, obtain the version number in the request to determine the corresponding deployment directory information, and Obtain the corresponding startup parameters (including dynamic configuration parameters and static configuration parameters), and then initialize the Spark engine of the target version according to the startup parameters to start the Spark engine of the target version, which can realize dynamic loading of different versions of Spark dependent jar packages, avoiding The problem of multi-version jar conflicts occurs, so that multi-version Spark can be executed in parallel under the same Linkis cluster. Therefore, the present invention only needs to deploy a set of Linkis service clusters, and different versions of Spark engines can be created in the Linkis service clusters to support the operation of multi-version Spark jobs. Compared with the prior art by deploying multiple sets of Linkis clusters By deploying different versions of the Spark engine, the present invention can greatly reduce machine resources and operation and maintenance costs.

Furthermore, since it takes a long time to start a Spark engine, and once it is started successfully, part of the cluster computing resources will be locked. When a user's job is completed, the engine that is already running will not be terminated immediately. Instead, the user's next job will be executed immediately to improve user experience. Engine reuse saves time and computing resources to a large extent. However, under the Session (session control) mode management of existing Linkis users, when reusing existing Spark engines, jobs are randomly submitted to idle Spark engines. of. When different versions of the Spark engine exist in an environment, part of the user's jobs will be submitted to the old version of the Spark engine, while some jobs will be submitted to the new version of the Spark engine, resulting in job execution failures.

In this regard, based on the first embodiment above, a second embodiment of the job execution method of the present invention is proposed.

In this embodiment, before the above step S20, the job execution method further includes:

Step A, obtaining the user ID corresponding to the execution request, and detecting whether there is an idle Spark engine corresponding to the user ID and the version number;

In this embodiment, when the external service portal obtains the execution request of the Spark job, it can also obtain the user ID corresponding to the execution request while obtaining the version number of the target Spark engine, the dynamic configuration parameters and the Spark job code, where , the user ID can be a user name or user ID, etc. Then, detect whether there is an idle Spark engine corresponding to the user ID and version number. During detection, the version number and user of the currently idle Spark engine can be obtained from the engine manager, and then matched with the obtained user ID and version number.

If it does not exist, then perform step S20: determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

If there is no idle Spark engine corresponding to the user ID and version number, a new Spark engine needs to be created. At this time, the deployment directory information and version loading rules of the target Spark engine are determined according to the version number, and then the next steps are executed. Specifically For the execution process, reference may be made to the above-mentioned first embodiment, which will not be described in detail this time.

If it exists, execute step B: submit the Spark job code to the idle Spark engine to execute the job.

If there is an idle Spark engine corresponding to the user ID and version number, directly submit the Spark job code to the idle Spark engine corresponding to the user ID and version number to execute the job.

In this embodiment, a version label (i.e. version number) is added to the execution request of the Spark job, and at the same time, the idle Spark engine corresponding to different users and version numbers is managed through the engine manager. After the external service entry of Linkis receives the execution request, By detecting whether there is an idle Spark engine corresponding to the user ID and version number, and then automatically submitting it to the idle Spark engine of the corresponding version according to the version label of the job, it can prevent Spark jobs from being randomly submitted to different versions while realizing engine reuse. If the Spark engine is used, the job execution fails.

Furthermore, when the Spark version needs to be updated, Linkis needs to kill (kill) all the running Spark engines first, then stop the existing Spark engine management services, update the configuration files of all Spark engine management servers, and update all Spark engines. The dependent Jar (a software package file format) package used by the engine management service has a great impact on the business, and the user's Spark job cannot be executed during the update period. In addition, after the update, there is a risk that some Spark jobs cannot be executed correctly under the new version of the Spark engine.

In this regard, based on the first embodiment above, a third embodiment of the job execution method of the present invention is proposed.

In this embodiment, before the above step S20, the job execution method further includes:

Step C, obtaining the user ID corresponding to the execution request, and judging whether the user is in the preset grayscale list according to the user ID;

In this embodiment, when the external service portal obtains the execution request of the Spark job, it can also obtain the user ID corresponding to the execution request while obtaining the version number of the target Spark engine, the dynamic configuration parameters and the Spark job code, where , the user ID can be a user name or user ID, etc. Then, judge whether the user is in the preset grayscale list according to the user identifier. Among them, the preset grayscale list is pre-set, and is used to specify some users to perform grayscale, that is, submit the role of the specified part of users to the new version of the Spark engine under the newly deployed Linkis service cluster for execution. In actual implementation, according to the success rate of the specified user's job execution on the new version of the Spark engine, the number of grayscale users specified in the preset grayscale list can be adjusted accordingly, so that the user's Spark job can be gradually migrated to the newly deployed Linkis Execute on the new version of the Spark engine under the service cluster.

If the user is not in the preset grayscale list, then perform step S20: determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

If the user is in the preset grayscale list, perform step D: create a grayscale Spark engine, and submit the Spark job code to the grayscale Spark engine to execute the job.

If the user is not in the preset grayscale list, the user does not belong to the specified grayscale user. At this time, only need to submit the job to the Spark engine under the original Linkis service cluster for execution. Specifically, according to the version No. Determine the deployment directory information and version loading rules of the target Spark engine, and then execute the subsequent steps. The specific execution process can refer to the first embodiment above, and will not be repeated this time.

If the user is in the default grayscale list, create a grayscale Spark engine, and submit the Spark job code to the grayscale Spark engine to execute the job. Among them, the grayscale Spark engine is a new version of the Spark engine under the newly deployed Linkis service cluster, and its creation method is the same as the creation method of the target Spark engine in the first embodiment above, which will not be repeated this time. It should be noted that, during specific implementation, the grayscale Spark engine may be pre-created. In this case, the Spark job code is directly submitted to the pre-created grayscale Spark engine to execute the job.

It should also be noted that since the version of the Spark engine to be submitted has changed, the corresponding version number and configuration parameters must be pre-set. When creating a grayscale Spark engine, it is based on the pre-set version number and configuration parameters. to build. In addition, due to the complexity of the current Spark version parameter definition, in actual implementation, the front and back ends can be configured into the database using a unified version code mechanism, such as v1 for Spark1.6. Allow the user to specify the loading at will, and only the version number with a unified number can be submitted correctly, otherwise the job submission will fail.

In this embodiment, by establishing a user-based version grayscale mechanism, the specified part of users is grayscaled, and the role request of the grayscale user is submitted to the grayscale Spark engine of the new version for execution. Through the above method, when updating the Spark version under Linkis, there is no need to stop the existing Spark engine management service, and the user's Spark job can still be executed during the update period, thereby avoiding adverse effects on the business.

Furthermore, due to different versions of the Spark engine, the Spark job codes will be somewhat different. When the user sends the Spark job, the code of the new Spark version that needs to be submitted may not be used, but the code of the old Spark version is still used to send the execution request of the Spark job. At this time, there will be a problem of job execution failure. .

In this regard, based on the above first to third embodiments, a fourth embodiment of the job execution method of the present invention is proposed.

In this embodiment, before the above step S40, the job execution method further includes:

Step E, modifying the Spark job code according to the version number;

In this embodiment, after obtaining the version number of the target Spark engine and the Spark job code according to the execution request of the Spark job, the Spark job code can be modified according to the version number, so that the modified Spark job code and its version number Compatible matches.

Specifically, a different version of the code parser is defined at the external service entrance, and the changes between different versions and the corresponding modification strategies are predefined through the code parser, so as to predefine the syntax of different versions based on known changes Replacement operations, such as the introduction of changes in different versions of packages, or changes in the API (Application Programming Interface, application program interface) function call interface. When modifying, the code parser matches the Spark job code with the predefined code differences and modification strategies of different versions according to the version number, and then performs corresponding modification according to the matching result.

At this point, step S40 includes: submitting the modified Spark job code to the target Spark engine to execute the job.

Then, submit the modified Spark job code to the target Spark engine to execute the job.

In this embodiment, by automatically modifying the Spark job code according to the code differences and modification strategies of different versions during the code submission stage, it can run compatible with multiple versions without the user needing to modify the existing code.

Further, based on the first to third embodiments above, a fifth embodiment of the job execution method of the present invention is proposed.

In this embodiment, step S40 includes:

Step a41, submitting the Spark job code to the driver node of the target Spark engine;

In this embodiment, the job execution process is as follows:

First submit the Spark job code to the driver node (Driver) of the target Spark engine. Among them, the Driver is used to convert the user's Spark job code into multiple physical execution units. These units are also called tasks (tasks). It is also used to track the running status of the Executor. Assign the appropriate Executor based on the location of the data.

Step a42, converting the Spark job code through the driver node to obtain the Spark task;

Then, convert the Spark job code through the driver node to get the Spark task. The specific conversion method can refer to the prior art.

Step a43, assigning the Spark task to the executor nodes deployed on the Yarn cluster to execute the job.

Finally, the Spark task is assigned to the executor node (Executor) deployed on the Yarn cluster to execute the job. Among them, the Yarn cluster provides a framework for job scheduling and cluster resource management in the big data platform, which can realize the management and scheduling of executor nodes. Executor nodes are used to run Spark tasks and return execution results to driver nodes.

Furthermore, since Spark is distributed, the Spark tasks dynamically generated by the Driver will be serialized and distributed to each Executor for distributed execution. The Driver also provides a remote class loading service, and the Executor deserializes the serialized code When returning to perform dynamic loading, there may be problems. In one case, the serialized result received by the Driver after Executor execution cannot be deserialized correctly. Since there is already a class loader when Linkis starts the Spark engine, this class loader is the default class loader of the Spark Driver, and the user's Spark job code is submitted to the Driver for interpretation and execution, so the Driver will also start a Scala interpreter , that is, when the Spark Driver is initialized, a SparkILoop will also be created. Since the SparkILoop reuses the code interpreter of the Scala language, a class loader will also be set in it. When the user defines a piece of code (that is, the Spark job code) and submits it to the Spark engine, it is first interpreted and executed by the Scala interpreter in the Spark Driver, so some classes newly defined by the user are completely in the class loader of the Scala interpreter. When the sequence The converted Spark task is submitted to the Executor for execution. If it is necessary to continue to return the operation reference of the instance object of the class defined by the user, the serialized execution result will be submitted to the Spark Driver, but at this time, the class loader of the Spark Driver is the default If the class loader lacks the class information defined by the user's dynamic code in the Scala interpreter, it will fail to run, that is, it can only support the operation of part of the code, and it cannot execute correctly when it needs to return the user's newly defined class object.

In view of the above situation, based on the fifth embodiment above, a sixth embodiment of the job execution method of the present invention is proposed.

In this embodiment, before step a42, the job execution method further includes:

Step F, during the initialization process, when creating a Scala interpreter in the driver node of the target Spark engine, inject the class loader of the main thread into the Scala interpreter, so that the class loader of the main thread becomes The parent of the Scala interpreter class loader, and make the Scala interpreter create a corresponding class loader according to the parent's class loader;

In the process of initializing the target Spark engine, a Scala interpreter will be created in the driver node of the target Spark engine. At this time, the class loader of the main thread is injected into the Scala interpreter, so that the class loader of the main thread becomes The parent of the Scala interpreter class loader, which in turn causes the Scala interpreter to create a corresponding class loader based on the parent's class loader;

At this point, step a42 includes:

The Spark job code is transformed by the class loader of the Scala interpreter created in the driver node to obtain the Spark task;

Then, convert the Spark job code through the class loader of the Scala interpreter created in the driver node Driver to obtain the Spark task, and then assign the Spark task to the Executor node deployed on the Yarn cluster to execute the job.

Further, after step a43, the job execution method further includes:

Step G, when receiving the serialized execution result returned by the executor node based on the Spark task, modify the class loader of the current thread of the target Spark engine to the class loader of the Scala interpreter to pass The class loader of the Scala interpreter deserializes the serialized execution result.

When receiving the serialized execution result returned by the executor node Executor based on the Spark task, modify the class loader of the current thread of the target Spark engine to the class loader of the Scala interpreter to serialize through the class loader of the Scala interpreter The execution result is deserialized.

In this embodiment, by dynamically modifying the class loader of the Spark engine, the consistency between the class loader of the Spark engine and the class loader of the Scala interpreter in the Driver is ensured, so that the execution result deserialized by the Executor can be read by the Driver correctly parsed.

Furthermore, it should be noted that due to differences in Spark versions, some codes are difficult to be compatible, and the compilation cannot be passed when switching Spark. In this case, a method combining dynamic compilation and reflection can be used. Normally, you can prepare two Spark job codes corresponding to different Spark versions, and then decide which code to compile at runtime. However, there is a problem with this method. If the value returned by dynamic compilation needs to be serialized and then sent to Executor, some anonymous classes generated in it do not exist in Executor, so there will be problems when deserializing .

In this regard, you can make your own encapsulation implementation for the changed classes, because in Spark, you can get the Spark version by defining a class: org.apache.Spark.SPARK_VERSION (used to get the spark version). Specifically, obtain different version parameters in the code according to the customized class, dynamically load the corresponding version of the class, and then call the method through reflection, which can avoid the above-mentioned compile-time errors. The problem of differences in calling interfaces of different versions is shielded by encapsulating the class itself. However, through reflection, the original UDF (Universal Disc Format, unified disc format) and other codes cannot be used. This is because the udf function requires the ability to deduce the specific type of the input and return value, and if through reflection, we cannot determine the return value (it may be org.apache.Spark.ml.linalg.Vector, or it may be org.apache.Spark.mllib.linalg.Vector), this time it cannot be compiled. In this case, it is necessary to modify the source code version of Spark. According to the current version parameters, the corresponding type can be dynamically loaded as the return value.

The invention also provides a job execution device.

Referring to FIG. 4 , FIG. 4 is a schematic diagram of the functional modules of the first embodiment of the job execution device of the present invention.

As shown in Figure 4, the job execution device includes:

The first acquisition module 10 is used to obtain the version number, dynamic configuration parameters and Spark job codes of the target Spark engine according to the execution request when receiving the execution request of the Spark job;

The first determining module 20 is used to determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

The engine initialization module 30 is used to obtain static configuration parameters according to the deployment directory information, and use the dynamic configuration parameters and the static configuration parameters to initialize the target Spark engine according to the version loading rules to start the target Spark engine;

The job execution module 40 is configured to submit the Spark job code to the target Spark engine to execute the job.

Further, the job execution device further includes:

A detection module, configured to obtain the user ID corresponding to the execution request, and detect whether there is an idle Spark engine corresponding to the user ID and the version number;

The first determination module 20 is also used for if there is no idle Spark engine corresponding to the user identification and the version number, then perform the step of: determining the deployment directory information and Version loading rules;

The first submission module is configured to submit the Spark job code to the idle Spark engine to execute the job if there is an idle Spark engine corresponding to the user ID and the version number.

Further, the job execution device further includes:

A judging module, configured to acquire the user ID corresponding to the execution request, and judge whether the user is in the preset grayscale list according to the user ID;

The first determining module 20 is also used for if the user is not in the preset grayscale list, then perform the step: determine the deployment directory information and version loading rules of the target Spark engine according to the version number;

The second submission module is used to create a grayscale Spark engine if the user is in the preset grayscale list, and submit the Spark job code to the grayscale Spark engine to execute the job.

Further, the job execution device further includes:

The second determination module is used to determine the target calling method according to the version number and the preset abstraction layer interface during the initialization process;

The file package loading module is used to load the file package that the target Spark engine depends on under the directory corresponding to the deployment directory information according to the target calling method.

Further, the job execution device further includes:

A first modification module, configured to modify the Spark job code according to the version number;

The job execution module 40 is also used for:

Submit the modified Spark job code to the target Spark engine to execute the job.

Further, the job execution module 40 includes:

A code submission unit, configured to submit the Spark job code to the driver node of the target Spark engine;

A task generating unit, configured to convert the Spark job code through the driver node to obtain a Spark task;

The task allocation unit is configured to allocate the Spark task to the executor nodes deployed on the Yarn cluster to execute the job.

Further, the job execution device further includes:

The second modification module is used to inject the class loader of the main thread into the Scala interpreter when creating the Scala interpreter in the driver node of the target Spark engine during the initialization process, so that the main thread The class loader becomes the parent of the Scala interpreter class loader, and makes the Scala interpreter create a corresponding class loader according to the parent's class loader;

The task generating unit is also used for: converting the Spark job code through the class loader of the Scala interpreter created in the driver node to obtain the Spark task;

The third modification module is used to modify the class loader of the current thread of the target Spark engine to the class loader of the Scala interpreter when receiving the serialized execution result returned by the executor node based on the Spark task device, so as to deserialize the serialized execution result through the class loader of the Scala interpreter.

Wherein, the function implementation of each module in the above-mentioned job execution device corresponds to each step in the above-mentioned job execution method embodiment, and its functions and implementation processes will not be repeated here.

The present invention also provides a computer-readable storage medium, where a job execution program is stored on the computer-readable storage medium, and when the job execution program is executed by a processor, the job execution method described in any one of the above embodiments is implemented. step.

The specific embodiments of the computer-readable storage medium of the present invention are basically the same as the embodiments of the above-mentioned job execution method, and will not be repeated here.

It should be noted that, as used herein, the term "comprises", "comprises" or any other variation thereof is intended to cover a non-exclusive inclusion such that a process, method, article or system comprising a set of elements includes not only those elements, It also includes other elements not expressly listed, or elements inherent in the process, method, article, or system. Without further limitations, an element defined by the phrase "comprising a..." does not preclude the presence of additional identical elements in the process, method, article or system comprising that element.

The serial numbers of the above embodiments of the present invention are for description only, and do not represent the advantages and disadvantages of the embodiments.

Through the description of the above embodiments, those skilled in the art can clearly understand that the methods of the above embodiments can be implemented by means of software plus a necessary general-purpose hardware platform, and of course also by hardware, but in many cases the former is better implementation. Based on such an understanding, the technical solution of the present invention can be embodied in the form of a software product in essence or in other words, the part that contributes to the prior art, and the computer software product is stored in a storage medium (such as ROM/RAM) as described above. , magnetic disk, optical disk), including several instructions to enable a terminal device (which may be a mobile phone, computer, server, air conditioner, or network device, etc.) to execute the methods described in various embodiments of the present invention.

The above are only preferred embodiments of the present invention, and are not intended to limit the patent scope of the present invention. Any equivalent structure or equivalent process conversion made by using the description of the present invention and the contents of the accompanying drawings, or directly or indirectly used in other related technical fields , are all included in the scope of patent protection of the present invention in the same way.

Claims (9)

1. A job execution method, the job execution method comprising:

when an execution request of Spark job is received, acquiring a version number, dynamic configuration parameters and Spark job codes of a target Spark engine according to the execution request;

determining deployment catalog information and version loading rules of the target Spark engine according to the version number;

acquiring static configuration parameters according to the deployment catalog information, and initializing the target Spark engine by using the dynamic configuration parameters and the static configuration parameters according to the version loading rules so as to start the target Spark engine;

submitting the Spark job code to the target Spark engine to execute a job;

before the step of determining the deployment catalog information and the version loading rule of the target Spark engine according to the version number, the method further comprises the following steps:

acquiring a user identifier corresponding to the execution request, and judging whether a user is in a preset gray list or not according to the user identifier;

if the user is not in the preset gray list, executing the steps of: determining deployment catalog information and version loading rules of the target Spark engine according to the version number;

If the user is in the preset gray list, a gray Spark engine is created, and the Spark job code is submitted to the gray Spark engine to execute the job.

2. The job execution method as set forth in claim 1, wherein before the step of determining the deployment catalog information and version loading rules of the target Spark engine according to the version number, further comprising:

acquiring a user identifier corresponding to the execution request, and detecting whether an idle Spark engine corresponding to the user identifier and the version number exists;

if not, executing the steps of: determining deployment catalog information and version loading rules of the target Spark engine according to the version number;

if so, submitting the Spark job code to the idle Spark engine to execute the job.

3. The job execution method as set forth in claim 1, wherein the job execution method further comprises:

in the initialization process, determining a target calling method according to the version number and a preset abstract layer interface;

and loading a file package which is dependent on the target Spark engine under the corresponding directory of the deployment directory information according to the target calling method.

4. A method of executing a job as claimed in any one of claims 1 to 3, wherein before the step of submitting the Spark job code to the target Spark engine to execute the job, further comprising:

modifying the Spark job code according to the version number;

the step of submitting the Spark job code to the target Spark engine to execute a job includes:

and submitting the modified Spark job code to the target Spark engine to execute the job.

5. A job execution method as claimed in any one of claims 1 to 3, wherein said step of submitting said Spark job code to said target Spark engine to execute a job comprises:

submitting the Spark job code to a driver node of the target Spark engine;

converting the Spark job code through the driver node to obtain a Spark task;

and distributing the Spark task to an executor node deployed on the Yarn cluster to execute the job.

6. The job execution method as set forth in claim 5, wherein before the step of converting the Spark job code by the driver node to obtain a Spark task, further comprising:

In the initialization process, when a Scala interpreter is created in a driver node of the target Spark engine, a class loader of a main thread is injected into the Scala interpreter, so that the class loader of the main thread becomes a parent level of the Scala interpreter class loader, and the Scala interpreter creates a corresponding class loader according to the class loader of the parent level;

the step of converting the Spark job code through the driver node to obtain a Spark task includes:

converting the Spark job code through a class loader of a Scala interpreter created in the driver node to obtain a Spark task;

after the step of distributing the Spark task to the executor nodes deployed on the Yarn cluster to execute the job, the method further includes:

and when receiving a serialization execution result returned by the executor node based on the Spark task, modifying a class loader of the current thread of the target Spark engine into a class loader of the Scala interpreter so as to deserialize the serialization execution result through the class loader of the Scala interpreter.

7. A job execution device, the job execution device comprising:

The first acquisition module is used for acquiring the version number, the dynamic configuration parameters and the Spark job code of the target Spark engine according to the execution request when the execution request of the Spark job is received;

the first determining module is used for determining deployment catalog information and version loading rules of the target Spark engine according to the version number;

the engine initialization module is used for acquiring static configuration parameters according to the deployment catalog information, and initializing the target Spark engine by using the dynamic configuration parameters and the static configuration parameters according to the version loading rules so as to start the target Spark engine;

the job execution module is used for submitting the Spark job code to the target Spark engine so as to execute a job;

wherein the job execution apparatus further includes:

the judging module is used for acquiring the user identification corresponding to the execution request and judging whether the user is in a preset gray list or not according to the user identification;

the first determining module is further configured to execute the steps if the user is not in the preset gray list: determining deployment catalog information and version loading rules of the target Spark engine according to the version number;

And the second submitting module is used for creating a gray Spark engine if the user is in the preset gray list and submitting the Spark job code to the gray Spark engine so as to execute the job.

8. A job execution system, the job execution system comprising: a memory, a processor, and a job execution program stored on the memory and executable on the processor, which when executed by the processor, implements the steps of the job execution method according to any one of claims 1 to 6.

9. A computer-readable storage medium, wherein a job execution program is stored thereon, which when executed by a processor, implements the steps of the job execution method according to any one of claims 1 to 6.

Images