【FFA 2026】多模态与向量计算 基于 Flink 的原生流式 多模态数据处理引擎探索与实践

付典背景与问题02面向多模态数据处理场景的API层03执行引擎与性能优化04效果与展望背景与问题为什么多模态数据处理变的越来越重要数据早已存在•图片/音频/视频/文档大量存在•受限模型能力数据早已存在•图片/音频/视频/文档大量存在•受限模型能力，难以规模化利用02能力拐点到来•LLM与多模态模型快速发展•理解/抽取/检索/生成/推理增强•数据开始可计算、可利用新需求涌现•到「做成生产级数据系统」•计算成本高→容错与GPU利用成核心高成本单条record处理贵比传统结构化计算高几个数量级如何摊薄单条计算成本、减少重算?→异构计算IO/高成本单条record处理贵比传统结构化计算高几个数量级如何摊薄单条计算成本、减少重算?→异构计算IO/CPU/GPU混合负载GPU常是稀缺资源避免被上下游拖慢?→多模态workload带来了哪些新挑战大负载如何降低数据传输与→长链路多阶段流水线预处理→推理大负载如何降低数据传输与→长链路多阶段流水线预处理→推理→后处理如何自然表达整条数据处理链路?→现有Flink的不足现有Flink的不足–CPU&GPUPipeline执行，数据不落地，将GPU资源充分打满–故障恢复避免大规模重算,省GPU成本02PythonDataFrameAPI·UDF·多模态类型·内置多模态处理能力PythonDataFrameAPI是什么DataFrameAPIVSTableAPIDataFrame/PythonUDF自动注册为临时对象建表前无需显式创建TableEnvironment（用户不需要理解TableEnvironment是什么）importimportpyflink.dataframeaspfImportpyflink.dataframeaspf@udf(result_type=DataTypes.STRING())returns.upper()env=EnvironmentSettings.in_streaming_mode()t_env=TableEnvironment.create(env)@pf.udf@pf.udfreturns.upper()t_env.create_temporary_view("t",tbl)t_env.create_temporary_function("SELECTid,to_upper(name)FROMmy_df")"SELECTid,to_upper(name)FROMmy_df")"SELECTid,to_upper(name)FROMt")简化使用方式，让API简单直观避免用户理解简化使用方式，让API简单直观读Source(Kafka)读Source(Kafka)Connector读取使用read_xxx的方式，参数用关键字直接传直接读写dict,贴近pandas习惯，同时避免用户理解Row这个Flink内部类defdefprocess(row:dict)->dit_env.create_temporary_table("src",TableDescriptor.for_connector("kafka").schema(Schema.new_builder().column("id",DataTypes.BIGINT())DataTypes.STRING()).build()).option("properties.bootstrap.servers",df=pf.read_kafka(@udf(result_type=DataTypes.Row([bootstrap_servers="...",topic="t",format="json",bootstrap_servers="...",topic="t",format="json",schema={"id":DataType.bigint(),"name":DataType.string()})DataTypes.FIELD("b",DataTypes.STRING())]))df.map(process,return_dtype=DataType.struct({...}))returnRow(row.a+1,row.b.upper())df.map(process,return_dtype=DataType.struct({...})).format("json")..format("json").build())src=t_env.from_path("src")避免用户理解Flink中的概念：Row多模态场景在API层需要什么特殊能力?多模态场景的需求需要的能力 下载→解码→图像质量计算→过滤低质量图片→调大模型算embedding：一条DataFrame链描述完整多模态pipeline链上用到的多模态能力在已有数据上额外添加一列(下载、解码图像)①②在已有数据上额外添加一列(下载、解码图像)①②整批Python函数算图像质量分③调远程大模型算embedding⑤pf.read_parquet("oss://imgs/",schema={"img_url":DataType.string()}).with_column("img_bytes",#①下载图片col("img_url").fetch_content())image_decode(col("img_bytes"))).map_batches(score_quality,#③批量算图像质量分batch_format="pandas",return_dtype=DataType.struct({"img":DataType.image(),"score":DataType.float32()})).llm.predict("img_url",#⑤调大模型算embeddingmodel="qwen-vl-embedding",output_type={"vec":DataType.list(DataType.float32())}))算子级资源声明&并发推理算子昂贵且依赖异构资源——把资源与并发声明在算子上，只放大该放大的那一段按算子单独设并发按算子单独设并发，只为昂贵的推理节点扩并发，不影响上下游廉价算子df.map_batches(embed,concurrency=8,batch_format="arrow",...)控制向量化批次大小；batch_format可选pandas/arrow，适配不同类型UDFdf.map_batches(embed,batch_size=64,batch_format="pandas",...)在UDF上声明GPU资源需求,框架据此调度到异构资源@@udf(func_type="pandas",num_gpus=0.5,gpu_type="A10",batch_size=64,return_dtype=...)defembed(imgs):...异步外部调用•远程模型/推理服务•高时延外部访问异步外部调用•远程模型/推理服务•高时延外部访问逐条处理•轻量预处理•格式转换Pandas/Arrow•模型推理@udf@udfreturns.upper()df.with_column("u",to_upper(col("text")))@udf(concurrency=32)asyncdefenrich(uid:int)->str:returnawaitclient.get(uid)df.with_column("name",enrich(col("uid")))@udf(func_type="arrow",return_dtype=DataType.string())defclassify(x:pa.Array):returnpa.array(model.predict(x.to_pylist()))df.with_column("label",classify(col("text")))原生多模态数据类型把多模态对象作为一等类型，比如对于图像，携带mode/shape等元信息——系统能理解语义，方便优化，用户用起来也方便图像类型图像类型，一般用于图像解码/视频抽帧之后的图像DataType.image()DataType.image(ImageMode.RGB)DataType.image(ImageMode.RGB，height=224,width=224)DataType.tensor(DataType.float32(),shape=(3,224,224))DataType.tensor(DataT8(),shape=(224,224,3))DataType.DataType.file()(uri，offset，length，size，mDataType.vector("float32",768)DataType.vector("uint8",128)•FLIP-589：/confluence/display/FLINK/FLIP-589%3A+Introduce+FILE+Type+for+Byte-Content+References•FLIP-590：/confluence/display/FLINK/FLIP-590%3A+Introduce+Multimodal+Data+Type%3A++Vector%2C+Tensor%2C+and+Image?src=contextnavpagetreem视频结构化读取video_metadata视频结构化读取video_metadata帧级处理video_extract_frames生成/拼接video_clip/video_from_frames/video_concatD音频文本规范化text_normalize切块text_chunk计数/拼接text_token_count/text_join_chunks转换image_decode/image_encode/image_resize/image_to_tensor/image_crop/image_flip识别image_detect_objects/image_ocr结构化信息image_sharpness/image_hash/image_metadata结构化信息audiometadata/audioresample/audiosize切分audio_split_by_speech/audio_split_by_timestamp转换audio_to_tensor/audio_decode/audio_encode详见FLIP-593：/confluence/display/FLINK/FLIP-593%3A+Introduce+Built-in+Operators+for+Multimodal+Data+Processing?src=contextnavpagetreemodeDataFrame构造：从内存对象创建配置管理Catalog管理DataFrame构造：从内存对象创建配置管理Catalog管理读写外部存储：Source&Sink(I/O)排序/去重数据类型聚合排序/去重数据类型聚合集合运算行操作Join集合运算行操作Joinjoin/join_asof/join_lateralAI/LLMAI/LLM空值处理转换&执行详见FLIP-591：/confluence/display/FLINK/FLIP-591%3A+Introducing+Python+DataFrame+API+in+PyFlink03执行引擎与性能优化Arrow列式·多层零拷贝·全链路列存非阻塞·Checkpoint常见直觉常见直觉–关键在于是否可以降低/消除整个数据处理链路上的通信/序列化开销决定overhead的不是「是不是Java通用交换标准–operator/通用交换标准–operator/subtask/Java-Python之间进程之间不同subtask之间》进程之间不同subtask之间》》二Operator间二Operator间•引入VectorizedColumnBatchRowData，避免反复行列转换•列式格式贯穿整条链路一Operator内•Operator内•Java/Python进程之间通过共享内存零拷贝传输数据•降低跨进程数据通信开销&序列化/反序列化开销•同TM内不同subtask之间通过共享内存零拷贝传输数据•降低跨进程数据通信开销&序列化/反序列化开销多模态放大了问题多模态放大了问题守