数据介入每一小时执行一次

src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelJDBCConfiguration.java

@@ -386,6 +386,9 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 })
                         .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_status_daily(well_id,prod_date,oil_prod_method,oil_nozzle,back_pres,tubing_pres,casing_pres,pump_depth)" +
                                 "values ('${body[JH]}','${body[RQ]}','${body[CYFS]}','${body[YZ]}','${body[HYSX]}','${body[YYSX]}','${body[TYSX]}','${body[BS]}')"))
+                        .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_status_daily(well_id,prod_date,oil_prod_method,oil_nozzle,back_pres,tubing_pres,casing_pres,pump_depth) " +
+                                "select '${body[JH]}','${body[RQ]}','${body[CYFS]}','${body[YZ]}','${body[HYSX]}','${body[YYSX]}','${body[TYSX]}','${body[BS]}' " +
+                                "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_pro_wellbore_status_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}')"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
@@ -401,10 +404,11 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .setBody(simple("SELECT  distinct jh,max(rq),dym FROM zd_zdgs.dba01@A2 WHERE dym is not null group by jh,dym"))
                         .to("jdbc:oracle")
                         .split(body())
-                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '${body[DYM]}' where well_id = '${body[JH]}' and prod_date = '${header.date}' "))
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '${body[DYM]}' where well_id = '${body[JH]}' "))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
+
                 //单独执行时间30s
                 from("timer:mytimer3?period=3600000")
                         .routeId("oracle-3")
@@ -423,6 +427,9 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 })
                         .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_vol_daily(well_id,prod_date,prod_time,liq_prod_daily,oil_prod_daily,gas_prod_daily,water_cut,remarks) " +
                                 "values ('${body[JH]}','${body[RQ]}','${body[SCSJ]}','${body[RCYL1]}','${body[RCYL]}','${body[RCQL]}','${body[HS]}','${body[BZ]}')"))
+                        .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_vol_daily(well_id,prod_date,prod_time,liq_prod_daily,oil_prod_daily,gas_prod_daily,water_cut,remarks) " +
+                                "select '${body[JH]}','${body[RQ]}','${body[SCSJ]}','${body[RCYL1]}','${body[RCYL]}','${body[RCQL]}','${body[HS]}','${body[BZ]}' " +
+                                "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_pro_wellbore_vol_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}' )"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
@@ -450,14 +457,29 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 })
                         .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,static_pressure,flow_pres,pump_diameter,pump_depth,pump_efficiency,rotate_frequency,stroke_length,stroke_frequency,pump_type,elec_frequency)" +
                                 "values ('${body[JH]}','${body[RQ]}','${body[JY]}','${body[LY]}','${body[BJ]}','${body[BS]}','${body[BX]}','${body[ZS]}','${body[CC]}','${body[CS]}','${body[BLX]}','${body[DL]}')"))
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,static_pressure,flow_pres,pump_diameter,pump_depth,pump_efficiency,rotate_frequency,stroke_length,stroke_frequency,pump_type,elec_frequency) " +
+                                "select '${body[JH]}','${body[RQ]}','${body[JY]}','${body[LY]}','${body[BJ]}','${body[BS]}','${body[BX]}','${body[ZS]}','${body[CC]}','${body[CS]}','${body[BLX]}','${body[DL]}' " +
+                                "where NOT EXISTS ( SELECT * FROM centralbase.cb_temp_well_mech_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}' )"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
 
+                from("timer:mytimer6?period=3600000")
+                        .routeId("oracle-6")
+                        .setHeader("date", constant(date1 + " 00:00:00"))
+                        //五个月之内bj不为空的数据
+                        //.setBody(simple("SELECT distinct jh,rq,dym FROM zd_zdgs.dba01@A2 WHERE  rq between to_date(TO_CHAR(ADD_MONTHS(SYSDATE,-3),'yyyy-MM-dd'),'yyyy-MM-dd')and to_date(TO_CHAR(SYSDATE,'yyyy-MM-dd'),'yyyy-MM-dd') and dym is not null;"))
+                        .setBody(simple("SELECT distinct jh,bj FROM zd_zdgs.dba01@A2 WHERE  rq between to_date(TO_CHAR(ADD_MONTHS(SYSDATE,-5),'yyyy-MM-dd'),'yyyy-MM-dd')and to_date(TO_CHAR(SYSDATE,'yyyy-MM-dd'),'yyyy-MM-dd') and bj is not null"))
+                        .to("jdbc:oracle")
+                        .split(body())
+                        .setBody(simple("update centralbase.cb_temp_well_mech_runtime set pump_diameter = '${body[BJ]}' where well_id = '${body[JH]}' "))
+                        .to("jdbc:centralbase")
+                        .log("insert !!!")
+                        .end();
                 //从天安哪里获取的数据
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行一小时的数据30s
-               /* from("timer:mytimer7?period=3600000")
+                from("timer:mytimer7?period=3600000")
                         .routeId("jdbc-gtsj-?")
                         .setBody(simple("select max(prod_date) from centralbase.cb_temp_well_mech_runtime "))
                         .to("jdbc:centralbase")
@@ -497,14 +519,18 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 })
                         .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
                                 "values ('${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}')"))
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
+                                "select '${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}' " +
+                                "where NOT EXISTS (SELECT * FROM centralbase.cb_temp_well_mech_runtime WHERE well_id = '${body[well_name]}' and  prod_date = '${body[dyna_create_time]}' )"))
                         .to("jdbc:centralbase")
-                        .log("insert!!!").end();*/
+                        .log("insert!!!").end();
 
                 from("timer:mytimer1?period=3600000")
                         .routeId("centralbase-1")
                         .setBody(simple("select so.well_id,so.well_common_name,so.org_id,ti.prod_date,ti.stroke_frequency,ti.stroke_length,ti.sgt from centralbase.cb_temp_well_mech_runtime ti, centralbase.cb_cd_well_source so where ti.well_id = so.well_id and ti.prod_date =(select max(prod_date) from centralbase.cb_temp_well_mech_runtime) "))
+                        //.setBody(simple("select so.well_id,so.well_common_name,so.org_id,ti.prod_date,ti.stroke_frequency,ti.stroke_length,ti.sgt from centralbase.cb_temp_well_mech_runtime ti, centralbase.cb_cd_well_source so where ti.well_id = so.well_id and ti.prod_date ='2021-07-12 10:25:00' and ti.well_id = '台24' limit 1 "))
                         .to("jdbc:centralbase")
-                        .split(body()).log("log{body}")
+                        .split(body())
                         .process(exchange -> {
                             Message in = exchange.getIn();
                             HashMap<String, Object> aRow = in.getBody(HashMap.class);
@@ -514,7 +540,6 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                             String prodDate = aRow.get("prod_date").toString().substring(0,19);
                             Double strokeLength = Double.valueOf(aRow.get("stroke_length").toString());
                             Double strokeFrequency = Double.valueOf(aRow.get("stroke_frequency").toString());
-                            System.out.println("strokeFrequency"+strokeFrequency);
                             String sgt = aRow.get("sgt").toString();
                             DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), strokeLength, strokeFrequency);
                             producer.send((MessageBody) diagnoseMsg);