Node.js调用Spark Job Server API无法获取任务结果求助

我之前也碰到过类似的情况，Spark Job Server（SJS）的任务提交API本质是异步触发的——你调用接口后，服务器只会先返回任务启动成功的确认（比如带任务ID的响应），而不是直接返回最终计算结果。这就是为什么你的res.on('end')会提前触发：初始响应已经结束，但后台Spark任务还在跑呢。

要拿到最终的空值统计结果，你需要调整调用逻辑，分三步来做：

1. 提交任务，获取任务ID #

首先发送POST请求提交Spark任务，拿到SJS返回的任务ID和初始状态（比如STARTED）。

示例代码（用原生http模块）：

const http = require('http');

const submitOptions = {
  hostname: 'your-sjs-host',
  port: 8090,
  path: '/jobs?appName=your-python-egg&classPath=your-job-class',
  method: 'POST',
  headers: {
    'Content-Type': 'application/json'
  }
};

const submitReq = http.request(submitOptions, (res) => {
  let data = '';
  res.on('data', (chunk) => {
    data += chunk;
  });
  res.on('end', () => {
    const result = JSON.parse(data);
    const jobId = result.jobId;
    console.log(`任务已启动，ID：${jobId}`);
    // 下一步：轮询任务状态
    pollJobStatus(jobId);
  });
});

submitReq.write(JSON.stringify({input: 'your-file-path'})); // 传入任务参数
submitReq.end();

2. 轮询任务状态，等待任务完成 #

拿到任务ID后，定时向SJS的任务状态接口发送GET请求，直到任务状态变为FINISHED（或者处理FAILED等异常状态）。

function pollJobStatus(jobId) {
  const statusOptions = {
    hostname: 'your-sjs-host',
    port: 8090,
    path: `/jobs/${jobId}`,
    method: 'GET'
  };

  const statusReq = http.request(statusOptions, (res) => {
    let data = '';
    res.on('data', (chunk) => {
      data += chunk;
    });
    res.on('end', () => {
      const status = JSON.parse(data);
      if (status.status === 'FINISHED') {
        console.log('任务完成，开始获取结果');
        // 下一步：获取最终结果
        getJobResult(jobId);
      } else if (status.status === 'FAILED') {
        console.error('任务执行失败：', status.result);
      } else {
        // 任务还在运行，1秒后再次轮询
        setTimeout(() => pollJobStatus(jobId), 1000);
      }
    });
  });

  statusReq.end();
}

3. 获取任务最终结果 #

任务状态变为FINISHED后，调用结果接口拿到空值统计的最终数据。

function getJobResult(jobId) {
  const resultOptions = {
    hostname: 'your-sjs-host',
    port: 8090,
    path: `/jobs/${jobId}/result`,
    method: 'GET'
  };

  const resultReq = http.request(resultOptions, (res) => {
    let data = '';
    res.on('data', (chunk) => {
      data += chunk;
    });
    res.on('end', () => {
      const finalResult = JSON.parse(data);
      console.log('空值统计结果：', finalResult);
      // 这里处理你的业务逻辑
    });
  });

  resultReq.end();
}

关键注意点 #

• 一定要根据SJS的实际API路径调整path参数，不同版本的SJS接口可能略有差异。
• 记得添加超时处理，比如轮询超过N次还没完成就终止，避免无限等待。
• 可以用axios等HTTP库替代原生模块，代码会更简洁，但逻辑是一样的。

内容的提问来源于stack exchange，提问作者rakesh_aiyer