Pass data between jobs in a workflow - SchedulerX - Alibaba Cloud Documentation Center

When a workflow chains multiple jobs, downstream jobs often need the output of upstream jobs to continue processing. Without a built-in data passing mechanism, you would need to store intermediate results in an external database or cache and manage the read/write logic yourself. SchedulerX workflows solve this by letting each job return a string result through ProcessResult, which the next job retrieves through JobContext.getUpstreamData().

How it works

An upstream job returns a ProcessResult containing a success flag and a string value.
SchedulerX stores the result internally after the job completes.
When a downstream job starts, it calls context.getUpstreamData() to get a list of JobInstanceData objects -- one per upstream job -- each holding the job name and result string.

jobA -- returns "1" --+
                      +---> jobC reads both results via getUpstreamData()
jobB -- returns "2" --+     and computes 1 + 2 = 3

Note: Data passing is supported only for simple Java jobs. For distributed Java jobs, use the MapReduce model instead.

Implement the job processors

Create three job processor classes that extend JavaProcessor. jobA and jobB each produce a result string, and jobC reads those results.

Before you begin, make sure that you have:

Three Java applications deployed in Enterprise Distributed Application Service (EDAS)
A SchedulerX group created for each application. See Create an application
jobA, jobB, and jobC created in SchedulerX. See Create a job
A workflow that connects jobA and jobB as upstream jobs of jobC. See Create a workflow

jobA: return a result

@Component
public class TestSimpleJobA extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobA " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        // Return "1" as the result that downstream jobs can read
        return new ProcessResult(true, String.valueOf(1));
    }
}

jobB: return a result

@Component
public class TestSimpleJobB extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobB " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        // Return "2" as the result that downstream jobs can read
        return new ProcessResult(true, String.valueOf(2));
    }
}

jobC: read upstream results

@Component
public class TestSimpleJobC extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        // Retrieve results from all upstream jobs
        List<JobInstanceData> upstreamDatas = context.getUpstreamData();

        int sum = 0;
        for (JobInstanceData jobInstanceData : upstreamDatas) {
            System.out.println("jobName=" + jobInstanceData.getJobName()
                + ", data=" + jobInstanceData.getData());
            sum += Integer.valueOf(jobInstanceData.getData());
        }

        System.out.println("TestSimpleJobC sum=" + sum);
        // Return the aggregated result
        return new ProcessResult(true, String.valueOf(sum));
    }
}

Key APIs

API	Description
`ProcessResult(boolean success, String result)`	Wraps the job result. The second parameter is the string value passed to downstream jobs.
`context.getUpstreamData()`	Returns a `List<JobInstanceData>` containing one entry per completed upstream job.
`JobInstanceData.getJobName()`	Returns the name of the upstream job.
`JobInstanceData.getData()`	Returns the result string that the upstream job set in `ProcessResult`.

Run the workflow and verify results

Open the Process Management page in the SchedulerX console.
Find your workflow, click More in the Operation column, and select Run once.
On the Workflow details page, right-click jobA and select Details. The Result or error field displays 1.
Right-click jobB and select Details. The result displays 2.
Right-click jobC and select Details. The result displays 3 (1 + 2), confirming that jobC received the outputs from both jobA and jobB.

Console output for jobC:

jobName=jobB, data=2
jobName=jobA, data=1
TestSimpleJobC sum=3

What's next

MapReduce model -- Pass data between distributed Java jobs.