Commit f40fd030 authored by ascrutae's avatar ascrutae
Browse files

提交最新的maper-reduce代码,统计部分未完成

parent 235ec75c
Loading
Loading
Loading
Loading
+0 −117
Original line number Diff line number Diff line
<?xml version="1.0" encoding="UTF-8"?>
<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
  <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_6" inherit-compiler-output="false">
    <output url="file://$MODULE_DIR$/target/classes" />
    <output-test url="file://$MODULE_DIR$/target/test-classes" />
    <content url="file://$MODULE_DIR$">
      <sourceFolder url="file://$MODULE_DIR$/src/main/java" isTestSource="false" />
      <sourceFolder url="file://$MODULE_DIR$/src/test/java" isTestSource="true" />
      <sourceFolder url="file://$MODULE_DIR$/src/main/resources" type="java-resource" />
      <excludeFolder url="file://$MODULE_DIR$/target" />
    </content>
    <orderEntry type="inheritedJdk" />
    <orderEntry type="sourceFolder" forTests="false" />
    <orderEntry type="library" scope="TEST" name="Maven: junit:junit:4.12" level="project" />
    <orderEntry type="library" scope="TEST" name="Maven: org.hamcrest:hamcrest-core:1.3" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-client:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-annotations:1.1.2" level="project" />
    <orderEntry type="module-library">
      <library name="Maven: jdk.tools:jdk.tools:1.7">
        <CLASSES>
          <root url="jar://D:/Programs/Java/jdk1.7.0_75/lib/tools.jar!/" />
        </CLASSES>
        <JAVADOC />
        <SOURCES />
      </library>
    </orderEntry>
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-common:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-protocol:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: commons-codec:commons-codec:1.9" level="project" />
    <orderEntry type="library" name="Maven: commons-io:commons-io:2.4" level="project" />
    <orderEntry type="library" name="Maven: commons-lang:commons-lang:2.6" level="project" />
    <orderEntry type="library" name="Maven: commons-logging:commons-logging:1.2" level="project" />
    <orderEntry type="library" name="Maven: com.google.guava:guava:12.0.1" level="project" />
    <orderEntry type="library" name="Maven: com.google.code.findbugs:jsr305:1.3.9" level="project" />
    <orderEntry type="library" name="Maven: com.google.protobuf:protobuf-java:2.5.0" level="project" />
    <orderEntry type="library" name="Maven: io.netty:netty-all:4.0.23.Final" level="project" />
    <orderEntry type="library" name="Maven: org.apache.zookeeper:zookeeper:3.4.6" level="project" />
    <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.6.1" level="project" />
    <orderEntry type="library" name="Maven: org.slf4j:slf4j-log4j12:1.6.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.htrace:htrace-core:3.1.0-incubating" level="project" />
    <orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-mapper-asl:1.9.13" level="project" />
    <orderEntry type="library" name="Maven: org.jruby.jcodings:jcodings:1.0.8" level="project" />
    <orderEntry type="library" name="Maven: org.jruby.joni:joni:2.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-auth:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.httpcomponents:httpclient:4.2.5" level="project" />
    <orderEntry type="library" name="Maven: org.apache.httpcomponents:httpcore:4.2.4" level="project" />
    <orderEntry type="library" name="Maven: org.apache.directory.server:apacheds-kerberos-codec:2.0.0-M15" level="project" />
    <orderEntry type="library" name="Maven: org.apache.directory.server:apacheds-i18n:2.0.0-M15" level="project" />
    <orderEntry type="library" name="Maven: org.apache.directory.api:api-asn1-api:1.0.0-M20" level="project" />
    <orderEntry type="library" name="Maven: org.apache.directory.api:api-util:1.0.0-M20" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-common:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-annotations:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.commons:commons-math3:3.1.1" level="project" />
    <orderEntry type="library" name="Maven: xmlenc:xmlenc:0.52" level="project" />
    <orderEntry type="library" name="Maven: commons-net:commons-net:3.1" level="project" />
    <orderEntry type="library" name="Maven: commons-el:commons-el:1.0" level="project" />
    <orderEntry type="library" name="Maven: commons-configuration:commons-configuration:1.6" level="project" />
    <orderEntry type="library" name="Maven: commons-digester:commons-digester:1.8" level="project" />
    <orderEntry type="library" name="Maven: commons-beanutils:commons-beanutils:1.7.0" level="project" />
    <orderEntry type="library" name="Maven: commons-beanutils:commons-beanutils-core:1.8.0" level="project" />
    <orderEntry type="library" name="Maven: org.apache.avro:avro:1.7.4" level="project" />
    <orderEntry type="library" name="Maven: com.thoughtworks.paranamer:paranamer:2.3" level="project" />
    <orderEntry type="library" name="Maven: org.xerial.snappy:snappy-java:1.0.4.1" level="project" />
    <orderEntry type="library" name="Maven: com.jcraft:jsch:0.1.42" level="project" />
    <orderEntry type="library" name="Maven: org.apache.commons:commons-compress:1.4.1" level="project" />
    <orderEntry type="library" name="Maven: org.tukaani:xz:1.0" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-core:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-common:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: javax.xml.bind:jaxb-api:2.2.2" level="project" />
    <orderEntry type="library" name="Maven: javax.xml.stream:stax-api:1.0-2" level="project" />
    <orderEntry type="library" name="Maven: javax.activation:activation:1.1" level="project" />
    <orderEntry type="library" name="Maven: io.netty:netty:3.6.2.Final" level="project" />
    <orderEntry type="library" name="Maven: com.github.stephenc.findbugs:findbugs-annotations:1.3.9-1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-server:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-procedure:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-common:tests:1.1.2" level="project" />
    <orderEntry type="library" scope="RUNTIME" name="Maven: org.apache.hbase:hbase-prefix-tree:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: commons-httpclient:commons-httpclient:3.1" level="project" />
    <orderEntry type="library" name="Maven: commons-collections:commons-collections:3.2.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-hadoop-compat:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hbase:hbase-hadoop2-compat:1.1.2" level="project" />
    <orderEntry type="library" name="Maven: com.yammer.metrics:metrics-core:2.2.0" level="project" />
    <orderEntry type="library" name="Maven: com.sun.jersey:jersey-core:1.9" level="project" />
    <orderEntry type="library" name="Maven: com.sun.jersey:jersey-server:1.9" level="project" />
    <orderEntry type="library" name="Maven: asm:asm:3.1" level="project" />
    <orderEntry type="library" name="Maven: commons-cli:commons-cli:1.2" level="project" />
    <orderEntry type="library" name="Maven: org.apache.commons:commons-math:2.2" level="project" />
    <orderEntry type="library" name="Maven: log4j:log4j:1.2.17" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:jetty:6.1.26" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:jetty-util:6.1.26" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:jetty-sslengine:6.1.26" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:jsp-2.1:6.1.14" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:jsp-api-2.1:6.1.14" level="project" />
    <orderEntry type="library" name="Maven: org.mortbay.jetty:servlet-api-2.5:6.1.14" level="project" />
    <orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-core-asl:1.9.13" level="project" />
    <orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-jaxrs:1.9.13" level="project" />
    <orderEntry type="library" name="Maven: tomcat:jasper-compiler:5.5.23" level="project" />
    <orderEntry type="library" name="Maven: tomcat:jasper-runtime:5.5.23" level="project" />
    <orderEntry type="library" name="Maven: org.jamon:jamon-runtime:2.3.1" level="project" />
    <orderEntry type="library" name="Maven: com.lmax:disruptor:3.3.0" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-client:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-app:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-common:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-client:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: com.sun.jersey:jersey-client:1.9" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-server-common:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-shuffle:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.fusesource.leveldbjni:leveldbjni-all:1.8" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-api:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-jobclient:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-hdfs:2.5.1" level="project" />
    <orderEntry type="library" name="Maven: commons-daemon:commons-daemon:1.0.13" level="project" />
    <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-core:2.4.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.logging.log4j:log4j-api:2.4.1" level="project" />
    <orderEntry type="module" module-name="skywalking-protocol" />
  </component>
