hidden email 在2021年04月06日 11:36,苗红宾 写道: 你好: 业务场景是:数据源是kafka,10分钟总数据量在10G左右,里面包括200个城市的数据,期望使用滚动窗口,按城市分组,每2分钟触发一次将所有城市的过去十分钟数据放到各自的list里,然后转换成pandas,针对该城市做一次整体计算,每次每个城市的计算耗时平均在60s左右。
现在的使用方式: 1、slide_window = Slide.over(f"10.minutes").every(f"2.minutes").on('ts').alias("w") 2、使用sql语句注册kafka connector, 3、result table使用普通的print: CREATE TABLE sink ( city_id STRING , start_time TIMESTAMP , end_time TIMESTAMP , flag STRING ) with ( 'connector' = 'print' ) 4、通过udaf函数,把source的数据写入csv文件,source.select("write_csv(xxxx)"),然后调用计算函数,读取csv文件内容 5、触发计算逻辑通过select里调用自定义函数strategy_flow_entry的方式:source.window(slide_window).group_by("w, city_id").select("strategy_flow_entry(concat_result)").execute_insert("sink").wait() 这种方式在运行过程中,总是出各种各样问题,比如数据延迟等。 所以想请教一下: 1、针对这个场景,推荐的使用方式是什么?目前的使用方式是不是不太对 2、推荐的任务提交参数要怎么设置?cpu core、内存、并发数、slot等 多谢