1、DataFunSummit#2024面向Data+AI时代的数据湖优化与实践演讲人：李志方 腾讯 高级工程师01AI时代下数据湖面临的挑战02基于PyIceberg的AI训练/推理链路03Apache Iceberg 简介04Iceberg向量表与向量查询目录 CONTENTDataFunSummit#202401AI时代下数据湖面临的挑战数据湖的发展趋势 第一代数据湖（03）只支持分区粒度重写集中式元数据（HMS）HDFS存储数据 第二代数据湖（18）支持行级事务(upsert)分散式元数据（Mainfest）云存储S3/GCS/etcIceberg数据组织离线T+1大数据存储&查询近实时大

2、数据存储&查询第三代数据湖？关系型数据多媒体数据向量数据图数据实时入湖数据加工、离线查询数据科学模型训练/推理向量查询AI任务Data任务数据湖在湖上管理AI数据AI场景的数据管理痛点source:https:/ column)数据版本难以管理，尤其是跨系统的一致性难以保证AI数据读写性能差不同文件格式的重复序列化/反序列化不同存储介质的重复读写I/OAI数据管理的前沿探索https:/www.deeplake.aiDeepLake：版本管理&可视化https:/ Alpha：基于ORC的AI列存格式Apple HQI(SIGMOD 23)：Batch向量查询优化MSRA VBase(OSDI

3、 23)：标量/向量混合查询阿里云 ABDV(VLDB 20)：标量/向量混合查询近期的重要论文DataFunSummit#202402Apache Iceberg 简介Iceberg表存储结构https:/iceberg.apache.org/spec/可插拔Storage Local FS HDFS S3 GCS COS 可插拔Catalog Hive Rest Hadoop 可插拔Format Parquet ORC AVROCatalog层存储 tablemetadata location的映射关系，并提供原子更新能力Metadata层Metadata File：存储表的历史快照（sn

4、apshot）信息Manifest List：代表单个快照（snapshot），表示表的某个时刻状态，可以通过partition range进行过滤Mainfest File：包含一组DataFile的集合，以及partition value，列级min-max等统计信息用于过滤Data 层Data File：存储表的底层数据Delete File（可选）：记录表中被删除的数据12345Iceberg支持数据更新IdNameAge1AAA302BBB403DDD504EEE20IdNameAge1AAA302BBB403ccc50文件重写IdNameAge1AAA302BBB403ccc50文

5、件不重写x.parquetx1.parquetIdNameAge1AAA302BBB403ccc50Delete File+filepositionx.parquet2id3Posistion DeleteEquality DeleteIdNameAge4EEE20+Read Task读时合并Copy-on-Write表（读优化）Merge-on-Read表（写优化）单个DataFile 本身是immuatable的x.parquety.parquetIceberg支持多引擎/多语言开放式 Table Spec社区实现AI生态圈（基于Python）无缝交互引入序列化开销(JVMPython w

6、orker)DataFunSummit#202403基于PyIceberg的AI训练/推理链路PyIceberg 编程API介绍import pyicebergfrom pyiceberg.catalog import load_catalogcatalog=load_catalog(”foo)t1=catalog.load_table(”default.t1)plan=t1.scan()df=session.sql(select*from t1)df.show()torch_dataset=plan.to_torch_dataset()tf_dataset=plan.to_tf_datase

7、t()加载catalog全表扫描Plan部分扫描Plan模型训练/推理plan=t1.scan(row_filter=GreaterThanOrEqual(colC,10.0),selected_fields=(colA,colB,colC),limit=100,)1223arrow_table=plan.to_arrow()3导出为arrowdf=plan.to_pandas()导出为pandas3ray_ds=plan.to_ray()3导出为ray3sql数据洞察自研AI有关特性PyIceberg:基于纯Python+native lib社区版默认DataLoader实现社区默认实现没有