</module>
 No newline at end of file
+0 −112
Original line number Diff line number Diff line
package com.ai.cloud.skywalking.analysis.categorize2chain;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.Cell;
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.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ai.cloud.skywalking.analysis.categorize2chain.filter.SpanNodeProcessChain;
import com.ai.cloud.skywalking.analysis.categorize2chain.filter.SpanNodeProcessFilter;
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainNode;
import com.ai.cloud.skywalking.analysis.categorize2chain.util.HBaseUtil;
import com.ai.cloud.skywalking.analysis.categorize2chain.util.SubLevelSpanCostCounter;
import com.ai.cloud.skywalking.analysis.chainbuild.util.VersionIdentifier;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;
import com.ai.cloud.skywalking.protocol.Span;

public class Categorize2ChainMapper extends TableMapper<Text, ChainInfo> {
	private Logger logger = LoggerFactory
			.getLogger(Categorize2ChainMapper.class.getName());

	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		ConfigInitializer.initialize();
	}

	@Override
	protected void map(ImmutableBytesWritable key, Result value, Context context)
			throws IOException, InterruptedException {
		if(!VersionIdentifier.enableAnaylsis(Bytes.toString(key.get()))){
			return;
		}
		
		List<Span> spanList = new ArrayList<Span>();
		ChainInfo chainInfo = null;
		try {
			for (Cell cell : value.rawCells()) {
				Span span = new Span(Bytes.toString(cell.getValueArray(),
						cell.getValueOffset(), cell.getValueLength()));
				spanList.add(span);
			}

			chainInfo = spanToChainInfo(Bytes.toString(key.get()), spanList);
			logger.info("Success convert span to chain info...."
					+ chainInfo.getCID() + " TraceId : " + Bytes.toString(key.get()));
			context.write(
					new Text(chainInfo.getUserId() + ":"
							+ chainInfo.getEntranceNodeToken()), chainInfo);
		} catch (Exception e) {
			logger.error("Failed to mapper call chain[" + key.toString() + "]",
					e);
		}
	}

	public static ChainInfo spanToChainInfo(String key, List<Span> spanList) {
		SubLevelSpanCostCounter costMap = new SubLevelSpanCostCounter();
		ChainInfo chainInfo = new ChainInfo();
		Collections.sort(spanList, new Comparator<Span>() {
			@Override
			public int compare(Span span1, Span span2) {
				String span1TraceLevel = span1.getParentLevel() + "."
						+ span1.getLevelId();
				String span2TraceLevel = span2.getParentLevel() + "."
						+ span2.getLevelId();
				return span1TraceLevel.compareTo(span2TraceLevel);
			}
		});

		Map<String, SpanEntry> spanEntryMap = mergeSpanDataSet(spanList);
		for (Map.Entry<String, SpanEntry> entry : spanEntryMap.entrySet()) {
			ChainNode chainNode = new ChainNode();
			SpanNodeProcessFilter filter = SpanNodeProcessChain
					.getProcessChainByCallType(entry.getValue().getSpanType());
			filter.doFilter(entry.getValue(), chainNode, costMap);
			chainInfo.addNodes(chainNode);
		}

		chainInfo.generateChainToken();
		HBaseUtil.saveCidTidMapping(key, chainInfo);
		return chainInfo;
	}

	private static Map<String, SpanEntry> mergeSpanDataSet(List<Span> spanList) {
		Map<String, SpanEntry> spanEntryMap = new LinkedHashMap<String, SpanEntry>();
		for (int i = spanList.size() - 1; i >= 0; i--) {
			Span span = spanList.get(i);
			SpanEntry spanEntry = spanEntryMap.get(span.getParentLevel() + "."
					+ span.getLevelId());
			if (spanEntry == null) {
				spanEntry = new SpanEntry();
				spanEntryMap.put(
						span.getParentLevel() + "." + span.getLevelId(),
						spanEntry);
			}
			spanEntry.setSpan(span);
		}
		return spanEntryMap;
	}
}
+0 −57
Original line number Diff line number Diff line
package com.ai.cloud.skywalking.analysis.categorize2chain;

