功图数据最大最小载荷

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

@@ -30,7 +30,7 @@ import java.util.concurrent.atomic.AtomicInteger;
  * @since 2021/4/14 下午3:16
  */
 @Configuration
-public class CamelJDBCConfiguration /*extends RouteBuilder */{
+public class CamelJDBCConfiguration /*extends RouteBuilder */ {
 
     @Autowired
     private RocketMQTemplate rocketMQTemplate;
@@ -255,35 +255,52 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
             };
         };
     }*/
+    //获取最小载荷
+    public Double min(String[] strings){
+        double[] doubles = new double[strings.length];
+        for (int i = 0; i < strings.length; i++) {
+            doubles[i] = Double.parseDouble(strings[i]);
+        }
+        return Arrays.stream(doubles).min().getAsDouble();
+    }
+    //获取最大载荷
+    public Double max(String[] strings){
+        double[] doubles = new double[strings.length];
+        for (int i = 0; i < strings.length; i++) {
+            doubles[i] = Double.parseDouble(strings[i]);
+        }
+        return Arrays.stream(doubles).max().getAsDouble();
+    }
     @Bean
     public RouteBuilder routeBuilderWithOracle1() {
-         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
-         Date date = new Date(System.currentTimeMillis());
-         String date1 = formatter.format(date);
-         String formatDate = formatter.format(date) + " 00:00:00";
+        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
+        Date date = new Date(System.currentTimeMillis());
+        String date1 = formatter.format(date);
+        String formatDate = formatter.format(date) + " 00:00:00";
         return new RouteBuilder() {
             private SortedSet<String> organization;
             private Map<String, Integer> orgIDs;
             private Integer orgID;
+
             //全部执行完成的大概时间在30-40分钟
             @Override
             public void configure() throws Exception {
                 //0 0 18 * * ?   每天下午6点执行一次
                 //单个执行时间30s左右,在之前有数据的情况下
                 from("timer:mytimer1?period=99999999999")
-                .routeId("oracle-1")
-                .setHeader("date",constant(date1))
-                .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) -> {
-                    organization = new TreeSet<>();
-                    orgID = 0;
-                    orgIDs = new HashMap<>();
-                    return result;
-                })
+                        .routeId("oracle-1")
+                        .setHeader("date", constant(date1))
+                        .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) -> {
+                            organization = new TreeSet<>();
+                            orgID = 0;
+                            orgIDs = new HashMap<>();
+                            return result;
+                        })
                         .step("1")
-                .split(body()).process(exchange -> {
+                        .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
                     String org_level3 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC") + "@" + aRow.get("ZK");
@@ -305,13 +322,13 @@ 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,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]}')" +
-                        " ON conflict(well_id) DO UPDATE set remarks = '${body[BZ]}' "))
-                .to("jdbc:centralbase")
-                .end()
+                        .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]}')" +
+                                " ON conflict(well_id) DO UPDATE set remarks = '${body[BZ]}' "))
+                        .to("jdbc:centralbase")
+                        .end()
 
