如何在NodeJS上下文内获取Apache Flink从KDS查询的作业结果

可行性结论 #

完全可行，核心是利用Flink批处理作业的执行结果返回机制，结合可选的AWS临时消息中转方案，严格符合你的限制要求。

具体实现步骤 #

1. 改造Flink批处理作业 #

你的KDS取数作业必须是批处理模式（流作业持续运行无法触发"完成"状态，也就没法返回结果）：

• 接收from和to时间戳参数：通过Flink的ParameterTool从启动参数中获取这两个值。
• 批处理读取KDS：使用Flink Kinesis批处理源，根据from和to时间戳定位到KDS shard的对应位置，读取该区间内的数据。
• 返回结果：将过滤后的数据集直接作为作业执行结果返回，Flink会自动把结果序列化后通过REST API的execution-result端点暴露。

伪代码示例：

public class KdsBatchJob {
    public static void main(String[] args) throws Exception {
        ParameterTool params = ParameterTool.fromArgs(args);
        long fromTs = params.getLong("from");
        long toTs = params.getLong("to");

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 配置Kinesis批处理源，指定时间范围
        KinesisBatchSource<MyRecord> kinesisSource = KinesisBatchSource.<MyRecord>builder()
                .setStreamName("your-kds-stream")
                .setAWSClientConfig(new AWSClientConfigBuilder().build())
                .setDeserializer(new JsonDeserializationSchema(MyRecord.class))
                .setStartTime(Instant.ofEpochMilli(fromTs))
                .setEndTime(Instant.ofEpochMilli(toTs))
                .build();

        // 读取并过滤数据
        DataSet<MyRecord> filteredData = env.createInput(kinesisSource)
                .filter(record -> record.getTimestamp() >= fromTs && record.getTimestamp() <= toTs);

        // 执行作业并返回结果（Flink自动处理结果序列化）
        filteredData.collect();
    }
}

2. NodeJS后端逻辑实现 #

当用户调用你的GET端点时，按以下流程执行：

• 解析请求中的from和to时间戳参数。
• 调用Flink REST API的POST /jars/:jarid/run，将from和to作为programArgs传入（格式为--from=xxx --to=xxx），拿到返回的jobid。
• 轮询Flink的GET /jobs/:jobid/execution-result端点：
  • 如果作业状态是RUNNING，继续轮询；
  • 当状态变为FINISHED时，该端点会返回作业的执行结果（即过滤后的KDS数据）。
• 将结果解析后直接作为GET端点的响应返回给调用方。

NodeJS代码片段示例：

const axios = require('axios');
const FLINK_REST_URL = 'http://your-flink-cluster:8081';
const JAR_ID = 'your-uploaded-jar-id';

async function handleDataRequest(req, res) {
    const { from, to } = req.query;
    if (!from || !to) {
        return res.status(400).send('Missing from/to timestamp parameters');
    }

    try {
        // 触发Flink作业
        const runResponse = await axios.post(
            `${FLINK_REST_URL}/jars/${JAR_ID}/run`,
            {
                programArgs: [`--from=${from}`, `--to=${to}`]
            },
            { headers: { 'Content-Type': 'application/json' } }
        );
        const jobId = runResponse.data.jobid;

        // 轮询作业执行结果
        let jobResult;
        while (true) {
            await new Promise(resolve => setTimeout(resolve, 1000)); // 每秒轮询一次
            const resultResponse = await axios.get(`${FLINK_REST_URL}/jobs/${jobId}/execution-result`);
            const { status, result } = resultResponse.data;

            if (status === 'FINISHED') {
                jobResult = result;
                break;
            } else if (status === 'FAILED' || status === 'CANCELED') {
                throw new Error(`Flink job failed with status: ${status}`);
            }
        }

        // 返回结果
        res.json(jobResult);
    } catch (error) {
        res.status(500).send(`Error processing request: ${error.message}`);
    }
}

// 注册GET端点
app.get('/fetch-kds-data', handleDataRequest);

3. 大结果量优化方案（可选） #

如果过滤后的数据量过大，超过Flink REST API的返回限制，或者担心NodeJS等待超时，可以结合AWS SQS临时队列：

• 给SQS队列设置消息TTL（比如1分钟），确保消息不会持久化存储。
• 改造Flink作业：将结果逐条发送到该SQS队列，作业完成后将队列URL作为执行结果返回给NodeJS。
• NodeJS拿到队列URL后，批量拉取所有消息，然后清理队列（或等待TTL自动删除），最后将消息拼接后返回。
• 此方案完全符合"不得存储数据"的要求，因为SQS消息会短时间内自动销毁。

关键注意事项 #

• 必须用Flink批处理作业：流作业没有"完成"状态，无法通过execution-result获取结果。
• KDS批处理读取：确保Flink正确配置了Kinesis批处理源的起始/结束时间，避免读取超出范围的数据。
• 结果序列化：Flink返回的结果要使用JSON等NodeJS易解析的格式，避免序列化问题。

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