import java.io.IOException;
import java.util.Iterator;

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 com.ai.cloud.skywalking.analysis.categorize2chain.entity.ChainRelationship;
import com.ai.cloud.skywalking.analysis.categorize2chain.entity.ChainSummaryWithoutRelationship;
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.categorize2chain.util.HBaseUtil;
import com.ai.cloud.skywalking.analysis.config.ConfigInitializer;

public class Categorize2ChainReducer extends Reducer<Text, ChainInfo, Text, IntWritable> {
    private static Logger logger = LoggerFactory.getLogger(Categorize2ChainReducer.class.getName());

    @Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		ConfigInitializer.initialize();
	}
    
    @Override
    protected void reduce(Text key, Iterable<ChainInfo> values, Context context) throws IOException, InterruptedException {
        int totalCount = reduceAction(key.toString(), values.iterator());
        context.write(new Text(key.toString()), new IntWritable(totalCount));
    }

    public static int reduceAction(String key, Iterator<ChainInfo> chainInfoIterator) throws IOException, InterruptedException {
        int totalCount = 0;
        try {
            ChainRelationship chainRelate = HBaseUtil.loadCallChainRelationship(key.toString());
            ChainSummaryWithoutRelationship summary = new ChainSummaryWithoutRelationship();
            while (chainInfoIterator.hasNext()) {
                ChainInfo chainInfo = chainInfoIterator.next();
                try {
                    chainRelate.categoryChain(chainInfo);
                    summary.summary(chainInfo);
                } catch (Exception e) {
                    continue;
                }
                totalCount++;
            }

            chainRelate.save();
            summary.save();
        } catch (Exception e) {
            logger.error("Failed to reduce key[" + key + "]", e);
        }

        return totalCount;
    }
}
+0 −77
Original line number Diff line number Diff line
package com.ai.cloud.skywalking.analysis.categorize2chain.entity;

