transactionLog 数据清洗 main类修改123456789121

src/main/java/cn/quantgroup/dbc/spark/transactionlog/CleanningTransactionLogMain.java

@@ -27,12 +27,12 @@ public class CleanningTransactionLogMain {
         System.out.println("读取hdfsPath完毕: "+JSON.toJSONString(hdfsArr));
 
         Dataset<String> dataset = ss.read().textFile(hdfsArr);
-        System.out.println("dataset: "+dataset.count());
-        dataset.repartition(4).foreachPartition(func -> {
+        System.out.println("dataset"+dataset.count());
+        dataset.repartition(4).foreachPartition(line -> {
             System.out.println("开始执行数据清洗");
-            ArrayList<TransactionLog> transactionLogs = new ArrayList<>();
+            ArrayList<TransactionLog> transactionLogs = new ArrayList<>(500);
             String sql = "INSERT INTO `call_record` (`request_url`, `transaction_id`, `uuid`, `url_type`, `code`, `created_at`, `updated_at`) VALUES (?,?,?,?,?,?,?)";
-            func.forEachRemaining(item -> {
+            line.forEachRemaining(item -> {
 //                System.out.println("单条数据清洗");
                 try {
                     String[] split = item.split("\t");
@@ -46,6 +46,7 @@ public class CleanningTransactionLogMain {
                     transactionLog.setUrl_type(split[2]);
                     transactionLog.setUpdated_at(timestamp);
                     transactionLogs.add(transactionLog);
+                    System.out.println("调用日志大小 "+transactionLogs.size());
                     if (transactionLogs.size() != 0 && transactionLogs.size() % 200 == 0) {
                         System.out.println("执行sql集合: "+transactionLogs.size());
                         JdbcExecuters.prepareBatchUpdateExecuteTransactionid(sql, transactionLogs);
@@ -57,6 +58,7 @@ public class CleanningTransactionLogMain {
                     e.printStackTrace();
                 }
             });
+            System.out.println("调用日志大小 "+transactionLogs.size());
             JdbcExecuters.prepareBatchUpdateExecuteTransactionid(sql, transactionLogs);
         });