在实际数据处理和实时计算场景中Flink 作为流处理引擎与大模型LLM的结合正成为一种探索方向。开发者希望利用 Flink 强大的实时数据流处理能力来调用大模型进行文本分析、内容生成、智能决策等任务。当面临模型选型时一个常见的问题是在 Flink 框架下调用GLM 和 DeepSeek 哪个更“厉害”这里的“厉害”通常指向几个维度API 调用的便捷性、推理速度、成本、模型能力如代码生成、文本理解以及对中文的支持度。本文将从工程实践的角度探讨在 Flink 项目中集成大模型 API 的方案并对比 GLM 与 DeepSeek 在关键指标上的差异帮助你根据项目需求做出技术选型。需要明确的是Flink 本身并不直接提供大模型推理能力其角色是作为数据流的编排和调度者。我们的目标是在 Flink 的算子例如ProcessFunction或通过Async I/O中异步调用外部的大模型 API 服务将模型推理无缝嵌入到实时数据处理管道中。因此选型的核心在于评估不同大模型 API 的服务质量、接口稳定性、成本以及它们与 Flink 异步编程模型的契合度。1. 理解 Flink 调用大模型的核心架构与挑战在 Flink 流处理作业中直接进行同步 HTTP 调用是危险的因为网络延迟和模型推理耗时可能长达数秒会严重阻塞数据处理管道导致背压甚至作业失败。因此异步调用是必须遵循的核心原则。1.1 为什么必须使用 Async I/OFlink 的 Async I/O 功能允许单个算子并发处理多个请求并异步等待结果从而在等待外部服务响应时不会阻塞算子的计算资源。这对于调用延迟高的大模型 API 至关重要。其工作流程可以概括为数据流中的每条记录触发一个异步请求。请求被分发到线程池由线程池管理并发请求。算子继续处理后续数据不等待当前请求返回。异步请求完成后结果被收集并发送到下游。1.2 通用集成架构一个典型的 Flink 作业调用大模型 API 的架构如下Kafka Source - Map/ProcessFunction (数据预处理) - Async I/O (调用大模型 API) - Sink (结果写入 Kafka/DB)在Async I/O算子中我们会封装一个AsyncFunction其内部使用 HTTP 客户端如 Apache HttpClient、OkHttp 或异步客户端如 AsyncHttpClient向大模型的 API 端点发起请求。1.3 主要技术挑战容错与重试网络波动或 API 服务暂时不可用。需要在AsyncFunction中实现指数退避等重试机制。速率限制所有大模型 API 都有 QPS每秒查询率或 RPM每分钟请求数限制。需要在 Flink 侧实现限流例如使用 Guava 的RateLimiter。结果解析与错误处理需要健壮地解析 API 返回的 JSON并处理各种错误码如429代表限流503代表服务过载。状态管理某些场景下可能需要关联请求与响应或者累计某些指标会用到 Flink 的状态编程。2. 环境准备与项目依赖配置在开始编写代码前需要搭建一个基础的 Flink 开发环境并引入必要的依赖。这里我们以 Java 项目为例使用 Maven 进行依赖管理。2.1 基础环境要求Java: JDK 8 或 11推荐 11与 Flink 1.17 兼容性更好。Flink: 版本 1.16 或 1.17本文示例基于 1.17.1。构建工具: Maven 3.6 或 Gradle。集成开发环境IDE: IntelliJ IDEA 或 Eclipse。2.2 Maven 核心依赖创建一个新的 Maven 项目在pom.xml中添加以下依赖properties flink.version1.17.1/flink.version scala.binary.version2.12/scala.binary.version /properties dependencies !-- Flink 核心依赖 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- Flink Async I/O 需要连接器依赖通常已包含在 streaming-java 中 -- !-- HTTP 客户端使用异步的 AsyncHttpClient -- dependency groupIdorg.asynchttpclient/groupId artifactIdasync-http-client/artifactId version2.12.3/version /dependency !-- JSON 处理 -- dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId version2.15.2/version /dependency !-- 日志 -- dependency groupIdorg.slf4j/groupId artifactIdslf4j-simple/artifactId version1.7.36/version scoperuntime/scope /dependency /dependencies注意Flink 核心依赖的scope设置为provided是因为在提交到集群运行时集群环境已经提供了这些 Jar 包。本地测试时IDE 或mvn exec:java命令可以正确处理。2.3 获取大模型 API 密钥要调用 GLM 或 DeepSeek 的 API你需要先注册相应的平台账号并获取 API Key。GLM (智谱AI): 访问智谱AI开放平台注册后可在控制台创建 API Key。DeepSeek: 访问 DeepSeek 开放平台完成注册和认证后获取 API Key。请妥善保管你的 API Key不要在代码中硬编码建议通过环境变量或配置文件传入。3. 实现 Flink AsyncFunction 调用大模型 API我们将实现一个通用的AsyncFunction它可以通过配置来适配不同的大模型 API。这里以文本补全Chat Completion任务为例。3.1 定义数据流 POJO 和配置类首先定义输入输出数据的结构。// 输入事件包含需要模型处理的文本 public class InputEvent { private String id; // 用于关联请求和响应 private String text; // 待处理的原始文本 // 省略构造函数、getter、setter } // 输出事件包含模型返回的结果 public class OutputEvent { private String id; private String originalText; private String modelResponse; private long timestamp; // 省略构造函数、getter、setter } // 大模型 API 配置 public class LLMConfig { private String apiKey; private String apiEndpoint; // 如 GLM 的 https://open.bigmodel.cn/api/paas/v4/chat/completions private String modelName; // 如 “glm-4”, “deepseek-chat” private int maxTokens; private double temperature; // 省略其他参数和 getter/setter }3.2 实现通用的 AsyncLLMInvokeFunction这是最核心的类继承RichAsyncFunction。import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.async.ResultFuture; import org.apache.flink.streaming.api.functions.async.RichAsyncFunction; import org.asynchttpclient.*; import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; public class AsyncLLMInvokeFunction extends RichAsyncFunctionInputEvent, OutputEvent { private transient AsyncHttpClient asyncHttpClient; private final LLMConfig llmConfig; private final RateLimiter rateLimiter; // 假设已引入Guava RateLimiter public AsyncLLMInvokeFunction(LLMConfig config) { this.llmConfig config; this.rateLimiter RateLimiter.create(10.0); // 初始限制 10 QPS } Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 初始化异步 HTTP 客户端 DefaultAsyncHttpClientConfig.Builder clientBuilder Dsl.config() .setConnectTimeout(5000) .setRequestTimeout(30000) // 大模型响应可能较慢超时设长 .setMaxRequestRetry(1); // 重试策略可在外部实现 this.asyncHttpClient Dsl.asyncHttpClient(clientBuilder.build()); } Override public void close() throws Exception { super.close(); if (asyncHttpClient ! null) { asyncHttpClient.close(); } } Override public void asyncInvoke(InputEvent input, ResultFutureOutputEvent resultFuture) throws Exception { // 1. 限流 rateLimiter.acquire(); // 2. 构建请求 JSON String requestBody buildRequestBody(input.getText()); // 3. 构建异步 HTTP 请求 BoundRequestBuilder requestBuilder asyncHttpClient.preparePost(llmConfig.getApiEndpoint()) .addHeader(Content-Type, application/json) .addHeader(Authorization, Bearer llmConfig.getApiKey()) .setBody(requestBody); // 4. 执行异步请求并将 Future 转换为 CompletableFuture CompletableFutureResponse responseFuture requestBuilder.execute() .toCompletableFuture() .exceptionally(ex - { // 记录异常返回一个自定义的错误响应或抛出 System.err.println(HTTP请求失败: ex.getMessage()); return null; // 实际应返回一个包含错误信息的Response包装对象 }); // 5. 处理响应完成后调用 resultFuture.complete responseFuture.thenAccept(response - { if (response ! null response.getStatusCode() 200) { String responseBody response.getResponseBody(); String modelOutput parseModelResponse(responseBody); OutputEvent output new OutputEvent(input.getId(), input.getText(), modelOutput, System.currentTimeMillis()); resultFuture.complete(Collections.singleton(output)); } else { // 处理错误例如记录日志、重试或发送到侧输出流 System.err.println(API调用失败状态码: (response ! null ? response.getStatusCode() : N/A)); resultFuture.completeExceptionally(new RuntimeException(LLM API call failed)); } }); } // 构建请求体以GLM API v4格式为例 private String buildRequestBody(String prompt) { // 使用Jackson或简单字符串拼接构建JSON // 示例{model: glm-4, messages: [{role: user, content: prompt}], max_tokens: 500} return String.format( {\model\: \%s\, \messages\: [{\role\: \user\, \content\: \%s\}], \max_tokens\: %d}, llmConfig.getModelName(), prompt.replace(\, \\\), // 简单转义 llmConfig.getMaxTokens() ); } // 解析响应体提取模型生成的文本 private String parseModelResponse(String responseBody) { // 使用Jackson解析JSON // 示例解析从 responseBody 的 JSON 中提取 choices[0].message.content // 这里为简化直接返回原始响应或截取部分 try { com.fasterxml.jackson.databind.JsonNode root objectMapper.readTree(responseBody); return root.path(choices).get(0).path(message).path(content).asText(); } catch (Exception e) { return Error parsing response: e.getMessage(); } } }3.3 在主程序中组装流处理作业现在在 Flink 主程序中创建数据流并使用这个异步函数。import org.apache.flink.streaming.api.datastream.AsyncDataStream; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.util.concurrent.TimeUnit; public class FlinkLLMJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); // 根据并发请求数设置并行度 // 1. 定义数据源这里用集合模拟实际可能是 Kafka DataStreamInputEvent inputStream env.fromElements( new InputEvent(1, 请用Java写一个快速排序函数), new InputEvent(2, 解释一下什么是机器学习), new InputEvent(3, 将Hello, world翻译成法语) ); // 2. 配置大模型参数 LLMConfig glmConfig new LLMConfig(); glmConfig.setApiKey(System.getenv(GLM_API_KEY)); glmConfig.setApiEndpoint(https://open.bigmodel.cn/api/paas/v4/chat/completions); glmConfig.setModelName(glm-4); glmConfig.setMaxTokens(500); // 3. 应用异步 I/O 转换 DataStreamOutputEvent outputStream AsyncDataStream .unorderedWait( // 使用无序等待效率更高除非需要严格顺序 inputStream, new AsyncLLMInvokeFunction(glmConfig), 30000, // 超时时间 30秒 TimeUnit.MILLISECONDS, 100 // 最大并发请求数 ); // 4. 输出结果这里打印实际可写入 Kafka、JDBC Sink 等 outputStream.print(); // 5. 执行作业 env.execute(Flink LLM API Invocation Job); } }4. GLM 与 DeepSeek 在 Flink 集成中的关键对比在 Flink 异步调用的架构下评估 GLM以智谱 GLM-4 为代表和 DeepSeek以 DeepSeek Chat 为代表的“厉害”之处需要聚焦于工程集成相关的指标。以下是从实际调用角度总结的对比对比维度GLM (智谱 AI)DeepSeekAPI 文档与规范性提供标准的 OpenAI 兼容格式的 Chat Completion API文档清晰社区示例丰富。同样提供 OpenAI 兼容格式的 API文档结构清晰上手快速。模型响应速度平均响应时间在 1-3 秒在流处理场景中属于可接受范围。平均响应时间较快通常在 1-2 秒对实时性要求更高的流水线更友好。上下文长度GLM-4 支持 128K 上下文适合处理长文本摘要、长文档分析等任务。DeepSeek-V3 支持 128K最新版本也支持超长上下文两者在此维度上持平。代码生成能力在代码补全、解释、调试方面表现优秀尤其对中文注释的理解和生成有优势。在代码生成和逻辑推理方面口碑极佳被认为是其强项在多项基准测试中排名靠前。中文理解与生成作为国产模型对中文语境、成语、网络用语的理解非常出色中文文本生成质量高。中文能力同样很强但在一些非常本土化的表达或文化相关任务上GLM 可能略有优势。API 调用成本按 token 计费价格透明。对于高频调用需要仔细评估成本。同样按 token 计费在特定时期或活动期间可能有更具竞争力的价格策略需实时对比。稳定性与可用性平台运营时间长服务稳定性较高有完善的 SLA 保障。作为后起之秀发展迅速服务稳定性也在不断提升但长期运营记录相对较短。Flink 集成复杂度低。使用标准 HTTP 客户端即可调用身份验证简单Bearer Token。低。与 GLM 类似集成方式几乎一致无额外复杂度。关键判断从纯技术集成角度看两者在 Flink 中调用的复杂度几乎没有区别。选型的决定性因素往往在于业务需求侧重点更看重代码生成还是中文创作、实时性要求毫秒级差异是否关键以及成本预算。4.1 如何根据场景选择选择 GLM 的场景业务内容以中文内容创作、润色、摘要为主。需要处理非常本土化的语境和表达。团队对智谱的生态工具如 ChatGLM 系列开源模型有前期技术积累。选择 DeepSeek 的场景任务核心是代码生成、补全、审查或算法逻辑推理。对推理速度有极致要求希望进一步降低流处理延迟。希望尝试在代码能力上表现更突出的模型。5. 生产环境部署的注意事项与最佳实践在本地测试通过后将 Flink 调用大模型的作业部署到生产环境如 YARN 或 Kubernetes 集群还需要考虑更多因素。5.1 配置管理切勿将 API Key 等敏感信息硬编码在代码中。推荐做法使用 Flink Configuration通过ExecutionConfig或ParameterTool从启动参数传入。使用外部化配置将配置存储在 Hadoop 分布式缓存、Kubernetes ConfigMap 或专门的配置中心如 Apollo, Nacos在open()方法中读取。环境变量在集群节点或容器中设置环境变量通过System.getenv()获取。// 在main方法中 ParameterTool parameters ParameterTool.fromArgs(args); String apiKey parameters.get(llm.api.key); // 在AsyncFunction的open方法中 LLMConfig config new LLMConfig(); config.setApiKey(getRuntimeContext().getExecutionConfig().getGlobalJobParameters().get(apiKey));5.2 性能与稳定性优化连接池与超时确保AsyncHttpClient配置了合理的连接池大小、超时时间和重试策略。背压处理如果大模型 API 响应变慢会导致 Flink 作业产生背压。需要监控 Flink Web UI 中的背压指标。解决方案包括增加Async I/O算子的并行度。在源端如 Kafka降低消费速率。在AsyncFunction中实现更严格的限流避免压垮下游 API。容错与重试实现一个带退避机制的重试策略而不是简单失败。// 简化的带指数退避的重试逻辑 private CompletableFutureResponse executeWithRetry(BoundRequestBuilder request, int maxRetries) { CompletableFutureResponse future new CompletableFuture(); retryInternal(request, future, maxRetries, 1); return future; } private void retryInternal(BoundRequestBuilder request, CompletableFutureResponse resultFuture, int maxRetries, int attempt) { request.execute().toCompletableFuture() .thenAccept(response - { if (response.getStatusCode() 200) { resultFuture.complete(response); } else if (response.getStatusCode() 429 attempt maxRetries) { // 限流等待后重试 long waitTime (long) (Math.pow(2, attempt) * 1000 Math.random() * 1000); scheduler.schedule(() - retryInternal(request, resultFuture, maxRetries, attempt 1), waitTime, TimeUnit.MILLISECONDS); } else { resultFuture.completeExceptionally(new RuntimeException(Failed after retries)); } }) .exceptionally(ex - { if (attempt maxRetries) { long waitTime (long) (Math.pow(2, attempt) * 1000); scheduler.schedule(() - retryInternal(request, resultFuture, maxRetries, attempt 1), waitTime, TimeUnit.MILLISECONDS); } else { resultFuture.completeExceptionally(ex); } return null; }); }5.3 监控与告警Flink Metrics利用 Flink 内置的 Metrics 系统暴露Async I/O算子的队列长度、平均等待时间、请求成功率等指标。日志聚合将作业日志集中收集到 ELK 或类似平台重点关注 HTTP 请求的异常状态码和超时日志。API 侧监控关注大模型服务商控制台提供的调用量、延迟、错误率仪表盘。6. 常见问题排查清单在开发和运行过程中你可能会遇到以下问题。这里提供一个排查路径。问题现象可能原因检查点与解决方案作业启动失败ClassNotFoundException依赖未正确打包或集群环境缺失 Jar 包。1. 使用mvn clean package生成包含所有依赖的 Uber Jar。2. 检查pom.xml中 Flink 依赖的scope是否为provided提交集群时确保集群有对应版本。Async I/O 算子无输出或输出缓慢1. API 调用超时。2. 并发请求数达到上限被阻塞。3. 背压导致源头停止消费。1. 检查AsyncFunction中的超时设置HTTP 客户端和 Async I/O 等待时间。2. 检查AsyncDataStream.unorderedWait的maxConcurrentRequests参数是否过小。3. 在 Flink Web UI 查看背压情况调整并行度或限流。大量429 Too Many Requests错误请求频率超过了大模型 API 的速率限制。1. 在AsyncFunction中实现更严格的令牌桶或漏桶限流算法。2. 联系服务商确认并调整 QPS 限制。3. 在重试逻辑中加入对 429 状态码的退避等待。返回结果解析失败 (JsonProcessingException)API 响应格式与预期不符或服务端返回了错误信息。1. 打印原始响应体 (responseBody)确认其结构。2. 检查parseModelResponse方法中的 JSON 路径是否正确。3. 处理非 200 状态码的响应将其视为业务异常。作业运行一段时间后内存溢出 (OOM)1. 异步请求队列积压导致内存中驻留过多未完成请求的上下文。2. HTTP 客户端连接池或响应体未释放。1. 减少maxConcurrentRequests或提高下游处理能力。2. 确保AsyncHttpClient在close()方法中被正确关闭。3. 增加 TaskManager 的堆内存。API Key 无效或认证失败环境变量或配置未正确加载或 Key 已过期/被撤销。1. 在open()方法中打印或安全地日志记录加载的配置确认 Key 正确。2. 在服务商控制台验证 API Key 的状态和剩余额度。7. 扩展方向与进阶思考在完成基础集成后可以考虑以下方向来增强系统的能力和鲁棒性动态模型路由实现一个RouterAsyncFunction根据输入内容的特征如语言、任务类型动态选择调用 GLM 或 DeepSeek甚至其他模型实现成本与效果的最优平衡。结果缓存对于重复或相似的查询可以在 Flink 状态中引入一个简单的缓存如 Guava Cache避免重复调用大模型显著降低成本并提升速度。流批结合与模型微调使用 Flink 实时处理用户反馈数据如对模型生成结果的点赞/点踩定期批处理用这些数据对开源小模型进行微调再通过 Flink 将轻量级的微调模型部署为实时服务形成闭环。复杂工作流编排将一次大模型调用升级为多步链式调用如先总结再翻译最后情感分析。可以利用 Flink 的迭代操作或拆分成多个连续的Async I/O步骤来实现但需仔细设计错误处理和状态一致性。最终在 Flink 中调用 GLM 还是 DeepSeek并非一个非此即彼的问题。更成熟的架构应该具备可插拔性允许你通过配置轻松切换或同时使用多个模型。工程上的重点始终是构建一个高吞吐、低延迟、具备容错和监控能力的异步调用框架这将是你应对未来任何大模型 API 变更或新模型出现的最坚实保障。建议在实际项目中针对你的核心业务场景用同样的测试数据集对两个模型进行效果和性能的基准测试让数据驱动决策。