可以用,textChunkDS,将文件进行分割成多个数据源,再用mapreduce对多个数据源进行遍历处理,加上一个新的字段,演示代码:
login(`admin,`123456) dbpath = "dfs://abc" if(existsDatabase(dbpath)){ dropDatabase(dbpath) } db1=database(,HASH,[SYMBOL,40]); db2=database(,VALUE,2020.01M..2022.01M); db=database(dbpath,COMPO,[db1, db2]) /** csv导入 **/ // 建表 t=extractTextSchema("/home/dolphindb/importData/20200601/20200601.csv") tt=table(1:0,t.name join `datetimev,t.type join DATETIME) db.createPartitionedTable(tt,`quotes, `symbol`datetimev) def importMut(mutable t, datetimev){ tmp = take(datetimev,t.rows()) // 新加datetime列数据 t=t join tmp // 直接append插入表中 loadTable("dfs://abc",`quotes).append!(t) } def importCsvFile(dir , db , datetimev){ names = exec filename from files(dir) where isDir = false; for ( n in names){ ds=textChunkDS(dir+n,800) mr(ds,importMut{,datetimev},,,false) } }