8、特别优化AI的训练/推理场景Remote FilesArrow Mem TableCPUGPU大于单机内存时发生OOMbulk load训练/推理GPU属于宝贵资源数据加载时，计算资源浪费支持大规模AI数据集的DataLoader内部实现为AI模型推理/训练进行一系列优化GPUHDFS FilesAlluxio模型推理（只读取一次）模型训练（多次读取）数据shuffle（仅对训练）CPU对外暴露为Iterator缓存热数据按streaming方式分段加载DataLoader 性能测评wikipedia-2022数据测试集GPU利用率提高到80%+在改造前只有30%+某内部AI业务改造数据链路准

9、备时间从2周缩短到3天，可以快速迭代模型效果batch size为64 配置：40 cores 256G mem引入缓存的优化效果Dataloader的性能提升轻量级数据洞察：PyIceberg SQLsession.sql(”select t1.a,t2.a from t1,t2 where t1.b=t2.b and t1.c 0”).show()abcdScan(t1)Scan(t2)filter(c0)proj(a,b)proj(a,b)JoinabcdabcdScan(t1)abcdtoArrowtoArrowfilter(c0)proj(a,b)proj(a,b)JoinDuckD

10、B-Exec逻辑计划物理计划SQL执行流程社区接口使用复杂列剪枝内存cacheScan(t2)PyIceberg社区版基于过程式API，需手写查询优化人工编写容易出现错误，难以检查涉及多表和子查询的编写困难对算法人员非常不友好轻量级数据洞察引擎：PyIceberg SQLSQL解析&自动Plan优化Scan算子（列裁剪+谓词下推+scan复用）非Scan算子（DuckDB加速）谓词下推结果导出轻量级数据洞察：PyIceberg SQL中等规模（TPCDS SF100）较小规模（TPCDS SF10）PyIceberg查询时间在10s内，相对presto最多具有3.6x性能中等数据集相对pres

11、to最多具有2.1x性能注：大数据集SF1000瓶颈在单机网卡，PyIceberg不适合此类场景查询优化效果（TPCDS SF-100 Q7）配置：40 cores 256G mem端到端效果：单机运行PyIceberg SQL vs presto/spark 集群Presto集群为5节点DataFunSummit#202404Iceberg向量表与向量查询向量查询简介向量(也称为vectors/embeddings)，是近年来AI技术最火热的概念，也是大模型和推荐系统的基石，即将NLP/图像等非结构化数据通过深度学习技术转化为结构化数组向量相似查询：将实体间的相似关系转化为空间上的距离（例如

12、，不同cat图片的向量距离应该很接近）常见向量查询Search(df,v,k)从数据集df中寻找和查询向量v距离最近的top-k个向量暴力搜索的复杂度为 O(size(df)*dim)SimJoin(dfA,dfB,d)dfA和dfB连接，找出满足距离小于d的组合暴力搜索的复杂度为 O(size(dfA)*size(dfB)*dim)返回2条有效结果Search返回3条有效结果SimJoinQuery vectorData vectorembedding model1,2,3,4,5,0,向量表Euclid 距离Jaccard距离向量距离度量向量数据库的近似索引HNSW(PAMI 18)IVF

13、_PQ（TKDE 17）基于聚类的索引用K-means对向量进行聚类，然后用中心点代替原向量基于图的索引将向量之间的邻近关系用边代替，问题转化为图的遍历挑战：数据湖上支持向量查询HNSW索引(PAMI 18)IVF_PQ索引（TKDE 17）数据湖实现向量存储和查询的难点 问题1：向量数据库的近似索引不适合数据湖IVF_PQ(基于聚类)：数据插入后需重新训练，未索引的数据需暴力检索，不适合实时入湖(milvus构建1M需近5分钟)HNSW(基于图)：需要内存维护状态，和很难serverless化，并且分布式查询难度大 问题2：主流向量数据库侧重search，没有优化simjoinMilvus只

14、支持右表大小最多为1K，pgvector/lanceDB不支持基于nestloop实现，重复加载，读放大(apple sigmod23)无法直接用在数据湖上（datafile均为immutable）基于哈希函数的向量索引：LSHLSH（Locality Sensitive Hashing），局部性敏感哈希普通哈希函数（如murmur3)设计目标是最小化碰撞，因此两个向量只要有一个byte不同，哈希值大相径庭LSH设计目标是最大化碰撞，因此相似性的向量往往会映射到同一个哈希桶，在同一个哈希桶中进行搜索复杂性大幅度降低LSH的定义：对于任意两个向量x和y，if distance(x,y)cr,th

