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

修复MapReduce部分bug

This commit is contained in:
ascrutae 2016-01-23 16:11:47 +08:00
parent 7614ca8f86
commit 5d8203b36e
9 changed files with 159 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
skywalking-analysis/pom.xml
View File

@ -25,16 +25,19 @@
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.1.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
<version>1.1.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.4.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.ai.cloud</groupId>
@ -77,6 +80,58 @@
<encoding>${project.build.sourceEncoding}</encoding>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<archive>
<manifest>
<addClasspath>true</addClasspath>
<classpathPrefix>libs/</classpathPrefix>
<mainClass>com.ai.cloud.skywalking.analysis.AnalysisServerDriver</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<phase>package</phase>
</execution>
</executions>
</plugin>
<!--<plugin>-->
<!--<artifactId>maven-dependency-plugin</artifactId>-->
<!--<executions>-->
<!--<execution>-->
<!--<id>copy-dependencies</id>-->
<!--<phase>prepare-package</phase>-->
<!--<goals>-->
<!--<goal>copy-dependencies</goal>-->
<!--</goals>-->
<!--</execution>-->
<!--</executions>-->
<!--<configuration>-->
<!--<excludeGroupIds>org.apache.hbase,org.apache.hadoop</excludeGroupIds>-->
<!--<outputDirectory>${project.build.directory}/classes/libs</outputDirectory>-->
<!--</configuration>-->
<!--</plugin>-->
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.4</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-my-jar-with-dependencies</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

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

@ -1,25 +1,32 @@
package com.ai.cloud.skywalking.analysis;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
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.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper;
import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReduce;
import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo;
import com.ai.cloud.skywalking.analysis.config.Config;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
public class AnalysisServerDriver extends Configured implements Tool {
@ -33,20 +40,29 @@ public class AnalysisServerDriver extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
// 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);
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");
System.exit(2);
}
Job job = Job.getInstance(conf);
job.setJarByClass(AnalysisServerDriver.class);
Scan scan = buildHBaseScan(args);
TableMapReduceUtil.initTableMapperJob(Config.HBase.CALL_CHAIN_TABLE_NAME, scan, Categorize2ChainMapper.class,
String.class, ChainInfo.class, job);
job.setReducerClass(Categorize2ChainReduce.class);
TableMapReduceUtil.initTableMapperJob(Config.HBase.TABLE_CALL_CHAIN, scan, Categorize2ChainMapper.class,
ImmutableBytesWritable.class, ChainInfo.class, job);
int regions = MetaTableAccessor.getRegionCount(conf, TableName.valueOf(Config.HBase.TABLE_CHAIN_SUMMARY));
if (regions == 0){
regions = 1;
}
job.setReducerClass(Categorize2ChainReducer.class);
job.setNumReduceTasks(regions);
FileOutputFormat.setOutputPath(job, new Path("/tmp/mr/mySummaryFile"));
return job.waitForCompletion(true) ? 0 : 1;
}

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

