数据导入代码修改

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

@@ -268,10 +268,11 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
 
             @Override
             public void configure() throws Exception {
+                //0 0 18 * * ?   每天下午6点执行一次
                 from("timer:mytimer1?period=999999999")
                 .routeId("oracle-1")
                 .setHeader("date",constant(date1))
-                .setBody(simple("select  jh,cydmc,zyq,zk,qyrq from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                .setBody(simple("select  distinct jh,cydmc,zyq,zk,qyrq,sccw,qk,bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                 .to("jdbc:oracle")
                 .transform()
                 .body((result) -> {
@@ -303,8 +304,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         orgIDs.put(org_level3, orgID);
                     }
                 })
-                .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,station_id,station_name) " +
-                        "values ('${body[JH]}','${body[JH]}','${body[QYRQ]}'::timestamp,'${body[station_id]}','${body[ZK]}')"))
+                .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,station_id,station_name,completion_name,PRODUCING_AREA_name,remarks) " +
+                        "values ('${body[JH]}','${body[JH]}','${body[QYRQ]}'::timestamp,'${body[station_id]}','${body[ZK]}','${body[SCCW]}','${body[QK]}','${body[BZ]}')"))
                 .to("jdbc:centralbase")
                 .end()
 
@@ -340,7 +341,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                     return rows;
                 }).split(body())
                         .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_code,org_name,org_level,parent_id,org_id_pre)" +
-                                "values('${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}')"))
+                                "values('${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}')" +
+                                "ON conflict(org_name) DO UPDATE set org_code = '${body[org_code]}' "))
                         .to("jdbc:centralbase")
                         .end()
                         .setBody(simple("select org_id,org_name from centralbase.cb_pc_organization where org_level = '3' "))
@@ -354,7 +356,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                 from("timer:mytimer2?period=999999999")
                         .routeId("oracle-2")
                         .setHeader("date",constant(date1))
-                        .setBody(simple("select  jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                        .setBody(simple("select  distinct jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                        //.setBody(simple("select  distinct jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from zd_zdgs.dba01@A2 where rq  = to_date('2021-07-05','yyyy-MM-dd') and qyrq is not null "))
                         .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                             Message in = exchange.getIn();
@@ -364,10 +367,24 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                             if (aRow.get("YYSX")==null) aRow.put("YYSX","0.0");
                             if (aRow.get("TYSX")==null) aRow.put("TYSX","0.0");
                             if (aRow.get("BS")==null)   aRow.put("BS","0.0");
-                            if (aRow.get("DYM")==null)  aRow.put("DYM","0.0");
                 })
-                        .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,start_pump_liq_level)" +
-                                "values ('${body[JH]}','${body[RQ]}','${body[CYFS]}','${body[YZ]}','${body[HYSX]}','${body[YYSX]}','${body[TYSX]}','${body[BS]}','${body[DYM]}')"))
+                        .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]}')"))
+                        .to("jdbc:centralbase")
+                        .log("insert !!!")
+                        .end();
+                //查询井对应dym不为空的数据 --目前是只要对应井能查到dym不为空的,无论是什么时间的,都放进去
+               //将查询到的DYM数据更新到cb_pc_pro_wellbore_status_daily中
+                //0 0 */1 * * ? 每1个小时执行一次
+                from("timer:mytimer5?period=999999999")
+                        .routeId("oracle-5")
+                        .setHeader("date",constant(date1))
+                        //三个月之内dym不为空的数据
+                        //.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,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 = '2021-07-05' "))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
@@ -393,6 +410,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .log("insert !!!")
                         .end();
 
+                //0 0 */1 * * ? 每1个小时执行一次
                 from("timer:mytimer4?period=999999999")
                         .routeId("oracle-4")
                         .setHeader("date",constant(date1))
@@ -401,7 +419,6 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    if (aRow.get("DYM")==null)   aRow.put("DYM","0.0");
                     if (aRow.get("JY")==null) aRow.put("JY","0.0");
                     if (aRow.get("LY")==null) aRow.put("LY","0.0");
                     if (aRow.get("BJ")==null) aRow.put("BJ","0.0");
@@ -413,17 +430,18 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                     if (aRow.get("BLX")==null)  aRow.put("BLX","");
                     if (aRow.get("DL")==null)  aRow.put("DL","0.0");
                 })
-                        .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,dynamic_liq_level,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[DYM]}','${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)" +
+                                "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]}')"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
 
                 //从天安哪里获取的数据
-                from("timer:mytimer1?period=10000")
+                //0 0 */1 * * ? 每1个小时执行一次
+                from("timer:mytimer7?period=10000")
                         .routeId("jdbc-gtsj-?")
                         .setHeader("dyna_create_time", constant(formatDate))
-                        .setBody(simple("select well_id,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > (select max(dyna_create_time) from public.pc_fd_pumpjack_dyna_dia_t ) "))
+                        .setBody(simple("select well_name,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > (select max(dyna_create_time) from public.pc_fd_pumpjack_dyna_dia_t ) "))
                         .to("jdbc:gtsj")
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
@@ -449,12 +467,12 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                     if (aRow.get("susp_max_load") == null) aRow.put("susp_max_load","0.0");
                     if (aRow.get("susp_min_load") == null) aRow.put("susp_min_load","0.0");
 
-                })//从这里开始
+                })
                         .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_id]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}')"))
+                                "values ('${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}')"))
                         .to("jdbc:centralbase")
                         .log("insert!!!").end();
-
+//-------------------------------------------------------------------------------------------------------------------------------------------------------------------
            /*     from("timer:mytimer2?period=999999999")
                         .routeId("oracle-2")
                         .setBody(simple("select distinct station_id,station_name from centralbase.cb_cd_well_source"))
@@ -530,16 +548,15 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .to("jdbc:centralbase")
                         .log("insert !!!")
                         .end();
+*/
+               /// from("quartz://name?cron=0 0 */1 * * ?") //0 0 */1 * * ? 每1个小时执行一次
+                 //       .routeId("oracle-6")
+                  //      .setBody(simple("select org_id,org_name from centralbase.cb_pc_organization where org_level = '3' "))
+                  //      .to("jdbc:centralbase")
+                  //      .split(body()).log("${body}")
+                 //       .end();
+
 
-                from("timer:mytimer6?period=999999999")
-                        .routeId("oracle-6")
-                        .setBody(simple("select org_id,org_name from centralbase.cb_pc_organization where org_level = '3' "))
-                        .to("jdbc:centralbase")
-                        .split(body())
-                        .setBody(simple("update centralbase.cb_cd_well_source set org_id = '${body[org_id]}' where station_name = '${body[org_name]}'"))
-                        .to("jdbc:centralbase")
-                        .log("insert !!!")
-                        .end();*/
             }
 
             ;