当前位置：首页 > news >正文

17网站一起做网店下载怎么创建一个网站

news 2025/8/5 22:08:49

17网站一起做网店下载,怎么创建一个网站,团购网站设计,青岛网页设计师背景项目中有很多ods层（mysql 通过cannal）kafka，需要对这些ods kakfa做一些etl操作后写入下一层的kafka（dwd层）。一开始采用的是executeSql方式来执行每个ods→dwd层操作，即类似： def main(…

背景

项目中有很多ods层（mysql 通过cannal）kafka，需要对这些ods kakfa做一些etl操作后写入下一层的kafka（dwd层）。

一开始采用的是executeSql方式来执行每个ods→dwd层操作，即类似：

 def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentval tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)val configuration: Configuration = tableEnv.getConfig.getConfigurationtableEnv.createTemporarySystemFunction("etl_handle", classOf[ETLFunction])// source/sink ddltableEnv.executeSql(CREATE_DB_DDL)tableEnv.executeSql(SOURCE_KAFKA_ODS_TABLE1)tableEnv.executeSql(SINK_KAFKA_DWD_TABLE1)tableEnv.executeSql(SOURCE_KAFKA_ODS_TABLE2)tableEnv.executeSql(SINK_KAFKA_DWD_TABLE2)....// insert dml,在insert语句中调用etl_handle进行预处理和写入tableEnv.executeSql(INSERT_DWD_TABLE1)tableEnv.executeSql(INSERT_DWD_TABLE2)... 
}

当有多个ods->dwd操作放在同一个flink作业中时，发现这种方式会导致每次insert操作都是单独的DAG,非常消耗资源，特别是这些处理都是比较轻量级的，最好是能融合在同一个DAG中共享资源。

解决方案

查看flink文档：INSERT 语句 | Apache Flink

因此，可以采用statementset的方式，将不同insert sql进行分组执行，每组的insert sql会先被缓存到 StatementSet 中，并在StatementSet.execute() 方法被调用时，同一组的 insert sql(sink) 会被优化成一张DAG共用taskmanager，减少资源浪费，即类似：

def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentval tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)val configuration: Configuration = tableEnv.getConfig.getConfigurationtableEnv.createTemporarySystemFunction("etl_handle", classOf[ETLFunction])// source/sink ddltableEnv.executeSql(CREATE_DB_DDL)tableEnv.executeSql(SOURCE_KAFKA_ODS_TABLE1)tableEnv.executeSql(SINK_KAFKA_DWD_TABLE1)tableEnv.executeSql(SOURCE_KAFKA_ODS_TABLE2)tableEnv.executeSql(SINK_KAFKA_DWD_TABLE2)....// insert dmltableEnv.createStatementSet().addInsertSql(INSERT_DWD_TABLE1).addInsertSql(INSERT_DWD_TABLE2).addInsertSql(INSERT_DWD_TABLE3).execute()tableEnv.createStatementSet().addInsertSql(INSERT_DWD_TABLE4).addInsertSql(INSERT_DWD_TABLE5).addInsertSql(INSERT_DWD_TABLE6).execute()
}