-                .transform().body((re) -> {
+                        .transform().body((re) -> {
                     List<Map<String, Object>> rows = new ArrayList<>();
                     int code = 0;
                     for (String s : organization) {
@@ -355,33 +372,33 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .log("insert")
                         .end();
                 //单独执行时间10s
-               /* from("timer:mytimer2?period=99999999999")
+                from("timer:mytimer2?period=99999999999")
                         .routeId("oracle-2")
-                        .setHeader("date",constant(date1))
+                        .setHeader("date", constant(date1))
                         .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();
-                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            if (aRow.get("YZ")==null)   aRow.put("YZ","0.0");
-                            if (aRow.get("HYSX")==null) aRow.put("HYSX","0.0");
-                            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");
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    if (aRow.get("YZ") == null) aRow.put("YZ", "0.0");
+                    if (aRow.get("HYSX") == null) aRow.put("HYSX", "0.0");
+                    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");
                 })
                         .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();*/
+                        .end();
                 //查询井对应dym不为空的数据 --目前是只要对应井能查到dym不为空的,无论是什么时间的,都放进去
-               //将查询到的DYM数据更新到cb_pc_pro_wellbore_status_daily中
+                //将查询到的DYM数据更新到cb_pc_pro_wellbore_status_daily中
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行时间是4m15s 317条数据
-             /*   from("timer:mytimer5?period=99999999999")
+                from("timer:mytimer5?period=99999999999")
                         .routeId("oracle-5")
-                        .setHeader("date",constant(date1+" 00:00:00"))
+                        .setHeader("date", constant(date1 + " 00:00:00"))
                         //三个月之内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"))
@@ -390,67 +407,67 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .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}' "))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
-                        .end();*/
+                        .end();
                 //单独执行时间30s
-            /*    from("timer:mytimer3?period=99999999999")
+                from("timer:mytimer3?period=99999999999")
                         .routeId("oracle-3")
-                        .setHeader("date",constant(date1))
-                        .setBody(simple("select  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                        .setHeader("date", constant(date1))
+                        .setBody(simple("select distinct  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                         .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    if (aRow.get("SCSJ")==null)   aRow.put("SCSJ","0.0");
-                    if (aRow.get("RCYL1")==null) aRow.put("RCYL1","0.0");
-                    if (aRow.get("RCYL")==null) aRow.put("RCYL","0.0");
-                    if (aRow.get("RCQL")==null) aRow.put("RCQL","0.0");
-                    if (aRow.get("HS")==null)   aRow.put("HS","0.0");
-                    if (aRow.get("BZ")==null)  aRow.put("DYM","");
+                    if (aRow.get("SCSJ") == null) aRow.put("SCSJ", "0.0");
+                    if (aRow.get("RCYL1") == null) aRow.put("RCYL1", "0.0");
+                    if (aRow.get("RCYL") == null) aRow.put("RCYL", "0.0");
+                    if (aRow.get("RCQL") == null) aRow.put("RCQL", "0.0");
+                    if (aRow.get("HS") == null) aRow.put("HS", "0.0");
+                    if (aRow.get("BZ") == null) aRow.put("DYM", "");
                 })
                         .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]}')"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
-                        .end();*/
+                        .end();
 
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行一次30s
-             /*   from("timer:mytimer4?period=99999999999")
+                from("timer:mytimer4?period=99999999999")
                         .routeId("oracle-4")
-                        .setHeader("date",constant(date1))
-                        .setBody(simple("select  jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
+                        .setHeader("date", constant(date1))
+                        .setBody(simple("select distinct jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
                         .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    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");
-                    if (aRow.get("BS")==null)   aRow.put("BS","0.0");
-                    if (aRow.get("BX")==null)  aRow.put("BX","0.0");
-                    if (aRow.get("ZS")==null)  aRow.put("ZS","0.0");
-                    if (aRow.get("CC")==null)  aRow.put("CC","0.0");
-                    if (aRow.get("CS")==null)  aRow.put("CS","0.0");
-                    if (aRow.get("BLX")==null)  aRow.put("BLX","");
-                    if (aRow.get("DL")==null)  aRow.put("DL","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");
+                    if (aRow.get("BS") == null) aRow.put("BS", "0.0");
+                    if (aRow.get("BX") == null) aRow.put("BX", "0.0");
+                    if (aRow.get("ZS") == null) aRow.put("ZS", "0.0");
+                    if (aRow.get("CC") == null) aRow.put("CC", "0.0");
+                    if (aRow.get("CS") == null) aRow.put("CS", "0.0");
+                    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,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();
-*/
+
                 //从天安哪里获取的数据
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行一小时的数据30s
-            /*    from("timer:mytimer7?period=99999999999")
+                from("timer:mytimer7?period=99999999999")
                         .routeId("jdbc-gtsj-?")
                         .setBody(simple("select max(prod_date) from centralbase.cb_temp_well_mech_runtime "))
                         .to("jdbc:centralbase")
                         .split(body())
-                        .setHeader("date",simple("${body[max]}"))
+                        .setHeader("date", simple("${body[max]}"))
                         //.setHeader("dyna_create_time", constant(formatDate))
-                        .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 > '${header.date}' "))
+                        .setBody(simple("select distinct 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 > '${header.date}' "))
                         .to("jdbc:gtsj")
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
@@ -460,6 +477,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                     if (aRow.get("displacement") != null && !aRow.get("displacement").equals("") && aRow.get("disp_load") != null && !aRow.get("disp_load").equals("")) {
                         String[] displacements = aRow.get("displacement").toString().split(";");
                         String[] disp_loads = aRow.get("disp_load").toString().split(";");
+                        Double susp_max_load = max(disp_loads);
+                        Double susp_min_load = min(disp_loads);
                         String sgt = "";
                         for (int i = 0; i < displacements.length; i++) {
                             sgt = sgt + displacements[i] + "," + disp_loads[i] + ",";
@@ -470,17 +489,19 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                             w += new BigDecimal(Math.round(Double.parseDouble(s[i]) * 100)).stripTrailingZeros().toPlainString() + ",";
                         }
                         aRow.put("sgt", w);
+                        aRow.put("susp_max_load",susp_max_load);
+                        aRow.put("susp_min_load",susp_min_load);
                     }
-                    if (aRow.get("stroke") == null) aRow.put("stroke","0.0");
-                    if (aRow.get("frequency") == null) aRow.put("frequency","0.0");
-                    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");
+                    if (aRow.get("stroke") == null) aRow.put("stroke", "0.0");
+                    if (aRow.get("frequency") == null) aRow.put("frequency", "0.0");
+                    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_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();*/
+                        .log("insert!!!").end();
 //-------------------------------------------------------------------------------------------------------------------------------------------------------------------
            /*     from("timer:mytimer2?period=999999999")
                         .routeId("oracle-2")
@@ -558,12 +579,12 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */{
                         .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("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();
 
 
             }