From bf0221d93297fb6ee2b1271163e75e8f5b71cae1 Mon Sep 17 00:00:00 2001 From: ascrutae Date: Sat, 23 Jan 2016 17:16:42 +0800 Subject: [PATCH] =?UTF-8?q?1.=E4=BF=AE=E5=A4=8DReduce=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/AnalysisServerDriver.java | 13 ++--- .../Categorize2ChainMapper.java | 7 ++- .../Categorize2ChainReducer.java | 5 +- .../categorize2chain/ChainRelate.java | 12 ++--- .../UncategorizeChainInfo.java | 15 ++++++ .../categorize2chain/model/ChainInfo.java | 53 ++++++------------- 6 files changed, 46 insertions(+), 59 deletions(-) diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java index ff4d2fa4a..fc9ba3261 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/AnalysisServerDriver.java @@ -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); diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java index 1a45201c2..82f2d9879 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainMapper.java @@ -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 { +public class Categorize2ChainMapper extends TableMapper { private Logger logger = LoggerFactory.getLogger(Categorize2ChainMapper.class.getName()); @Override @@ -35,7 +34,7 @@ public class Categorize2ChainMapper extends TableMapper 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 chainInfoIterator) throws IOException, InterruptedException { diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/ChainRelate.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/ChainRelate.java index 2e6441e92..beabfd55a 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/ChainRelate.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/ChainRelate.java @@ -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 categorizedChainInfoMap = new HashMap(); - private List uncategorizeChainInfoList = new ArrayList(); + private Set uncategorizeChainInfoList = new HashSet(); private Map chainDetailMap = new HashMap(); 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 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 getUncategorizeChainInfoList() { + public Set getUncategorizeChainInfoList() { return uncategorizeChainInfoList; } diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/UncategorizeChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/UncategorizeChainInfo.java index b3e0ce278..f938a8e2e 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/UncategorizeChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/UncategorizeChainInfo.java @@ -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(); diff --git a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/model/ChainInfo.java b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/model/ChainInfo.java index 533a68988..14a7d08e1 100644 --- a/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/model/ChainInfo.java +++ b/skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/model/ChainInfo.java @@ -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(); - 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>() { + }.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) {