申请单数据清洗test

src/main/java/cn/quantgroup/dbc/spark/applyList/ApplyListExportToBizFlowMain.java

@@ -85,7 +85,8 @@ public class ApplyListExportToBizFlowMain {
         System.out.println("创建三张表视图完成");
 
         SQLContext sqlContext = ss.sqlContext();
-        String sql = "INSERT INTO `apply_list` (`channel_group_no`, `channel_id`, `product_type`, `uuid`, `apply_no`, `apply_status`, `audit_result`, `apply_submit_time`) VALUES (?,?,?,?,?,?,?,?)";
+//        String sql = "INSERT INTO `apply_list` (`channel_group_no`, `channel_id`, `product_type`, `uuid`, `apply_no`, `apply_status`, `audit_result`, `apply_submit_time`) VALUES (?,?,?,?,?,?,?,?)";
+        String sql = "INSERT INTO `apply_list` (`apply_no`) VALUES (?)";
 
         Dataset<Row> dataset = sqlContext.sql("select a.product_type,a.channel,a.order_no,a.receive_at,a.credit_status,b.apply_time,b.apply_status,b.risk_notify_time,c.uuid  from quotaCreditDataSetMapView a left join applyQuotaRecordDataSetMapView b on a.order_no = b.order_no left join userInfoListDataSetMapView c on b.user_id = c.id");
         System.out.println("dataset" + dataset.count());
@@ -95,38 +96,38 @@ public class ApplyListExportToBizFlowMain {
             line.forEachRemaining(result -> {
                 try {
                     ApplyListRow applyListRow = new ApplyListRow();
-                    applyListRow.setChannelGroupNo(result.getAs("channel"));
-                    applyListRow.setChannelId(result.getAs("channel"));
-                    applyListRow.setProductType(result.getAs("product_type"));
-                    applyListRow.setUuid(result.getAs("uuid"));
+//                    applyListRow.setChannelGroupNo(result.getAs("channel"));
+//                    applyListRow.setChannelId(result.getAs("channel"));
+//                    applyListRow.setProductType(result.getAs("product_type"));
+//                    applyListRow.setUuid(result.getAs("uuid"));
                     applyListRow.setApplyNo(result.getAs("order_no"));
-
-                    String credit_status = result.getAs("credit_status");
-                    applyListRow.setApplyStatus(getApplyStatus(credit_status, result.getAs("apply_status")));
-
-                    String apply_status = result.getAs("apply_status");
-                    applyListRow.setAuditResult("0".equals(apply_status) ? null : "1".equals(apply_status) ? "0" : "1");
-
-                    Timestamp receive_at = result.getAs("receive_at");
-                    String str = "2000-01-01 00:00:00";
-                    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-                    Date parse = dateFormat.parse(str);
-                    applyListRow.setApplySubmitTime(receive_at.compareTo(new Timestamp(parse.getTime())) == 0 ? null : receive_at);
+//
+//                    String credit_status = result.getAs("credit_status");
+//                    applyListRow.setApplyStatus(getApplyStatus(credit_status, result.getAs("apply_status")));
+//
+//                    String apply_status = result.getAs("apply_status");
+//                    applyListRow.setAuditResult("0".equals(apply_status) ? null : "1".equals(apply_status) ? "0" : "1");
+//
+//                    Timestamp receive_at = result.getAs("receive_at");
+//                    String str = "2000-01-01 00:00:00";
+//                    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+//                    Date parse = dateFormat.parse(str);
+//                    applyListRow.setApplySubmitTime(receive_at.compareTo(new Timestamp(parse.getTime())) == 0 ? null : receive_at);
                     applyListRows.add(applyListRow);
 
                     longAccumulator.add(1);
 
-//                    if (applyListRows.size() != 0 && applyListRows.size() % 200 == 0) {
-//                        JdbcExecuters.prepareBatchUpdateExecuteApply(sql, applyListRows);
-//                        applyListRows.clear();
-//                    }
+                    if (applyListRows.size() != 0 && applyListRows.size() % 200 == 0) {
+                        JdbcExecuters.prepareBatchUpdateExecuteApply(sql, applyListRows);
+                        applyListRows.clear();
+                    }
 
                 } catch (Exception e) {
                     System.out.println("申请单清洗异常" + result.toString() + "----" + JSON.toJSONString(applyListRows));
                     e.printStackTrace();
                 }
             });
-//            JdbcExecuters.prepareBatchUpdateExecuteApply(sql, applyListRows);
+            JdbcExecuters.prepareBatchUpdateExecuteApply(sql, applyListRows);
         });
         System.out.println("longAccumulator大小:"+longAccumulator.count());
 

src/main/java/cn/quantgroup/dbc/utils/JdbcExecuters.java

@@ -127,14 +127,15 @@ public class JdbcExecuters {
             conn.setAutoCommit(false);
             for (int i = 0; i < applyListRows.size(); i++) {
                 ApplyListRow applyListRow = applyListRows.get(i);
-                ps.setString(1, applyListRow.getChannelGroupNo());
-                ps.setString(2, applyListRow.getChannelId());
-                ps.setString(3, applyListRow.getProductType());
-                ps.setString(4, applyListRow.getUuid());
-                ps.setString(5, applyListRow.getApplyNo());
-                ps.setString(6, applyListRow.getApplyStatus());
-                ps.setString(7, applyListRow.getAuditResult());
-                ps.setTimestamp(8, applyListRow.getApplySubmitTime()==null?null:applyListRow.getApplySubmitTime());
+                ps.setString(1, applyListRow.getApplyNo());
+//                ps.setString(1, applyListRow.getChannelGroupNo());
+//                ps.setString(2, applyListRow.getChannelId());
+//                ps.setString(3, applyListRow.getProductType());
+//                ps.setString(4, applyListRow.getUuid());
+//                ps.setString(5, applyListRow.getApplyNo());
+//                ps.setString(6, applyListRow.getApplyStatus());
+//                ps.setString(7, applyListRow.getAuditResult());
+//                ps.setTimestamp(8, applyListRow.getApplySubmitTime()==null?null:applyListRow.getApplySubmitTime());
 
                 ps.addBatch();
             }