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

1.修复Reduce bug

This commit is contained in:
ascrutae 2016-01-23 17:16:42 +08:00
parent 9a1bed87fd
commit bf0221d932
6 changed files with 46 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java
View File

@ -4,17 +4,14 @@ 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.config.ConfigInitializer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.mapred.JobContext;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
@ -40,10 +37,10 @@ public class AnalysisServerDriver extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
// ConfigInitializer.initialize();
// ConfigInitializer.initialize();
Configuration conf = new Configuration();
conf.set("hbase.zookeeper.quorum", Config.HBase.ZK_QUORUM);
conf.set("hbase.zookeeper.property.clientPort", Config.HBase.ZK_CLIENT_PORT);
conf.set("hbase.zookeeper.property.clientPort", Config.HBase.ZK_CLIENT_PORT);
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: AnalysisServer yyyy-MM-dd/HH:mm:ss yyyy-MM-dd/HH:mm:ss");
@ -55,9 +52,9 @@ public class AnalysisServerDriver extends Configured implements Tool {
Scan scan = buildHBaseScan(args);
TableMapReduceUtil.initTableMapperJob(Config.HBase.TABLE_CALL_CHAIN, scan, Categorize2ChainMapper.class,
ImmutableBytesWritable.class, ChainInfo.class, job);
Text.class, ChainInfo.class, job);
int regions = MetaTableAccessor.getRegionCount(conf, TableName.valueOf(Config.HBase.TABLE_CHAIN_SUMMARY));
if (regions == 0){
if (regions == 0) {
regions = 1;
}
job.setReducerClass(Categorize2ChainReducer.class);

7
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java
View File

@ -11,15 +11,14 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.mapred.JobContext;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
public class Categorize2ChainMapper extends TableMapper<ImmutableBytesWritable, ChainInfo> {
public class Categorize2ChainMapper extends TableMapper<Text, ChainInfo> {
private Logger logger = LoggerFactory.getLogger(Categorize2ChainMapper.class.getName());
@Override
@ -35,7 +34,7 @@ public class Categorize2ChainMapper extends TableMapper<ImmutableBytesWritable,
chainInfo = spanToChainInfo(key.toString(), spanList);
logger.info("Success convert span to chain info...." + chainInfo.getCID());
context.write(new ImmutableBytesWritable((chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()).getBytes()), chainInfo);
context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo);
} catch (Exception e) {
logger.error("Failed to mapper call chain[" + key.toString() + "]", e);
}

5
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReducer.java
View File

@ -1,8 +1,8 @@
package com.ai.cloud.skywalking.analysis.categorize2chain;
import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
import com.ai.cloud.skywalking.analysis.util.HBaseUtil;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
@ -17,9 +17,8 @@ public class Categorize2ChainReducer extends Reducer<Text, ChainInfo, Text, IntW
@Override
protected void reduce(Text key, Iterable<ChainInfo> values, Context context) throws IOException, InterruptedException {
//ConfigInitializer.initialize();
int totalCount = reduceAction(key.toString(), values.iterator());
context.write(key, new IntWritable(totalCount));
context.write(new Text(key.toString()), new IntWritable(totalCount));
}
public static int reduceAction(String key, Iterator<ChainInfo> chainInfoIterator) throws IOException, InterruptedException {

12
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/ChainRelate.java
View File

@ -6,6 +6,7 @@ import com.ai.cloud.skywalking.analysis.config.Constants;
import com.ai.cloud.skywalking.analysis.util.HBaseUtil;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.hadoop.hbase.client.Put;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -19,7 +20,7 @@ public class ChainRelate {
private String key;
private Map<String, CategorizedChainInfo> categorizedChainInfoMap = new HashMap<String, CategorizedChainInfo>();
private List<UncategorizeChainInfo> uncategorizeChainInfoList = new ArrayList<UncategorizeChainInfo>();
private Set<UncategorizeChainInfo> uncategorizeChainInfoList = new HashSet<UncategorizeChainInfo>();
private Map<String, ChainDetail> chainDetailMap = new HashMap<String, ChainDetail>();
public ChainRelate(String key) {
@ -46,17 +47,12 @@ public class ChainRelate {
isContained = true;
} else if (entry.getValue().isContained(child)) {
entry.getValue().add(child);
chainDetailMap.put(child.getCID(), new ChainDetail(child.getChainInfo()));
isContained = true;
}
}
if (!isContained) {
uncategorizeChainInfoList.add(child);
if (!uncategorizeChainInfoList.contains(child)) {
chainDetailMap.put(child.getCID(), new ChainDetail(child.getChainInfo()));
}
}
}
@ -110,7 +106,7 @@ public class ChainRelate {
Put put = new Put(getKey().getBytes());
put.addColumn(Config.HBase.CHAIN_RELATIONSHIP_COLUMN_FAMILY.getBytes(), Constants.UNCATEGORIZED_QUALIFIER_NAME.getBytes()
, new Gson().toJson(getUncategorizeChainInfoList()).getBytes());
, new GsonBuilder().excludeFieldsWithoutExposeAnnotation().create().toJson(getUncategorizeChainInfoList()).getBytes());
for (Map.Entry<String, CategorizedChainInfo> entry : getCategorizedChainInfoMap().entrySet()) {
put.addColumn(Config.HBase.CHAIN_RELATIONSHIP_COLUMN_FAMILY.getBytes(), entry.getKey().getBytes()
@ -137,7 +133,7 @@ public class ChainRelate {
return categorizedChainInfoMap;
}
public List<UncategorizeChainInfo> getUncategorizeChainInfoList() {
public Set<UncategorizeChainInfo> getUncategorizeChainInfoList() {
return uncategorizeChainInfoList;
}

15
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/UncategorizeChainInfo.java
View File

@ -45,6 +45,21 @@ public class UncategorizeChainInfo {
return chainInfo;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (!(o instanceof UncategorizeChainInfo)) return false;
UncategorizeChainInfo that = (UncategorizeChainInfo) o;
return cid != null ? cid.equals(that.cid) : that.cid == null;
}
@Override
public int hashCode() {
return cid != null ? cid.hashCode() : 0;
}
@Override
public String toString() {
GsonBuilder gsonBuilder = new GsonBuilder();

53
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/model/ChainInfo.java
View File

@ -1,6 +1,10 @@
package com.ai.cloud.skywalking.analysis.categorize2chain.model;
import com.ai.cloud.skywalking.analysis.util.TokenGenerator;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
@ -28,42 +32,18 @@ public class ChainInfo implements Writable {
@Override
public void write(DataOutput out) throws IOException {
out.write(cid.getBytes());
out.writeChar(chainStatus.getValue());
out.write(userId.getBytes());
out.writeInt(nodes.size());
for (ChainNode chainNode : nodes) {
out.write(chainNode.getNodeToken().getBytes());
out.write(chainNode.getViewPoint().getBytes());
out.writeChar(chainNode.getStatus().getValue());
out.writeLong(chainNode.getCost());
out.write(chainNode.getParentLevelId().getBytes());
out.writeInt(chainNode.getLevelId());
out.write(chainNode.getBusinessKey().getBytes());
}
out.write(new Gson().toJson(this).getBytes());
}
@Override
public void readFields(DataInput in) throws IOException {
cid = in.readLine();
chainStatus = ChainStatus.convert(in.readChar());
userId = in.readLine();
int nodeSize = in.readInt();
this.nodes = new ArrayList<ChainNode>();
for (int i = 0; i < nodeSize; i++) {
ChainNode chainNode = new ChainNode();
chainNode.setNodeToken(in.readLine());
chainNode.setViewPoint(in.readLine());
chainNode.setStatus(ChainNode.NodeStatus.convert(in.readChar()));
chainNode.setCost(in.readLong());
chainNode.setParentLevelId(in.readLine());
chainNode.setLevelId(in.readInt());
chainNode.setBusinessKey(in.readLine());
nodes.add(chainNode);
}
JsonObject jsonObject = (JsonObject) new JsonParser().parse(in.readLine());
cid = jsonObject.get("cid").getAsString();
chainStatus = ChainStatus.convert(jsonObject.get("chainStatus").getAsCharacter());
nodes = new Gson().fromJson(jsonObject.get("nodes"),
new TypeToken<List<ChainNode>>() {
}.getType());
userId = jsonObject.get("userId").getAsString();
}
public String getCID() {
@ -129,10 +109,6 @@ public class ChainInfo implements Writable {
this.value = value;
}
public char getValue() {
return value;
}
public static ChainStatus convert(char value) {
switch (value) {
case 'N':
@ -143,6 +119,11 @@ public class ChainInfo implements Writable {
throw new IllegalStateException("Failed to convert[" + value + "]");
}
}
@Override
public String toString() {
return value + "";
}
}
public void setCID(String cid) {