15、en LSH(y)=LSH(y)的概率最多为P2常用LSH函数MinhashBRP(BucketRandomProjection)Source：https:/www.pinecone.io/learn/series/faiss/locality-sensitive-hashing/Iceberg向量表：LSH索引0000333355552222444411110bucket 0bucket 1bucket 2145LSH-key0insert into vtable select id,array_to_tensor(data)from source_tableLSHbcuketingLSH向

16、量索引构建步骤计算LSH：将相似的向量赋予相同值的LSH-Key计算Bucketing：将相同LSH-Key的向量映射到同一个bucket优势支持实时更新增量构建索引，避免重新训练索引和数据具有相同的可见性支持近似SimJoin复杂度scalable：O(size(dfA)+size(dfB)*dim)索引增量构建向量之间的LSH没有数据依赖，具有幂等性，可分布式索引构建和查询1.Minhash（集合相似性）2.BRP（数值相似性）索引支持ACID兼容Iceberg内部的事务机制，datafile提交后索引可见数据插入&索引构建以wiki数据为例（有效维160），索引单核生成速度：约6k/se

17、c 并行执行create table VT(id string,emb tensor(dim=128,hash_func=brp)partitioned by(bucket(3,emb)using iceberg Iceberg向量表 DDL语法Iceberg向量表实现：SimJoin执行0000333344441bucket 0bucket 1bucket 0bucket 1localsortlocalsortlocaljoinfilter（distd）localsortlocaljoin1task 0task 1scan data_tblscan

18、 query_tblSortMergeJoinfilter（distd）-向量表simJoinselect t1.id,t2.idfrom data_tbl as t1,query_tbl as t2where tensor_distance(t1.emb,t2.emb)0.5Spark 执行计划（BatchScan+SortMergeJoin）Step2：本地排序（避免读放大）将读取后的数据按LSH key排序，以便批量处理相同LSH值的数据Step3：本地向量搜索（降低计算量）仅在具有相同LSH值的向量之间计算距离，并返回满足距离的结果Step1：bucket匹配（shuffle消除）将左

19、表和右表对应bucket映射对应的到scan task，并初始化reader33334444localsort34欧式距离dquery vector返回三条有效结果data vector复杂度降低为O(4x2xdim)并行执行注意：t1和t2需要采用完成相同的LSH函数和bucket方式Iceberg向量表性能优化1.51.03.002531.51.03.01.061.00.5nulsizevaluesindices1.5,0.0,1.0,0.0,0.0,3.0 1.5,1.0,1.0,3.0,1.0,1.0 稀疏向量稠密向量00003333bucket 0411bucket 1031buck

20、et 0拆分较大的bucketbucket 1localjoinlocaljoin0003bucket 0_0411bucket 1031bucket 0bucket 1localjoinlocaljoin0333bucket 0_103bucket 0localjointype负载不均衡问题1：真实环境下的AI数据集往往包含稀疏向量wikipedia用词频模型处理后有100W维，但有效维平均160Parquet支持稀疏数据压缩，但RunLength编码开销极高Milvus和LanceDB均不支持稀疏向量01解决方案兼容稀疏/稠密向量的自适应列存格式，并提供给用户算子自适应向量类型(array

21、_to_tensor/tensor_distance)解决方案动态拆分倾斜bucket最大task数据量是中位数的84倍数据倾斜基本缓解问题2：真实环境下的AI数据集往往存在数据倾斜例如wikipedia某些热点词条出现概率远高于其他词条SimJoin查询性能提升wikipedia-2022上进行SimJoin(data_tab,query_tab)实验配置：5 executors（2core+10g）shuffle量优化为0Iceberg相对Spark原生实现具有23倍性能向量计算开销从62%降低到32%(LSH计算分摊到insert时)JVM CPU profilingSpark web ui正确性验证：100%SimJoin和Spark ML结果集合比对感谢观看非常感谢您的观看Q&A