kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

年月日时报表分批次统计，通过修改MapReduce的Key的不同统计不同的值

This commit is contained in:
ascrutae 2016-05-01 11:31:22 +08:00
parent c84c607420
commit 89dbf1e5d6
6 changed files with 76 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildMapper.java
View File

@ -5,6 +5,7 @@ import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessChain;
import com.ai.cloud.skywalking.analysis.chainbuild.filter.SpanNodeProcessFilter;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode;
import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType;
import com.ai.cloud.skywalking.analysis.chainbuild.util.SubLevelSpanCostCounter;
import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
@ -57,7 +58,13 @@ public class ChainBuildMapper extends TableMapper<Text, Text> {
+ "] to chain with cid[" + chainInfo.getCID() + "].");
if (chainInfo.getCallEntrance() != null && chainInfo.getCallEntrance().length() > 0) {
context.write(
new Text(chainInfo.getCallEntrance()), new Text(new Gson().toJson(chainInfo)));
new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.MINUTER.getValue()), new Text(new Gson().toJson(chainInfo)));
context.write(
new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.HOUR.getValue()), new Text(new Gson().toJson(chainInfo)));
context.write(
new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.DAY.getValue()), new Text(new Gson().toJson(chainInfo)));
context.write(
new Text(chainInfo.getCallEntrance() + "@#!" + SummaryType.MONTH.getValue()), new Text(new Gson().toJson(chainInfo)));
}
} catch (Exception e) {
logger.error("Failed to mapper call chain[" + key.toString() + "]",

15
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/ChainBuildReducer.java
View File

@ -3,6 +3,7 @@ package com.ai.cloud.skywalking.analysis.chainbuild;
import com.ai.cloud.skywalking.analysis.chainbuild.entity.CallChainTree;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.chainbuild.po.SpecificTimeCallTreeMergedChainIdContainer;
import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType;
import com.ai.cloud.skywalking.analysis.config.Config;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
import com.google.gson.Gson;
@ -32,10 +33,16 @@ public class ChainBuildReducer extends Reducer<Text, Text, Text, IntWritable> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
doReduceAction(Bytes.toString(key.getBytes()), values.iterator());
String reduceKey = Bytes.toString(key.getBytes());
int index = reduceKey.indexOf("@#!");
if (index == -1){
return;
}
doReduceAction(reduceKey.substring(0, index - 1), SummaryType.convert(reduceKey.substring(index + 3)), values.iterator());
}
public void doReduceAction(String key, Iterator<Text> chainInfoIterator)
public void doReduceAction(String key, SummaryType summaryType, Iterator<Text> chainInfoIterator)
throws IOException, InterruptedException {
CallChainTree chainTree = CallChainTree.load(key);
SpecificTimeCallTreeMergedChainIdContainer container
@ -46,7 +53,7 @@ public class ChainBuildReducer extends Reducer<Text, Text, Text, IntWritable> {
try {
chainInfo = new Gson().fromJson(callChainData, ChainInfo.class);
container.addMergedChainIfNotContain(chainInfo);
chainTree.summary(chainInfo);
chainTree.summary(chainInfo, summaryType);
} catch (Exception e) {
logger.error(
"Failed to summary call chain, maybe illegal data:"
@ -54,7 +61,7 @@ public class ChainBuildReducer extends Reducer<Text, Text, Text, IntWritable> {
}
}
try {
container.saveToHBase();
container.saveToHBase(summaryType);
chainTree.saveToHbase();
} catch (Exception e) {
logger.error("Failed to save summaryresult/chainTree.", e);

5
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTree.java
View File

@ -6,6 +6,7 @@ import java.util.Map;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode;
import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType;
import com.ai.cloud.skywalking.analysis.chainbuild.util.TokenGenerator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -35,7 +36,7 @@ public class CallChainTree {
return chain;
}
public void summary(ChainInfo chainInfo) throws IOException {
public void summary(ChainInfo chainInfo, SummaryType summaryType) throws IOException {
for (ChainNode node : chainInfo.getNodes()) {
CallChainTreeNode newCallChainTreeNode = new CallChainTreeNode(node);
CallChainTreeNode callChainTreeNode = nodes.get(newCallChainTreeNode.getTreeNodeId());
@ -43,7 +44,7 @@ public class CallChainTree {
callChainTreeNode = newCallChainTreeNode;
nodes.put(newCallChainTreeNode.getTreeNodeId(), callChainTreeNode);
}
callChainTreeNode.summary(treeToken, node);
callChainTreeNode.summary(treeToken, node, summaryType);
}
}

20
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/entity/CallChainTreeNode.java
View File

@ -1,6 +1,7 @@
package com.ai.cloud.skywalking.analysis.chainbuild.entity;
import com.ai.cloud.skywalking.analysis.chainbuild.po.ChainNode;
import com.ai.cloud.skywalking.analysis.chainbuild.po.SummaryType;
import com.ai.cloud.skywalking.analysis.chainbuild.util.HBaseUtil;
import com.ai.cloud.skywalking.analysis.config.Config;
import com.ai.cloud.skywalking.analysis.config.HBaseTableMetaData;
@ -60,14 +61,21 @@ public class CallChainTreeNode {
this.viewPointId = node.getViewPoint();
}
public void summary(String treeId, ChainNode node) throws IOException {
public void summary(String treeId, ChainNode node, SummaryType summaryType) throws IOException {
Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date(node.getStartDate()));
summaryMinResult(treeId, node, calendar);
summaryHourResult(treeId, node, calendar);
summaryDayResult(treeId, node, calendar);
summaryMonthResult(treeId, node, calendar);
if (summaryType == SummaryType.MINUTER) {
summaryMinResult(treeId, node, calendar);
}
if (summaryType == SummaryType.DAY) {
summaryHourResult(treeId, node, calendar);
}
if (summaryType == SummaryType.DAY) {
summaryDayResult(treeId, node, calendar);
}
if (summaryType == SummaryType.MONTH) {
summaryMonthResult(treeId, node, calendar);
}
}
private void summaryMonthResult(String treeId, ChainNode node, Calendar calendar) throws IOException {

15
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SpecificTimeCallTreeMergedChainIdContainer.java
View File

@ -42,9 +42,9 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
//
if (chainInfo.getChainStatus() == ChainInfo.ChainStatus.NORMAL) {
callChainDetailMap.put(chainInfo.getCID(), new CallChainDetailForMysql(chainInfo,treeToken));
callChainDetailMap.put(chainInfo.getCID(), new CallChainDetailForMysql(chainInfo, treeToken));
}
}else{
} else {
}
}
@ -56,14 +56,17 @@ public class SpecificTimeCallTreeMergedChainIdContainer {
return treeToken + "@" + calendar.get(Calendar.YEAR) + "-" + calendar.get(Calendar.MONTH);
}
public void saveToHBase() throws IOException, InterruptedException, SQLException {
public void saveToHBase(SummaryType summaryType) throws IOException, InterruptedException, SQLException {
batchSaveCurrentHasBeenMergedChainInfo();
batchSaveMergedChainId();
batchSaveToMysql();
batchSaveToMysql(summaryType);
}
private void batchSaveToMysql() throws SQLException {
for (Map.Entry<String, CallChainDetailForMysql> entry : callChainDetailMap.entrySet()){
private void batchSaveToMysql(SummaryType summaryType) throws SQLException {
if (summaryType != SummaryType.MONTH) {
return;
}
for (Map.Entry<String, CallChainDetailForMysql> entry : callChainDetailMap.entrySet()) {
entry.getValue().saveToMysql();
}
}

31
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/chainbuild/po/SummaryType.java Normal file
View File

@ -0,0 +1,31 @@
package com.ai.cloud.skywalking.analysis.chainbuild.po;
public enum SummaryType {
MINUTER('m'), HOUR('H'), DAY('D'), MONTH('M');
private char value;
SummaryType(char value) {
this.value = value;
}
public static SummaryType convert(String value) {
char valueChar = value.charAt(0);
switch (valueChar) {
case 'm':
return MINUTER;
case 'H':
return HOUR;
case 'D':
return DAY;
case 'M':
return MONTH;
default:
throw new RuntimeException("Can not find the summary type[" + value + "]");
}
}
public char getValue() {
return value;
}
}