import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainInfo;
import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainNode;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.regex.Pattern;

public class CategorizedChainInfo {
    private String cid;
    private String chainFullToken;

    private List<String> children_Token;

    public CategorizedChainInfo(ChainInfo chainInfo) {
        cid = chainInfo.getCID();

        StringBuilder stringBuilder = new StringBuilder();
        boolean flag = false;
        for (ChainNode chainNode : chainInfo.getNodes()) {
            if (flag) {
                stringBuilder.append(";");
            }
            stringBuilder.append(chainNode.getTraceLevelId() + "-" + chainNode.getNodeToken());
            flag = true;
        }

        chainFullToken = stringBuilder.toString();
        children_Token = new ArrayList<String>();
    }

    public CategorizedChainInfo(String value) {
        JsonObject jsonObject = (JsonObject) new JsonParser().parse(value);
        cid = jsonObject.get("chainToken").getAsString();
        chainFullToken = jsonObject.get("chainFullToken").getAsString();
        children_Token = new Gson().fromJson(jsonObject.get("children_Token"),
                new TypeToken<List<String>>() {
                }.getType());
    }

    public String getChainFullToken() {
        return chainFullToken;
    }

    public boolean isContained(UncategorizeChainInfo uncategorizeChainInfo) {
        Pattern pattern = Pattern.compile(uncategorizeChainInfo.getNodeRegEx());
        return pattern.matcher(this.chainFullToken).find();
    }

    public static void main(String[] args) {
        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(1453507835957L)));
    }

    public boolean isAlreadyContained(UncategorizeChainInfo uncategorizeChainInfo) {
        return children_Token.contains(uncategorizeChainInfo.getCID());
    }

    public void add(UncategorizeChainInfo uncategorizeChainInfo) {
        children_Token.add(uncategorizeChainInfo.getCID());
    }

    @Override
    public String toString() {
        return new Gson().toJson(this);
    }

    public List<String> getChildren_Token() {
        return children_Token;
    }
}
+0 −71
Original line number Diff line number Diff line
package com.ai.cloud.skywalking.analysis.categorize2chain.entity;

import java.util.HashMap;
import java.util.Map;

import com.ai.cloud.skywalking.analysis.categorize2chain.po.ChainNode;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;

public class ChainNodeSpecificTimeWindowSummary {
	public static final long INTERVAL = 1L;

    private String traceLevelId;
    private String nodeToken;
    // key : 分钟
    private Map<String, ChainNodeSpecificTimeWindowSummaryValue> summerValueMap;

    public static ChainNodeSpecificTimeWindowSummary newInstance(String traceLevelId, String nodeToken) {
        ChainNodeSpecificTimeWindowSummary cns = new ChainNodeSpecificTimeWindowSummary();
        cns.traceLevelId = traceLevelId;
        cns.nodeToken = nodeToken;
        return cns;
    }

    private ChainNodeSpecificTimeWindowSummary() {
        summerValueMap = new HashMap<String, ChainNodeSpecificTimeWindowSummaryValue>();
    }

    public ChainNodeSpecificTimeWindowSummary(String value) {
        JsonObject jsonObject = (JsonObject) new JsonParser().parse(value);
        traceLevelId = jsonObject.get("traceLevelId").getAsString();
        summerValueMap = new Gson().fromJson(jsonObject.get("summerValueMap").toString(),
                new TypeToken<Map<String, ChainNodeSpecificTimeWindowSummaryValue>>() {
                }.getType());
        nodeToken = jsonObject.get("nodeToken").getAsString();
    }

    public String getTraceLevelId() {
        return traceLevelId;
    }

    public void summary(ChainNode node) {
        String key = generateKey(node.getStartDate());
        ChainNodeSpecificTimeWindowSummaryValue summaryResult = summerValueMap.get(key);
        if (summaryResult == null) {
            summaryResult = new ChainNodeSpecificTimeWindowSummaryValue();
            summerValueMap.put(key, summaryResult);
        }
        summaryResult.summary(node);
    }

    private String generateKey(long startTime) {
        long minutes = (startTime % (1000 * 60 * 60)) / (1000 * 60);
        return String.valueOf(minutes / INTERVAL);
    }

    @Override
    public String toString() {
        return new Gson().toJson(this);
    }

    public String getNodeToken() {
        return nodeToken;
    }

    public Map<String, ChainNodeSpecificTimeWindowSummaryValue> getSummerValueMap() {
        return summerValueMap;
    }
}
Loading