package com.ai.cloud.skywalking.analysis.categorize2chain;

import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo;
import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainNode;
import com.ai.cloud.skywalking.analysis.categorize2chain.util.HBaseUtil;

import org.apache.hadoop.hbase.client.Put;

import java.io.IOException;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;

public class ChainSummary {

    private Map<String, ChainSpecificTimeWindowSummary> loadedChainSpecificTimeWindowSummary;
    private Map<String, Timestamp> updateChainInfo;

    public ChainSummary() {
        loadedChainSpecificTimeWindowSummary = new HashMap<String, ChainSpecificTimeWindowSummary>();
        updateChainInfo = new HashMap<String, Timestamp>();
    }

    public void summary(ChainInfo chainInfo) {
        for (ChainNode node : chainInfo.getNodes()) {
            String csk = generateChainSummaryKey(chainInfo.getCID(), node.getStartDate());
            if (loadedChainSpecificTimeWindowSummary.containsKey(csk)) {
                loadedChainSpecificTimeWindowSummary.put(csk, ChainSpecificTimeWindowSummary.load(csk));
            }

            loadedChainSpecificTimeWindowSummary.get(csk).summaryNodeValue(node);
        }

        updateChainInfo.put(chainInfo.getCID(), new Timestamp(System.currentTimeMillis()));
    }

    private String generateChainSummaryKey(String chainToken, long startDate) {
        return chainToken + "-" + new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").
                format(new Date(startDate / (1000 * 60 * 60) * (1000 * 60 * 60)));
    }

    public void save() throws IOException, InterruptedException, SQLException {
        batchSaveChainSpecificTimeWindowSummary();
        updateChainLastActiveTime();
    }

    private void updateChainLastActiveTime() throws SQLException {
        DBCallChainInfoDao.updateChainLastActiveTime(updateChainInfo);
    }

    private void batchSaveChainSpecificTimeWindowSummary() throws IOException, InterruptedException {
        List<Put> puts = new ArrayList<Put>();
        for (Map.Entry<String, ChainSpecificTimeWindowSummary> entry : loadedChainSpecificTimeWindowSummary.entrySet()) {
            Put put = new Put(entry.getKey().getBytes());
            entry.getValue().save(put);
            puts.add(put);
        }

        HBaseUtil.batchSaveChainSpecificTimeWindowSummary(puts);
    }
}

@@ -274,17 +274,16 @@
 seUtil;%0A
-%0A
 import o
@@ -316,16 +316,73 @@
 ent.Put;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -419,17 +419,55 @@
 ava.sql.
-*
+SQLException;%0Aimport java.sql.Timestamp
 ;%0Aimport
@@ -518,31 +518,8 @@
 l.*;
-%0Aimport java.util.Date;
 %0A%0Apu
@@ -545,16 +545,106 @@
 mary %7B%0A%0A
+    private static Logger logger = LoggerFactory.getLogger(ChainSummary.class.getName());%0A
     priv
@@ -1138,25 +1138,16 @@
 hainInfo
-.getCID()
 , node.g
@@ -1178,16 +1178,17 @@
     if (
+!
 loadedCh
@@ -1347,33 +1347,32 @@
 ;%0A            %7D%0A
-%0A
             load
@@ -1439,33 +1439,32 @@
 ode);%0A        %7D%0A
-%0A
         updateCh
@@ -1593,25 +1593,27 @@
 Key(
-String
+ChainInfo
  chain
-Token
+Info
 , lo
@@ -1652,13 +1652,51 @@
 hain
-Token
+Info.getCID() + %22-%22 + chainInfo.getUserId()
  + %22
@@ -2300,24 +2300,150 @@
 ist%3CPut%3E();%0A
+        logger.info(%22There are %5B%22 + loadedChainSpecificTimeWindowSummary.size() + %22%5D summary data will be storage to HBase%22);%0A
         for 

修复： 1. 第一次MapReduce没有统计结果的问题 2. Summary表中的Rowkey添加上UserId的标识