Task splitting and execution
Cluster tasks split a single task into multiple parts. These parts can run concurrently on different clients based on your business needs. This is achieved through multilayer splitting.
Developing a cluster task involves two stages: the splitting stage and the execution stage.
Splitting stage: The data is sharded. You can use any number of splitting layers. The splitting results are reported to the server-side. The server-side then notifies clients to pull and process the data based on the resulting chunks. A chunk is a collection of indexes for data that needs to be processed.
Execution stage: After receiving a notification, a client pulls and processes the data. When finished, it continues to pull new data until all data is processed.
Application scenario
This topic uses a sample scenario to describe the development process of a cluster task.
Suppose a fund company needs to perform a daily user settlement. Because the company has many users, its user data is distributed across 100 tables. Each table contains about 100,000 user records. The company uses a two-layer cluster task to improve data processing efficiency through parallel processing.
The specific implementation steps are as follows:
Task splitting stage: Split the user data. For more information, see Task splitting stage.
First layer of splitting: Split the data by user table.
Second layer of splitting: Split the data by page.
Task execution stage: Process each user's data. For more information, see Task execution stage.
Execution mode: Supports local and remote execution modes.
Thread pool configuration: Supports default or custom thread pool configurations.
Configure the cluster task in the console. For more information, see Create a scheduling task.
Task splitting stage
Task scheduling provides the IClusterJobSplitHandler interface for task splitting. Two splitting methods are supported:
ShardingChunkData splitting: This method splits tasks by a specified index. Each subtask (chunk) is uniquely identified by its index. The client pulls and processes data based on this index. You must specify the sharding rule.
public class ShardingChunkData implements IChunkData { /** * Part number */ private String shardingRule; ...... }NoteFor a code example, see
ClusterFstSplitHandlerin the sample project.shardingRuleis an identifier for a sharding rule. For example, you can useuser_00, user_01 … user_99as sharding identifiers. In this scenario, the first layer of task splitting can useshardingChunkDatato split the 100 user tables. The following code shows an example:public class UserSplitterByTable implements IClusterJobSplitHandler<ShardingChunkData> { @Override public SplitChunkDataResult<ShardingChunkData> handle(ClusterJobSplitContext context) { SplitChunkDataResult<ShardingChunkData> splitChunkDataResult = new SplitChunkDataResult<>(); ArrayList<ShardingChunkData> shardingChunkDatas = new ArrayList<>(); // user_00 to user_99 for (int i = 0; i < 100; i++) { String shardingRule = "user_"; if (i < 10) { shardingRule = shardingRule +"0"; } shardingRule = shardingRule + i; // Split the shard by table. ShardingChunkData shardingChunkData = new ShardingChunkData(shardingRule); shardingChunkDatas.add(shardingChunkData); } splitChunkDataResult.setChunkDatum(shardingChunkDatas); splitChunkDataResult.setSuccess(true); return splitChunkDataResult; } @Override public String getName() { // Processor name. return "USER_SPLITTER_BY_TABLE"; } @Override public ThreadPoolExecutor getThreadPool() { // Thread pool. Use a custom thread pool. If null is returned, the default thread pool of the SDK is used. return null; } }RangeChunkData splitting: This method splits tasks by a specified range. Each subtask processes data within a specific range, which is similar to paging. You must specify the start index, end index, and sharding rule.
public class RangeChunkData implements IChunkData { /** * Sharding rule */ private String shardingRule; /** * Start index */ private String start; /** * End index */ private String end; ...... }NoteFor a code example, see
ClusterSecSplitHandlerin the project example.The parameters are described as follows:
shardingRule: The sharding rule. For example, you can specify the tableuser_01.The following configuration processes data from index 1000 to 2000 in the
user_01table.RangeChunkData chunk =new RangeChunkData (); chunk.setShardingRule("user_01"); chunk.setStart("1000"); chunk.setEnd("2000");start: The start index of the data range. For example, 1000.end: The end index of the data range (inclusive). For example, 2000.
In this scenario, the second layer can use RangeChunkData to split user data by page. The following code shows an example:
public class UserSplitterByPage implements IClusterJobSplitHandler<RangeChunkData> { @Override public SplitChunkDataResult<RangeChunkData> handle(ClusterJobSplitContext context) { // Shard from the first layer of splitting ShardingChunkData shardingChunkData = (ShardingChunkData) context.getChunkData(); String shardingRule = shardingChunkData.getShardingRule(); // 1. Query the quantity based on the shard int count = queryCountByTable(shardingChunkData.getShardingRule()); SplitChunkDataResult<RangeChunkData> splitChunkDataResult = new SplitChunkDataResult<>(); ArrayList<RangeChunkData> shardingChunkDatas = new ArrayList<>(); // 2. Perform paging. Process 1,000 records per page int pageCount = 1000; for (int page = 0; page < count / pageCount; page++) { String startRows = String.valueOf(page * pageCount); // inclusive String endRows = String.valueOf((page + 1) * pageCount - 1); RangeChunkData rangeChunkData = new RangeChunkData(shardingRule, startRows, endRows); shardingChunkDatas.add(rangeChunkData); } splitChunkDataResult.setChunkDatum(shardingChunkDatas); splitChunkDataResult.setSuccess(true); return splitChunkDataResult; } // mock private int queryCountByTable(String shardingRule) { if ("user_00".equals(shardingRule)) { // The user_00 table has 100,000 users return 100000; } else { // Other tables have 90,000 users return 90000; } } @Override public String getName() { return "USER_SPLITTER_BY_PAGE"; } @Override public ThreadPoolExecutor getThreadPool() { return null; } }
Task execution stage
The cluster task execution stage is divided into the following three sub-stages:
Read stage: Reads data from the data source.
The data reading service must implement the
IReaderinterface:public interface IReader<T>{ /** * Read data * * @param context * @return */ LoadData<T> read(ClusterJobExecuteContext context) throws Exception; }The LoadData object has the following data structure:
public class LoadData<T> extends MultiDataItem<T>{ /** * Indicates whether there is more data to be pulled and processed. If the value is false, the server-side is called back after the current batch of data is processed. If the value is true, more data is available for processing. The client continues to pull data after the current batch is processed. */ private boolean hasMore; public LoadData(List<T> itemList,boolean hasMore){ super(itemList); this.hasMore=hasMore; } public static boolean isEmpty(LoadData loadData){ return loadData==null || loadData.isEmpty(); } public boolean isHasMore(){ return hasMore; } public void setHasMore(boolean hasMore){ this.hasMore=hasMore; } }When you use the read method to read data in batches, note the value of hasMore. If this value is
true, the system reads data again after the current batch is processed.To open and close files, you can use the
IStreamReaderinterface:public interface IStream { /** * Open the stream * * @param context * @throws Exception */ void open(ClusterJobExecuteContext context) throws Exception; /** * Close the stream * * @param context * @throws Exception */ void close(ClusterJobExecuteContext context) throws Exception; } public interface IStreamReader<T> extends IStream, IReader<T> { }This interface inherits from the
IStreaminterface. It provides the open and close methods. These methods are called before and after a shard is executed.Process stage: Transforms the read data object into an object to be written to the data source. This stage is optional.
The data processing service must implement the
IProcessorinterface:public interface IProcessor<I, O> { /** * Data processing. Transforms the read object into the object to be processed. * * @param r * @return */ DataProcessResult<O> process(ClusterJobExecuteContext context, I i) throws Exception; /** * Data processing thread pool. If not configured, the handler's thread pool is used. * * @return */ ThreadPoolExecutor getProcessThreadPool(); }If you do not need to transform the read data, you can skip this service. The read data is then passed directly to the write service. You can specify a separate thread pool for this service. If you do not specify a thread pool, the task handler's thread pool is used by default.
Write stage: Writes data to the data source.
NoteFor code related to the Write stage of a cluster task, see
ClusterExecuteHandlerin the project example.The data writing service must implement the
IWriterinterface:public interface IWriter<T> { /** * The maximum amount of data that a block contains for each execution. * * @return */ int getCountPerWrite(); /** * Write data * * @param context * @param dataItem * @return * @throws Exception */ ClientCommonResult write(ClusterJobExecuteContext context, IDataItem<T> dataItem) throws Exception; /** * Data writing thread pool. If not configured, the handler's thread pool is used. * * @return */ ThreadPoolExecutor getWriteThreadPool(); }The
getCountPerWritemethod sets the amount of data for each batch write. If the return value is less than or equal to 0, only one piece of data is written at a time. You can set a separate thread pool for this service. If you do not set a thread pool, the task handler's thread pool is used by default.IDataItemis the data object to be processed. It has two types:SingleDataItem: Contains only one piece of data. This type is used when the return value of thegetCountPerWritemethod is less than or equal to 1.MultiDataItem: Contains multiple pieces of data. This type is used when the return value of thegetCountPerWritemethod is greater than 1.
You can convert the type of
dataItemas needed. For example:switch(dataItem.getType()){ case SINGLE: ((SingleDataItem<Integer>)dataItem).getItem(); break; case MULTIPLE: (MultiDataItem<Integer>)dataItem).getItemList(); break; default: break; }To open and close files, you can use the
IStreamWriterinterface:public interface IStreamWriter<T> extends IStream,IWriter<T>{ }This interface inherits from the
IStreaminterface. It provides the open and close methods. These methods are called before and after a shard is executed.
Code example
In this application scenario, the execution stage involves pulling data from a database or another data source based on the split shards, processing the data, and then writing the processed data to the database.
Read stage: Pulls user data within a specified range from a specified table based on the shard information (shardingRule, start, and end).
Process stage: Processes the pulled user data.
Write stage: Writes the processed data to the database.
The following shows the code:
public class UserProcessHandler implements IClusterJobExecuteHandler {
private final Logger LOGGER = LoggerFactory.getLogger(ClusterJobExecutor.class);
private ThreadPoolExecutor threadPool;
@Override
public void preExecute(ClusterJobExecuteContext context) {
// Pre-processing
LOGGER.info(String.format("preExecute chunkId:%s, param:%s",
context.getChunkId(), context.getCustomParams()));
}
@Override
public IReader getReader() {
return new IReader() {
@Override
public LoadData<String> read(ClusterJobExecuteContext context) throws Exception {
IChunkData chunkData = context.getChunk();
RangeChunkData rangeChunkData = (RangeChunkData) chunkData;
// Query data based on the index
// You can perform paged queries from the database based on the start and end indexes. The cluster task splits the data into shards according to your sharding rules and then executes the shards. You must implement the specific shard content and execution logic.
String shardingRule = rangeChunkData.getShardingRule();
int start = Integer.parseInt(rangeChunkData.getStart());
int end = Integer.parseInt(rangeChunkData.getEnd());
// Simulate data query
List<String> stringList = queryUserInfo(shardingRule, start, end);
// One-time query, no more data follows
boolean hasMore = false;
return new LoadData<String>(stringList, hasMore);
}
private List<String> queryUserInfo(String shardingRule, int start, int end) {
List<String> stringList = Lists.newArrayList();
for (int userId = start; userId <= end; userId++) {
stringList.add(shardingRule + "" + userId);
}
return stringList;
}
};
}
@Override
public IProcessor getProcessor() {
return new IProcessor() {
@Override
public DataProcessResult process(ClusterJobExecuteContext context,
Object data) throws Exception {
// Process data
System.out.println("process user" + data);
return new DataProcessResult(true, "", data);
}
@Override
public ThreadPoolExecutor getProcessThreadPool() {
return null;
}
};
}
@Override
public IWriter getWriter() {
return new IWriter<String>() {
@Override
public int getCountPerWrite() {
return 1;
}
@Override
public ClientCommonResult write(ClusterJobExecuteContext context,
IDataItem<String> dataItem)throwsException {
// Data storage
switch (dataItem.getType()) {
case SINGLE:
// A single data block that contains only one piece of data
SingleDataItem<String> singleDataItem = (SingleDataItem<String>) dataItem;
LOGGER.info(String.format("getWriter write single data:%s", singleDataItem.getItem()));
System.out.println(String.format("write single data:%s", singleDataItem.getItem()));
break;
case MULTIPLE:
// A composite data block that contains multiple pieces of data. For example, when a task stops, multiple pieces of data are passed in.
MultiDataItem<String> multiDataItem = (MultiDataItem<String>) dataItem;
LOGGER.info(String.format("getWriter write multi data:%s", multiDataItem.getItemList()));
break;
default:
break;
}
return ClientCommonResult.buildSuccessResult();
}
@Override
public ThreadPoolExecutor getWriteThreadPool() {
return BlockingThreadPool.getThreadPool();
}
};
}
@Override
public ILimiter getLimiter() {
return new DefaultLimiter(10);
}
@Override
public void postExecute(ClusterJobExecuteContext context) {
// Post-processing
LOGGER.info(String.format("JobExecuteHandler postExecute chunkId:%s, param:%s",
context.getChunkId(), context.getCustomParams()));
}
@Override
public Progress calProgress(ClusterJobExecuteContext context) {
return new Progress();
}
@Override
public boolean isProcessAsync() {
return true;
}
@Override
public String getName() {
return "USER_PROCESS_HANDLER";
}
@Override
public ThreadPoolExecutor getThreadPool() {
return this.threadPool;
}
public void setThreadPool(ThreadPoolExecutor threadPool) {
this.threadPool = CommonThreadPool.getThreadPool("REMOTE_EXECUTE");
}
}Execution modes
The execution stage of a cluster task supports two modes:
Local execution
For local execution, you must implement the IClusterJobExecuteHandler interface, as shown below:
public interface IJobHandler {
/**
* The name of the handler
*
* @return
*/
String getName();
/**
* Can be left empty. The default execution thread pool is used.
*
* @return
*/
ThreadPoolExecutor getThreadPool();
}
public interface IClusterJobExecuteHandler<I, O> extends IJobHandler {
/**
* Pre-processing
*
* @param context
*/
void preExecute(ClusterJobExecuteContext context);
/**
* Get the data reading service
*
* @return
*/
IReader<I> getReader();
/**
* Get the data cleaning service. If NULL is returned, the data cleaning step is skipped, and the read data is processed directly.
*
* @return
*/
IProcessor<I, O> getProcessor();
/**
* @return
*/
IWriter<O> getWriter();
/**
* Get the limiter. If NULL is returned, the default limiter is used.
*
* @return
*/
ILimiter getLimiter();
/**
* Post-processing after the current execution is complete
*
* @param context
* @return
*/
void postExecute(ClusterJobExecuteContext context);
/**
* Calculate the processing progress
*
* @param context
* @return
*/
Progress calProgress(ClusterJobExecuteContext context);
/**
* Indicates whether to process asynchronously. If asynchronous execution is enabled, the reader, process, and write stages are all processed asynchronously.
*
* @return
*/
boolean isProcessAsync();
}The class that implements IClusterJobExecuteHandler is responsible for processing the task. When you implement it, you must set the following:
IReader
IWriter
IProcessor: Optional. If IProcessor is not set, the read data is passed directly to the IWriter service.
ILimiter: If not set, the default limiting service is used.
If isProcessAsync returns true, the Read, Process, and Write stages are all processed asynchronously. This means that after the read data is placed in the queue for the Process or Write stage, the next read operation starts immediately. If it returns false, the next read operation starts only after the current data is processed by the Write stage.
If the IReader service is configured as a bean, ensure that the service is stateless. Otherwise, interference may occur in multi-threaded scenarios. To store data during processing, create a new IReader service instance in the getReader method. The same rule applies to the IWriter service.
Remote execution
For remote execution, you must implement the IRemoteClusterJobExecuteHandler interface, as shown below:
public interface IRemoteClusterJobExecuteHandler<T> extends IClusterJobExecuteHandler<T>{
/**
* Set the routing policy. If not set, the default is polling.
*
* @return
*/
IClusterRouter getClusterRouter();
/**
* The hibernation time after distributing the loaded data. This prevents repeated pulling of data that has been distributed but not yet processed. The unit is ms. If the return value is <= 0, no hibernation occurs.
*
* @return
*/
int getSleepAfterPerLoad();
}This interface inherits from IClusterJobExecuteHandler and requires two additional methods: getClusterRouter and getSleepAfterPerLoad.
When data is processed remotely, the client distributes the pulled data to other machines in the business cluster for processing. The
getClusterRoutermethod sets the routing rule for this distribution. Supported rules include Random (RandomClusterRouter) and polling (RoundRobinClusterRouter). If this method returnsnull, the polling rule is used by default.The
getSleepAfterPerLoadmethod sets the hibernation time after the read data is processed. This setting prevents the repeated pulling of data that has been distributed but not yet processed. The unit is milliseconds (ms). This method takes effect only whenisProcessAsyncreturnsfalse.
Remote RPC calls within the cluster use the oneway pattern. Therefore, you cannot retrieve the processing results of the data. You must record the processing results yourself.
You can set the thread pool for receiving remote distribution requests by implementing the IRemoteProcessorExecutor interface. Publish the implementation class as a bean. The interface is as follows:
public interface IRemoteProcessorExecutor{
/**
* Get the thread pool
*
* @return
*/
Executor getExecutor();
}In addition, remote calls require you to start an RpcServer on the client:
For SOFABoot projects, use the configuration items in the
application.propertiesfile of the project:com.alipay.sofa.antscheduler.remote.execute.enable=true com.alipay.sofa.antscheduler.remote.execute.port=xxxFor non-SOFABoot projects, set the following two parameters during client initialization:
public class Config{ ...... /** * Indicates whether remote execution is supported */ private boolean isEnableRemoteExecute =false; /** * Remote service port. If not set, the default port 9989 is used. */ private int remoteServerPort; ...... }
Default thread pool configurations
If the thread pools for IProcessor and IWriter are not set, the thread pool set in IJobHandler is used by default. If a thread pool for IJobHandler is also not set, the client's default thread pool is used. The configuration parameters are as follows:
minPoolSize: 20
maxPoolSize: 300
queueSize: 100
keepAliveTime: 1 hourIf IRemoteProcessExecutor is not set, the default thread pool provided by the bolt protocol is used. The configuration parameters are as follows:
minPoolSize: 20
maxPoolSize: 400
queueSize: 600
keepAliveTime: 60 secondsYou can also customize the thread pools for IProcessor, IWriter, IJobHandler, and IRemoteProcessExecutor.