RFC-004: Design of TaskManager and related offline jobs

[tobegit3hub]

Problem

Currently, OpenMLDB doesn't have any service to manager the internal batch jobs.

For example, users have to submit the Spark-based OpenMLDB batch job by themselves and initialize the Java/Hadoop/Spark environment in advance. Users have to implement the Spark or MapReduce jobs for importing data in OpenMLDB and submit by themselves. The batch jobs are out of control since OpenMLDB doesn't have job manager to monitor users batch jobs.

Moreover, setting up the client of OpenMLDB is complex if it need to submit the distributed jobs to Yarn or Kubernetes cluster.

For example, the Java package, pre-built Spark distribution files and Hadoop configuration files are required to submit OpenMLDB batch job.

Solution

We will add a new internal module named TaskManager to implement management of batch jobs.

Here is the architecture of TaskManager in OpenMLDB.

TaskManager is the RPC services which provides APIs to submit/monitor/terminate batch jobs. Here is the workflow when clients request to submit batch jobs.

1. Clients like CLI, JDBC, Python SDK and C++ SDK can request NameServer to submit the batch job.
2. NameServer will handle client requests and send requests to TaskManager to submit the job.
3. TaskManager will handle requests and submit the jobs to distributed cluster like Yarn or Kubernetes.
4. For some kinds of batch jobs, TaskManager will request to get the job status including Yarn Application ID or Kubernetes BatchJob ID.
5. TaskManager will return the job status to NameServer.
6. Meanwhile, the metadata of the submitted jobs is stored in internal system tables which will persist in tablet.
7. Then clients can get response from NameServer and check the job status with the job id later.

Notice that batch jobs may run for serval minutes or serval days. TaskManager may response once it gets the basic job status in seconds while the job is still running. Users have to request the job status to check if the job has been finished by themselves.

After adding TaskManager service, we will implement some internal batch jobs which is useful for OpenMLDB users.

• Run OpenMLDB Batch SQL jobs in Yarn cluster
• Import data in offline storage with Spark in Yarn cluster
• Import data in online storage with Spark in Yarn cluster

More jobs will be added in the future but we will not discuss in this RFC, such as submitting Kubernetes or Flink jobs.

The basic job info will be defined below.

Attribute	Type	Examples	Notice
id	int	1, 2, 3...
job	String	SparkBatchSql, ImportOfflineData, ImportOnineData
state	String	Submitted, Running, Finished, Failed, Killed, Unknown, Lost
start_time	timestamp	145954243....
end_time	timestamp	145954243....
cluster	String	Local, Yarn
parameter	String	"{fileType: parquet, filePath: hdfs:///tmp/, db: db1, outputTable, t2}"...
application_id	String	application_1459542433815_0002....
error	String

There are extra job info can be returned when getting detail job info.

Attribute	Type	Examples	Notice
tracking_url	String	http:///xxx/xxx
progress	double	70, 80...

Changes and Additions to Public Interfaces

TaskManager will add some public APIs. The APIs are defined as bRPC Protobuf functions and users can request with HTTP endpoints as well.

API(protobuf function)	Parameter	Response	Notice
RunBatchSql	String sql, String defaultDb, String outputTable	int code, String msg, int jobId
ImportOfflineData	String fileType, String filePath, String db, String outputTable	int code, String msg, int jobId
GetJobStatus	int jobId	int code, String msg, String status
GetJobs
StopJob	int jobId	int code, String msg, String status

TaskManager will add some commands for CLI.

Command	Parameter	Response	Notice
SHOW JOBS		List of job info
SHOW JOB	String jobId	String id, String status
STOP JOB	String jobId

Performance Impact

This is the new component and will not affect the current performance.

Backwards Compatibility and Upgrade Path

This is the new component and doesn't has backwards compatibility and upgrade issues.

Related Work

• Define the metadata of TaskManager job  #720
• Generate internal unique job id #721
• Create job info system table when NameServer starts #722
• Use system table to store job info #723
• Support flexible Spark parameters for TaskManager #724
• Generate physical plan of "LOAD DATA" and " SELECT INTO" in SQL parser #725
• Support physical plan of "LOAD DATA" in Spark planner #726
• Support symbol-import without copying data #727
• Support Spark connector to write data in online storage #728
• Support online data import with Spark #729
• Support High availability for TaskManager #730
• Support SELECT INTO syntax and export data with Spark #731