From 0f1c27e6cbfb45a8c3c30ac305b3f2111214b37f Mon Sep 17 00:00:00 2001 From: ascrutae Date: Wed, 24 Feb 2016 20:13:33 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E4=BF=AE=E5=A4=8D=E4=BA=8C=E6=AC=A1MapRed?= =?UTF-8?q?uce=E9=83=A8=E5=88=86bug=202.=20=E6=B7=BB=E5=8A=A0=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../chain2summary/Chain2SummaryMapper.java | 3 +- .../chain2summary/Chain2SummaryReducer.java | 33 ++++++++++---- .../ChainSpecificTimeSummary.java | 6 +++ .../analysis/mapper/CallChainMapperTest.java | 44 ++++++++++++++++++- 4 files changed, 76 insertions(+), 10 deletions(-) diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryMapper.java index 96b9a3636..ab6ed26a1 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryMapper.java @@ -12,7 +12,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; -public class Chain2SummaryMapper extends TableMapper { +public class Chain2SummaryMapper extends TableMapper { private Logger logger = LoggerFactory .getLogger(Chain2SummaryMapper.class.getName()); @@ -34,6 +34,7 @@ public class Chain2SummaryMapper extends TableMapper { } context.write(new Text(summary.buildMapperKey()), summary); } catch (Exception e) { + e.printStackTrace(); logger.error("Failed to mapper call chain[" + key.toString() + "]", e); } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryReducer.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryReducer.java index 66e176849..6d1045ff9 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryReducer.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/Chain2SummaryReducer.java @@ -5,11 +5,15 @@ import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Iterator; public class Chain2SummaryReducer extends Reducer { + private Logger logger = LoggerFactory + .getLogger(Chain2SummaryReducer.class.getName()); @Override protected void setup(Context context) throws IOException, InterruptedException { @@ -18,14 +22,27 @@ public class Chain2SummaryReducer extends Reducer values, Context context) throws IOException, InterruptedException { - ChainRelationship4Search chainRelationship = ChainRelationship4Search.load(Bytes.toString(key.getBytes())); - Iterator summaryIterator = values.iterator(); - Summary summary = new Summary(); - while (summaryIterator.hasNext()) { - ChainSpecificTimeSummary timeSummary = summaryIterator.next(); - summary.summary(timeSummary, chainRelationship); - } + doReduceAction(Bytes.toString(key.getBytes()), values.iterator()); + } - summary.saveToHBase(); + public void doReduceAction(String key, Iterator summaryIterator) { + try { + ChainRelationship4Search chainRelationship = ChainRelationship4Search.load(Bytes.toString(key.getBytes())); + Summary summary = new Summary(); + while (summaryIterator.hasNext()) { + try { + ChainSpecificTimeSummary timeSummary = summaryIterator.next(); + summary.summary(timeSummary, chainRelationship); + } catch (Exception e) { + e.printStackTrace(); + logger.error("Failed to reduce", e); + } + } + + summary.saveToHBase(); + } catch (Exception e) { + e.printStackTrace(); + logger.error("Failed to reduce", e); + } } } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/ChainSpecificTimeSummary.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/ChainSpecificTimeSummary.java index 2071ea594..89e428b03 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/ChainSpecificTimeSummary.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chain2summary/ChainSpecificTimeSummary.java @@ -25,6 +25,9 @@ public class ChainSpecificTimeSummary implements Writable { private Map summaryMap; private long summaryTimestamp; + public ChainSpecificTimeSummary() { + } + public ChainSpecificTimeSummary(String rowKey) throws ParseException { String[] splitValue = rowKey.split("-"); this.cId = splitValue[0]; @@ -43,6 +46,9 @@ public class ChainSpecificTimeSummary implements Writable { JsonObject jsonObject = (JsonObject) new JsonParser().parse(in.readLine()); cId = jsonObject.get("cId").getAsString(); userId = jsonObject.get("userId").getAsString(); + if (jsonObject.get("entranceNodeToken") == null) { + throw new IOException("No entryNode Token CId[" + cId + "]"); + } entranceNodeToken = jsonObject.get("entranceNodeToken").getAsString(); summaryMap = new Gson().fromJson(jsonObject.get("summaryMap").toString(), new TypeToken>() { diff --git a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java index 06d9421d4..3e431654d 100644 --- a/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java +++ b/skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java @@ -4,7 +4,8 @@ package com.ai.cloud.skywalking.analysis.mapper; import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper; import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReducer; import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo; -import com.ai.cloud.skywalking.analysis.config.Config; +import com.ai.cloud.skywalking.analysis.chain2summary.Chain2SummaryReducer; +import com.ai.cloud.skywalking.analysis.chain2summary.ChainSpecificTimeSummary; import com.ai.cloud.skywalking.analysis.config.ConfigInitializer; import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData; import com.ai.cloud.skywalking.protocol.Span; @@ -19,8 +20,11 @@ import org.junit.Before; import org.junit.Test; import java.io.IOException; +import java.text.ParseException; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; /** * Created by astraea on 2016/1/15. @@ -31,6 +35,7 @@ public class CallChainMapperTest { private static String ZK_CLIENT_PORT = "29181"; // private static String chain_Id = "1.0a2.1453430186581.3efa259.4296.56.1"; private static String chain_Id = "1.0a2.1453429608422.2701d43.6468.56.1"; + private static String[] summaryRowKeys = {"CID_FF44EB45B69FCC1BE1C09F25DB1DFCEF-5-2016/01/26 20:00:00","CID_EE53EE49E36184A9901A7E5C4993C4F2-5-2016/01/27 14:00:00","CID_53AAB989D315C98CBA7352474EEFA57F-5-2016/01/30 10:00:00"}; private static Configuration configuration = null; private static Connection connection; @@ -69,4 +74,41 @@ public class CallChainMapperTest { connection = ConnectionFactory.createConnection(configuration); } } + + @Test + public void testSummary() throws IOException, ParseException { + ConfigInitializer.initialize(); + + Map> chainSpecificTimeSummaries = new HashMap(); + for (String rowkey : summaryRowKeys) { + ChainSpecificTimeSummary chainSpecificDaySummary = selectSummary(rowkey); + List chainSpecificTimeSummaries1 = chainSpecificTimeSummaries.get(chainSpecificDaySummary.buildMapperKey()); + if (chainSpecificTimeSummaries1 == null){ + chainSpecificTimeSummaries1 = new ArrayList<>(); + } + chainSpecificTimeSummaries1.add(chainSpecificDaySummary); + chainSpecificTimeSummaries.put(chainSpecificDaySummary.buildMapperKey(), chainSpecificTimeSummaries1); + } + + Chain2SummaryReducer reducer = new Chain2SummaryReducer(); + for (Map.Entry> entry : chainSpecificTimeSummaries.entrySet()) { + reducer.doReduceAction(entry.getKey(), entry.getValue().iterator()); + } + + System.out.println("1"); + } + + private ChainSpecificTimeSummary selectSummary(String key) throws IOException, ParseException { + ChainSpecificTimeSummary summary = new ChainSpecificTimeSummary(Bytes.toString(key.getBytes())); + + Table table = connection.getTable(TableName.valueOf(HBaseTableMetaData.TABLE_CHAIN_ONE_MINUTE_SUMMARY_EXCLUDE_RELATIONSHIP.TABLE_NAME)); + Get g = new Get(Bytes.toBytes(key)); + Result value = table.get(g); + for (Cell cell : value.rawCells()) { + summary.addChainNodeSummaryResult(Bytes.toString(cell.getValueArray(), + cell.getValueOffset(), cell.getValueLength())); + } + + return summary; + } } \ No newline at end of file