@ -6,20 +6,20 @@ import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo;
import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainNode;
import com.ai.cloud.skywalking.analysis.util.HBaseUtil;
import com.ai.cloud.skywalking.protocol.Span;
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.apache.hadoop.io.RawComparator;
import org.apache.hadoop.mapred.JobContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
public class Categorize2ChainMapper extends TableMapper<Text, ChainInfo> {
public class Categorize2ChainMapper extends TableMapper<ImmutableBytesWritable, ChainInfo> {
private Logger logger = LoggerFactory.getLogger(Categorize2ChainMapper.class.getName());
@Override
@ -34,8 +34,8 @@ public class Categorize2ChainMapper extends TableMapper<Text, ChainInfo> {
}
chainInfo = spanToChainInfo(key.toString(), spanList);
context.write(new Text(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()), chainInfo);
logger.info("Success convert span to chain info...." + chainInfo.getCID());
context.write(new ImmutableBytesWritable((chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken()).getBytes()), chainInfo);
} catch (Exception e) {
logger.error("Failed to mapper call chain[" + key.toString() + "]", e);
}

32
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReduce.java → skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/Categorize2ChainReducer.java
View File

@ -1,26 +1,29 @@
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.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;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.ai.cloud.skywalking.analysis.categorize2chain.model.ChainInfo;
import com.ai.cloud.skywalking.analysis.util.HBaseUtil;
public class Categorize2ChainReduce extends TableReducer<Text, ChainInfo, Put> {
private static Logger logger = LoggerFactory.getLogger(Categorize2ChainReduce.class.getName());
public class Categorize2ChainReducer extends Reducer<Text, ChainInfo, Text, IntWritable> {
private static Logger logger = LoggerFactory.getLogger(Categorize2ChainReducer.class.getName());
@Override
protected void reduce(Text key, Iterable<ChainInfo> values, Context context) throws IOException, InterruptedException {
reduceAction(key.toString(), values.iterator());
//ConfigInitializer.initialize();
int totalCount = reduceAction(key.toString(), values.iterator());
context.write(key, new IntWritable(totalCount));
}
public static void reduceAction(String key, Iterator<ChainInfo> chainInfoIterator) throws IOException, InterruptedException {
public static int reduceAction(String key, Iterator<ChainInfo> chainInfoIterator) throws IOException, InterruptedException {
int totalCount = 0;
try {
ChainRelate chainRelate = HBaseUtil.selectCallChainRelationship(key.toString());
Summary summary = new Summary();
@ -32,6 +35,7 @@ public class Categorize2ChainReduce extends TableReducer<Text, ChainInfo, Put> {
} catch (Exception e) {
continue;
}
totalCount++;
}
chainRelate.save();
@ -39,5 +43,7 @@ public class Categorize2ChainReduce extends TableReducer<Text, ChainInfo, Put> {
} catch (Exception e) {
logger.error("Failed to reduce key[" + key + "]", e);
}
return totalCount;
}
}

6
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/categorize2chain/dao/CallChainInfoDao.java
View File

@ -17,10 +17,10 @@ public class CallChainInfoDao {
static {
try {
Class.forName(Config.MySql.driverClass);
connection = DriverManager.getConnection(Config.MySql.url, Config.MySql.userName, Config.MySql.password);
Class.forName(Config.MySql.DRIVER_CLASS);
connection = DriverManager.getConnection(Config.MySql.URL, Config.MySql.USERNAME, Config.MySql.PASSWORD);
} catch (ClassNotFoundException e) {
logger.error("Failed to find jdbc driver class[" + Config.MySql.driverClass + "]", e);
logger.error("Failed to find jdbc driver class[" + Config.MySql.DRIVER_CLASS + "]", e);
System.exit(-1);
} catch (SQLException e) {
logger.error("Failed to connection database.", e);

28
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/Config.java
View File

@ -3,7 +3,6 @@ package com.ai.cloud.skywalking.analysis.config;
public class Config {
public static class HBase {
public static String TABLE_CHAIN_DETAIL = "sw_chain_detail";
public static String TRACE_DETAIL_FAMILY_COLUMN = "chain_detail";
@ -11,22 +10,23 @@ public class Config {
public static String TRACE_INFO_COLUMN_FAMILY = "trace_info";
public static String CALL_CHAIN_TABLE_NAME = "sw-call-chain";
public static String ZK_QUORUM = "10.1.235.197,10.1.235.198,10.1.235.199";
public static String ZK_QUORUM;
public static String ZK_CLIENT_PORT = "29181";
public static String ZK_CLIENT_PORT;
public static String TRACE_INFO_TABLE_NAME = "trace-info";
public static String TRACE_INFO_TABLE_NAME = "trace_info";
public static String TABLE_CALL_CHAIN_RELATIONSHIP = "sw-chain-relationship";
public static String TABLE_CALL_CHAIN_RELATIONSHIP = "sw_chain_relationship";
public static String CHAIN_RELATIONSHIP_COLUMN_FAMILY = "chain-relationship";
public static String CHAIN_RELATIONSHIP_COLUMN_FAMILY = "chain_relationship";
public static String TABLE_CHAIN_INFO = "sw-chain-info";
public static String TABLE_CHAIN_INFO = "sw_chain_info";
public static String TABLE_CHAIN_SUMMARY = "sw-chain-summary";
public static String TABLE_CHAIN_SUMMARY = "sw_chain_summary";
public static String TABLE_CHAIN_DETAIL = "sw-chain-detail";
public static String TABLE_CALL_CHAIN = "sw-call-chain";
}
public static class TraceInfo {
@ -35,17 +35,17 @@ public class Config {
public static class MySql {
public static String url = "jdbc:mysql://10.1.228.202:31316/test";
public static String URL = "jdbc:mysql://10.1.228.202:31316/test";
public static String userName = "devrdbusr21";
public static String USERNAME = "devrdbusr21";
public static String password = "devrdbusr21";
public static String PASSWORD = "devrdbusr21";
public static String driverClass = "com.mysql.jdbc.Driver";
public static String DRIVER_CLASS = "com.mysql.jdbc.Driver";
}
public static class Filter {
public static String FILTER_PACKAGE_NAME;
public static String FILTER_PACKAGE_NAME = "com.ai.cloud.skywalking.analysis.categorize2chain.filter.impl";
}
public class ChainNodeSummary {

4
skywalking-analysis/src/main/java/com/ai/cloud/skywalking/analysis/config/ConfigInitializer.java
View File

@ -13,7 +13,9 @@ public class ConfigInitializer {
private static Logger logger = Logger.getLogger(ConfigInitializer.class.getName());
public static void initialize() {
InputStream inputStream = ConfigInitializer.class.getResourceAsStream("/config.properties");
InputStream inputStream = Thread.currentThread()
.getContextClassLoader().getResourceAsStream("/config.properties");
//InputStream inputStream = ConfigInitializer.class.getResourceAsStream("/config.properties");
if (inputStream == null) {
logger.log(Level.ALL, "No provider sky-walking certification documents, sky-walking api auto shutdown.");
} else {

35
skywalking-analysis/src/main/resources/config.properties
View File

@ -1,14 +1,35 @@
#拦截器的类中的包名
filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl
hbase.call_chain_table_name=sw-call-chain
hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199
hbase.zk_client_port=29181
hbase.trace_info_table_name=sw-trace-info
hbase.trace_detail_family_column=chain_detail
hbase.chain_summary_column_family=chain_summary
hbase.trace_info_column_family=trace_info
hbase.zk_quorum=10.1.235.197,10.1.235.198,10.1.235.199
hbase.zk_client_port=29181
hbase.trace_info_table_name=trace-info
hbase.table_call_chain_relationship=sw-chain-relationship
hbase.chain_relationship_column_family=chain-relationship
hbase.table_chain_info=sw-chain-info
hbase.table_chain_summary=sw-chain-summary
hbase.table_chain_detail=sw-chain-detail
hbase.table_call_chain=sw-call-chain
traceinfo.trace_info_column_cid=cid
mysql.url=jdbc:mysql://10.1.228.202:31316/test
mysql.username=devrdbusr21
mysql.password=devrdbusr21
mysql.driverclass=com.mysql.jdbc.Driver
mysql.driver_class=com.mysql.jdbc.Driver
filter.filter_package_name=com.ai.cloud.skywalking.analysis.filter.impl
chainnodesummary.interval=5

6
skywalking-analysis/src/test/java/com/ai/cloud/skywalking/analysis/mapper/CallChainMapperTest.java
View File

@ -2,7 +2,7 @@ package com.ai.cloud.skywalking.analysis.mapper;
import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainMapper;
import com.ai.cloud.skywalking.analysis.categorize2chain.Categorize2ChainReduce;
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;
@ -43,12 +43,12 @@ public class CallChainMapperTest {
List<ChainInfo> chainInfos = new ArrayList<ChainInfo>();
chainInfos.add(chainInfo);
Categorize2ChainReduce.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator());
Categorize2ChainReducer.reduceAction(chainInfo.getUserId() + ":" + chainInfo.getEntranceNodeToken(), chainInfos.iterator());
}
public static List<Span> selectByTraceId(String traceId) throws IOException {
List<Span> entries = new ArrayList<Span>();
Table table = connection.getTable(TableName.valueOf(Config.HBase.CALL_CHAIN_TABLE_NAME));
Table table = connection.getTable(TableName.valueOf(Config.HBase.TABLE_CALL_CHAIN));
Get g = new Get(Bytes.toBytes(traceId));
Result r = table.get(g);
for (Cell cell : r.rawCells()) {