commit 5260718fbaf5b72d06902be1fe378520c0875d4f Author: 李玺康 Date: Tue Oct 27 14:51:16 2020 +0800 Storm版本liveCharts功能代码更新 diff --git a/.idea/compiler.xml b/.idea/compiler.xml new file mode 100644 index 0000000..27585f5 --- /dev/null +++ b/.idea/compiler.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/encodings.xml b/.idea/encodings.xml new file mode 100644 index 0000000..b26911b --- /dev/null +++ b/.idea/encodings.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..99d32dd --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,14 @@ + + + + + + + + + \ No newline at end of file diff --git a/.idea/storm-olap-aggregation.iml b/.idea/storm-olap-aggregation.iml new file mode 100644 index 0000000..ccc1c19 --- /dev/null +++ b/.idea/storm-olap-aggregation.iml @@ -0,0 +1,72 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..94a25f7 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..4ab0fc3 --- /dev/null +++ b/pom.xml @@ -0,0 +1,234 @@ + + 4.0.0 + + cn.ac.iie + storm-olap-aggregation + v3.20.09.23 + jar + + + storm-olap-aggregation + http://maven.apache.org + + + + + + nexus + Team Nexus Repository + http://192.168.40.125:8099/content/groups/public + + + + + maven-ali + http://maven.aliyun.com/nexus/content/groups/public/ + + true + + + true + always + fail + + + + + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 2.4.2 + + + package + + shade + + + + + cn.ac.iie.storm.topology.StreamAggregateTopology + + + META-INF/spring.handlers + + + META-INF/spring.schemas + + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 2.3.2 + + 1.8 + 1.8 + + + + + + properties + + **/*.properties + + false + + + properties + + log4j.properties + + false + + + + + + UTF-8 + 1.0.0 + 1.0.2 + 2.2.1 + 2.7.1 + + + + + + org.apache.kafka + kafka_2.11 + 1.0.0 + + + slf4j-log4j12 + org.slf4j + + + + + + org.apache.storm + storm-core + ${storm.version} + + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + org.apache.storm + storm-kafka + ${storm.version} + + + + junit + junit + 4.12 + test + + + + com.alibaba + fastjson + 1.2.70 + + + + cglib + cglib-nodep + 3.2.4 + + + + com.zdjizhi + galaxy + 1.0.3 + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.apache.zookeeper + zookeeper + 3.4.9 + + + slf4j-log4j12 + org.slf4j + + + log4j-over-slf4j + org.slf4j + + + + + + + org.junit.jupiter + junit-jupiter-api + 5.3.2 + compile + + + + + org.apache.httpcomponents + httpclient + 4.5.2 + + + + org.apache.httpcomponents + httpcore + 4.4.1 + + + + + + + + + + + + + + + + + diff --git a/properties/service_flow_config.properties b/properties/service_flow_config.properties new file mode 100644 index 0000000..8394de4 --- /dev/null +++ b/properties/service_flow_config.properties @@ -0,0 +1,60 @@ +#管理kafka地址 +bootstrap.servers=192.168.44.12:9092 + +#latest/earliest +auto.offset.reset=latest + +#压缩模式 none or snappy +kafka.compression.type=none + +#kafka broker下的topic名称 +kafka.topic=CONNECTION-RECORD-COMPLETED-LOG + +#读取topic,存储该spout id的消费offset信息，可通过该拓扑命名;具体存储offset的位置，确定下次读取不重复的数据； +group.id=test-20201026 + +#输出topic +results.bootstrap.servers=192.168.44.12:9092 + +#输出topic +results.output.topic=TRAFFIC-PROTOCOL-STAT-LOG + +#聚合时间,单位秒 +agg.time=15 + +#更新APP-ID时间 +update.app.id.time=60 + +#storm topology workers +topology.workers=1 + +#spout并行度 建议与kafka分区数相同 +spout.parallelism=1 + +#处理补全操作的bolt并行度-worker的倍数 +datacenter.bolt.parallelism=1 + +#写入kafka的并行度10 +kafka.bolt.parallelism=1 + +#kafka批量条数 +batch.insert.num=2000 + +#网关的schema位置 +schema.http=http://192.168.44.67:9999/metadata/schema/v1/fields/liveChart + +#网关的schema位置 +app.id.http=http://192.168.44.67:9999/open-api/appDicList + +#数据中心（UID） +data.center.id.num=15 + +#ack设置 1启动ack 0不启动ack +topology.num.acks=0 + +#spout接收睡眠时间 +topology.spout.sleep.time=1 + +#允许发送kafka最大失败数 +max.failure.num=20 + diff --git a/src/main/java/cn/ac/iie/storm/bolt/AggregateBolt.java b/src/main/java/cn/ac/iie/storm/bolt/AggregateBolt.java new file mode 100644 index 0000000..a17fbd5 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/bolt/AggregateBolt.java @@ -0,0 +1,177 @@ +package cn.ac.iie.storm.bolt; + +import cn.ac.iie.storm.utils.combine.AggregateUtils; +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import cn.ac.iie.storm.utils.http.HttpClientUtil; +import com.alibaba.fastjson.JSONObject; +import org.apache.log4j.Logger; +import org.apache.storm.Config; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseBasicBolt; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.TupleUtils; + +import java.util.HashMap; +import java.util.Map; + +/** + * @ClassNameAggregateBolt + * @Author lixkvip@126.com + * @Date2020/6/24 13:39 + * @Version V1.0 + **/ +public class AggregateBolt extends BaseBasicBolt { + private final static Logger logger = Logger.getLogger(AggregateBolt.class); + private static final long serialVersionUID = -7666031217706448622L; + private HashMap metricsMap; + private HashMap actionMap; + private HashMap cacheMap; + private static String timestamp; + + /** + * 只会在程序初始化的时候执行一次 + * + * @param stormConf + * @param context + */ + @Override + public void prepare(Map stormConf, TopologyContext context) { +// timestampValue = System.currentTimeMillis() / 1000; + String schema = HttpClientUtil.requestByGetMethod(StreamAggregateConfig.SCHEMA_HTTP); + timestamp = AggregateUtils.getTimeMetric(schema); + cacheMap = new HashMap<>(32); + + // TODO 获取action标签的内容 + actionMap = AggregateUtils.getActionMap(schema); + metricsMap = AggregateUtils.getMetircsMap(schema); + + } + + @Override + public void execute(Tuple input, BasicOutputCollector collector) { + try { + if (TupleUtils.isTick(input)) { + long timestampValue = System.currentTimeMillis() / 1000; + for (String s : cacheMap.keySet()) { + JSONObject result = JSONObject.parseObject(s); + result.put(timestamp, timestampValue); + result.putAll(cacheMap.get(s)); + collector.emit(new Values(result.toString())); + } + cacheMap.clear(); + + } else { + String label = input.getStringByField("label"); + + System.out.println("recv label=================="+label); + //action中某个协议的所有function,如果没有就默认 + String[] metrics = actionMap.getOrDefault(label, actionMap.get("Default")); + + String dimensions = input.getStringByField("dimensions"); + + System.out.println("recv dimensions=============="+dimensions); + String message = input.getStringByField("message"); + + System.out.println("recv message==================="+message); + //一条数据 + JSONObject event = JSONObject.parseObject(message); + //数据中的key值 (protocol,device_id,isp) + //map中对应的数据，可能为空,为空就默认创建一个对象 + JSONObject cacheMessage = cacheMap.getOrDefault(dimensions, new JSONObject()); + //TODO 接下来遍历所有的函数，去metrics的Map中去找对应的方法，并执行 + for (String metric : metrics) { + String name = metricsMap.get(metric).getString("name"); + //可能为空 + String fieldName = metricsMap.get(name).getString("fieldName"); + String nameValue = cacheMessage.getString(name); + //map中的字段值 + nameValue = (nameValue == null) ? "0" : nameValue; + + String fieldNameValue = event.getString(fieldName); + //数据中的字段值 + fieldNameValue = (fieldNameValue == null) ? "0" : fieldNameValue; + + //TODO 每次新增函数，需要改动此处代码 + functionSet(name, cacheMessage, nameValue, fieldNameValue); + } + cacheMap.put(dimensions, cacheMessage); + + } + } catch (Exception e) { + logger.error("计算节点异常,异常信息:" + e); + } + + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + declarer.declare(new Fields("message")); + } + + @Override + public Map getComponentConfiguration() { + Map conf = new HashMap(16); + conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, StreamAggregateConfig.AGG_TIME); + return conf; + } + + /** + * 根据schema描述对应字段进行操作的 函数集合 + * + * @param name 函数名称 + * @param cacheMessage 结果集 + * @param nameValue 当前值 + * @param fieldNameValue 新加值 + */ + private static void functionSet(String name, JSONObject cacheMessage, String nameValue, String fieldNameValue) { + switch (name) { + case "sessions": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "c2s_byte_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "s2c_byte_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "c2s_pkt_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "s2c_pkt_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "c2s_ipfrag_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "s2c_ipfrag_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "s2c_tcp_lostlen": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "c2s_tcp_lostlen": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "c2s_tcp_unorder_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + case "s2c_tcp_unorder_num": + cacheMessage.put(name, AggregateUtils.longSum(Long.parseLong(nameValue), Long.parseLong(fieldNameValue))); + break; + + case "unique_sip_num": + //TODO + //cacheMessage.put(name, AggregateUtils.) + break; + case "unique_cip_num": + //TODO + break; + default: + break; + } + } +} diff --git a/src/main/java/cn/ac/iie/storm/bolt/ParseKvBolt.java b/src/main/java/cn/ac/iie/storm/bolt/ParseKvBolt.java new file mode 100644 index 0000000..2bb8062 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/bolt/ParseKvBolt.java @@ -0,0 +1,168 @@ +package cn.ac.iie.storm.bolt; + + +import cn.ac.iie.storm.utils.combine.AggregateUtils; +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import cn.ac.iie.storm.utils.http.HttpClientUtil; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import com.zdjizhi.utils.StringUtil; +import org.apache.log4j.Logger; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseBasicBolt; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.TupleUtils; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; + +import static cn.ac.iie.storm.utils.combine.AggregateUtils.transDimensions; +import static cn.ac.iie.storm.utils.combine.AggregateUtils.updateAppRelation; + +/** + * @ClassNameMyWindowBolt + * @Author lixkvip@126.com + * @Date2020/6/9 14:45 + * @Version V1.0 + **/ +public class ParseKvBolt extends BaseBasicBolt { + private final static Logger logger = Logger.getLogger(ParseKvBolt.class); + private static final long serialVersionUID = -999382396035310355L; + private JSONArray transforms; + private JSONArray dimensions; + private static HashMap appMap = new HashMap<>(32); + + + /** + * 此方法只在程序启动的时候执行一次，用来初始化 + * + * @param stormConf Map + * @param context TopologyContext + */ + @Override + public void prepare(Map stormConf, TopologyContext context) { + String schema = HttpClientUtil.requestByGetMethod(StreamAggregateConfig.SCHEMA_HTTP); + JSONObject jsonObject = JSONObject.parseObject(schema); + String data = JSONObject.parseObject(jsonObject.getString("data")).getString("doc"); + //TODO 解析 schema + transforms = JSONObject.parseArray(JSONObject.parseObject(data).getString("transforms")); + + //TODO 获取dimensions + dimensions = JSONObject.parseArray(JSONObject.parseObject(data).getString("dimensions")); + updateAppRelation(appMap); + + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector collector) { + try { + if (TupleUtils.isTick(tuple)) { + updateAppRelation(appMap); + } else { + //TODO 解析tuple的 message + JSONObject message = JSONObject.parseObject(tuple.getStringByField("source")); + + //TODO 新建一个dimensions的Json对象 + JSONObject dimensionsObj = transDimensions(dimensions, message); + + for (Object transform : transforms) { + JSONObject transformObj = JSONObject.parseObject(transform.toString()); + String function = transformObj.getString("function"); + String name = transformObj.getString("name"); + String fieldName = transformObj.getString("fieldName"); + String parameters = transformObj.getString("parameters"); + + switch (function) { + case "alignment": + if (StringUtil.isNotBlank(parameters)) { + if (message.containsKey(fieldName)) { + alignmentUtils(parameters, message, name, fieldName); + } + } + break; + case "combination": + if (StringUtil.isNotBlank(parameters)) { + combinationUtils(parameters, message, name, fieldName, dimensionsObj); + } + break; + case "hierarchy": + String hierarchyValue = message.getString(fieldName); + //TODO 执行拆分代码 + if (StringUtil.isNotBlank(hierarchyValue) && StringUtil.isNotBlank(parameters)) { + String[] hierarchyPars = parameters.split(StreamAggregateConfig.FORMAT_SPLITTER); + String[] dimensionsArr = hierarchyValue.split(hierarchyPars[0]); + //TODO 递归拼接tuple并发送出去 + AggregateUtils.reSend(1, dimensionsArr, "", collector, message, dimensionsObj, name); + } + break; + default: + //数据原样输出 + collector.emit(new Values(null, null, message.toString())); + break; + } + } + } + } catch (Exception e) { + logger.error("上层解析原始日志/拼接计算日志发送异常,异常信息:" + e); + e.printStackTrace(); + } + + } + + @Override + public Map getComponentConfiguration() { + Map conf = new HashMap<>(16); + conf.put(org.apache.storm.Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, StreamAggregateConfig.UPDATE_APP_ID_TIME); + return conf; + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + declarer.declare(new Fields("label", "dimensions", "message")); + } + + /** + * alignment ID替换操作 + * + * @param parameters 参数集 + * @param message 原始日志 + * @param name 结果日志列名 + * @param fieldName 原始日志列名 + */ + private static void alignmentUtils(String parameters, JSONObject message, String name, String fieldName) { + String[] alignmentPars = parameters.split(StreamAggregateConfig.FORMAT_SPLITTER); + String data = message.getString(fieldName); + System.out.println("alignmentPars=" + Arrays.toString(alignmentPars) + "data=" + data); + int subscript = Integer.parseInt(alignmentPars[0]); + String[] fieldSplit = data.split(alignmentPars[1]); + Long appID = Long.valueOf(fieldSplit[subscript]); + int length = fieldSplit[subscript].length(); + StringBuilder sb = new StringBuilder(data); + message.put(name, sb.replace(0, length, appMap.get(appID))); + } + + /** + * combination 拼接操作 + * + * @param parameters 参数集 + * @param message 原始日志 + * @param name 结果日志列名 + * @param fieldName 原始日志列名 + * @param dimensionsObj 结果集 + */ + private static void combinationUtils(String parameters, JSONObject message, String name, String fieldName, JSONObject dimensionsObj) { + String[] combinationPars = parameters.split(StreamAggregateConfig.FORMAT_SPLITTER); + String parameter0Value = message.getString(combinationPars[0]); + if (StringUtil.isNotBlank(parameter0Value)) { + String combinationValue = parameter0Value + combinationPars[1] + message.getString(fieldName); + message.put(fieldName, combinationValue); + dimensionsObj.put(name, combinationValue); + } + } +} diff --git a/src/main/java/cn/ac/iie/storm/bolt/ResultSendBolt.java b/src/main/java/cn/ac/iie/storm/bolt/ResultSendBolt.java new file mode 100644 index 0000000..36e57a6 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/bolt/ResultSendBolt.java @@ -0,0 +1,44 @@ +package cn.ac.iie.storm.bolt; + + +import cn.ac.iie.storm.utils.common.LogSendKafka; +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import org.apache.log4j.Logger; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseBasicBolt; +import org.apache.storm.tuple.Tuple; + +import java.util.Map; + +/** + * 发送到kafka的bolt + * + * @author qidaijie + */ +public class ResultSendBolt extends BaseBasicBolt { + + private static final long serialVersionUID = 3237813470939823159L; + private static Logger logger = Logger.getLogger(ResultSendBolt.class); + private LogSendKafka logSendKafka; + + @Override + public void prepare(Map stormConf, TopologyContext context) { + logSendKafka = LogSendKafka.getInstance(); + } + + @Override + public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) { + try { + logSendKafka.sendMessage(tuple.getStringByField("message")); + } catch (Exception e) { + logger.error(StreamAggregateConfig.RESULTS_OUTPUT_TOPIC + "日志发送Kafka过程出现异常,异常信息:" + e); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } +} diff --git a/src/main/java/cn/ac/iie/storm/bolt/change/FilterBolt.java b/src/main/java/cn/ac/iie/storm/bolt/change/FilterBolt.java new file mode 100644 index 0000000..6093a94 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/bolt/change/FilterBolt.java @@ -0,0 +1,68 @@ +package cn.ac.iie.storm.bolt.change; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseBasicBolt; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; + +/** + * @ClassNameFilterBolt + * @Author lixkvip@126.com + * @Date2020/7/1 12:02 + * @Version V1.0 + **/ +public class FilterBolt extends BaseBasicBolt { + @Override + public void execute(Tuple input, BasicOutputCollector collector) { + JSONObject source = JSONObject.parseObject(input.getStringByField("source")); + String schema = ""; + + String data = JSONObject.parseObject(schema).getString("data"); + + String filters = JSONObject.parseObject(data).getString("filters"); + + boolean flag = true; + String type = JSONObject.parseObject(filters).getString("type"); + JSONArray fieldsArr = JSONObject.parseArray(JSONObject.parseObject(filters).getString("fields")); + if ("and".equals(type)) { + for (int i = 0; i < fieldsArr.size(); i++) { + + JSONObject field = JSONObject.parseObject(fieldsArr.get(i).toString()); + String name = field.getString("fieldName"); + String fieldType = field.getString("type"); + Object values = field.get("values"); + + Object nameValue = source.get(name); + + System.out.println("nameValue ========" +nameValue); + + if ("not".equals(fieldType)) { + + if (nameValue == values) { + //满足过滤条件 + flag = false; + } + + } else if ("in".equals(fieldType)) { + if (!values.toString().contains(nameValue.toString())) { + //满足过滤条件 + flag = false; + } + } + }} + + + + collector.emit(new Values(source)); + + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + declarer.declare(new Fields("filter")); + } +} diff --git a/src/main/java/cn/ac/iie/storm/bolt/print/PrintBolt.java b/src/main/java/cn/ac/iie/storm/bolt/print/PrintBolt.java new file mode 100644 index 0000000..e251f21 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/bolt/print/PrintBolt.java @@ -0,0 +1,40 @@ +package cn.ac.iie.storm.bolt.print; + +import org.apache.log4j.Logger; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseBasicBolt; +import org.apache.storm.tuple.Tuple; + +/** + * @ClassNamePrintBolt + * @Author lixkvip@126.com + * @Date2020/6/28 15:34 + * @Version V1.0 + **/ +public class PrintBolt extends BaseBasicBolt { + private final static Logger logger = Logger.getLogger(PrintBolt.class); + private static long a; + private long b; + public static long c; + + @Override + public void execute(Tuple input, BasicOutputCollector collector) { + logger.error("==================================一批数据========================="); + + a= System.currentTimeMillis(); + b= System.currentTimeMillis(); + c= System.currentTimeMillis(); + + + logger.error(Thread.currentThread() + "private static long a======:" + a); + logger.error(Thread.currentThread() + "private long b======:" + b); + logger.error(Thread.currentThread() + "public static long c======:" + c); + + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } +} diff --git a/src/main/java/cn/ac/iie/storm/spout/CustomizedKafkaSpout.java b/src/main/java/cn/ac/iie/storm/spout/CustomizedKafkaSpout.java new file mode 100644 index 0000000..fd6dfba --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/spout/CustomizedKafkaSpout.java @@ -0,0 +1,80 @@ +package cn.ac.iie.storm.spout; + +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.log4j.Logger; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseRichSpout; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; + +import java.util.Collections; +import java.util.Map; +import java.util.Properties; + +/** + * kafkaSpout + * + * @author Administrator + */ +public class CustomizedKafkaSpout extends BaseRichSpout { + private static final long serialVersionUID = -3363788553406229592L; + private KafkaConsumer consumer; + private SpoutOutputCollector collector = null; + private TopologyContext context = null; + private final static Logger logger = Logger.getLogger(CustomizedKafkaSpout.class); + + + private static Properties createConsumerConfig() { + Properties props = new Properties(); + props.put("bootstrap.servers", StreamAggregateConfig.BOOTSTRAP_SERVERS); + props.put("group.id", StreamAggregateConfig.GROUP_ID); + props.put("session.timeout.ms", "60000"); + props.put("max.poll.records", 3000); + props.put("max.partition.fetch.bytes", 31457280); + props.put("auto.offset.reset", StreamAggregateConfig.AUTO_OFFSET_RESET); + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + return props; + } + + @Override + public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { + // TODO Auto-generated method stub + this.collector = collector; + this.context = context; + Properties prop = createConsumerConfig(); + this.consumer = new KafkaConsumer<>(prop); + this.consumer.subscribe(Collections.singletonList(StreamAggregateConfig.KAFKA_TOPIC)); + } + + @Override + public void close() { + consumer.close(); + } + + @Override + public void nextTuple() { + try { + // TODO Auto-generated method stub + ConsumerRecords records = consumer.poll(10000L); + Thread.sleep(StreamAggregateConfig.TOPOLOGY_SPOUT_SLEEP_TIME); + for (ConsumerRecord record : records) { + this.collector.emit(new Values(record.value())); + } + } catch (Exception e) { + logger.error("KafkaSpout发送消息出现异常!", e); + e.printStackTrace(); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + // TODO Auto-generated method stub + declarer.declare(new Fields("source")); + } +} diff --git a/src/main/java/cn/ac/iie/storm/topology/StormRunner.java b/src/main/java/cn/ac/iie/storm/topology/StormRunner.java new file mode 100644 index 0000000..6e77c66 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/topology/StormRunner.java @@ -0,0 +1,35 @@ +package cn.ac.iie.storm.topology; + + +import org.apache.storm.Config; +import org.apache.storm.LocalCluster; +import org.apache.storm.StormSubmitter; +import org.apache.storm.generated.AlreadyAliveException; +import org.apache.storm.generated.AuthorizationException; +import org.apache.storm.generated.InvalidTopologyException; +import org.apache.storm.topology.TopologyBuilder; + +/** + * @author Administrator + */ +public final class StormRunner { + private static final int MILLS_IN_SEC = 1000; + + private StormRunner() {} + + public static void runTopologyLocally(TopologyBuilder builder, String topologyName, Config conf, int runtimeInSeconds) throws InterruptedException { + + LocalCluster localCluster = new LocalCluster(); + localCluster.submitTopology(topologyName, conf, builder.createTopology()); + Thread.sleep((long) runtimeInSeconds * MILLS_IN_SEC); + localCluster.shutdown(); + + } + + public static void runTopologyRemotely(TopologyBuilder builder, String topologyName, Config conf ) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException { + + StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, builder.createTopology()); + } + + +} diff --git a/src/main/java/cn/ac/iie/storm/topology/StreamAggregateTopology.java b/src/main/java/cn/ac/iie/storm/topology/StreamAggregateTopology.java new file mode 100644 index 0000000..ed38bce --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/topology/StreamAggregateTopology.java @@ -0,0 +1,103 @@ +package cn.ac.iie.storm.topology; + +import cn.ac.iie.storm.bolt.AggregateBolt; +import cn.ac.iie.storm.bolt.ResultSendBolt; +import cn.ac.iie.storm.bolt.ParseKvBolt; +import cn.ac.iie.storm.spout.CustomizedKafkaSpout; +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import org.apache.log4j.Logger; +import org.apache.storm.Config; +import org.apache.storm.generated.AlreadyAliveException; +import org.apache.storm.generated.AuthorizationException; +import org.apache.storm.generated.InvalidTopologyException; +import org.apache.storm.topology.TopologyBuilder; +import org.apache.storm.tuple.Fields; + +/** + * @ClassNameFlowAggregateTopo + * @Author lixkvip@126.com + * @Date2020/6/24 10:14 + * @Version V1.0 + **/ +public class StreamAggregateTopology { + + + private static Logger logger = Logger.getLogger(StreamAggregateTopology.class); + private final String topologyName; + private final Config topologyConfig; + private TopologyBuilder builder; + + private StreamAggregateTopology() { + this(StreamAggregateTopology.class.getSimpleName()); + } + + private StreamAggregateTopology(String topologyName) { + this.topologyName = topologyName; + topologyConfig = createTopologConfig(); + } + + /** + * 测试配置 + * conf.setTopologyWorkerMaxHeapSize(6144); + * conf.put(Config.WORKER_CHILDOPTS, "-Xmx4G -Xms2G"); + */ + private Config createTopologConfig() { + Config conf = new Config(); + conf.setDebug(false); + conf.setMessageTimeoutSecs(60); + conf.setMaxSpoutPending(StreamAggregateConfig.SPOUT_PARALLELISM); + conf.setNumAckers(StreamAggregateConfig.TOPOLOGY_NUM_ACKS); + return conf; + } + + private void runLocally() throws InterruptedException { + topologyConfig.setMaxTaskParallelism(1); + StormRunner.runTopologyLocally(builder, topologyName, topologyConfig, 6000); + } + + private void runRemotely() throws AlreadyAliveException, InvalidTopologyException, AuthorizationException { + topologyConfig.setNumWorkers(StreamAggregateConfig.TOPOLOGY_WORKERS); + //设置过高会导致很多问题，如心跳线程饿死、吞吐量大幅下跌 + topologyConfig.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 8); + StormRunner.runTopologyRemotely(builder, topologyName, topologyConfig); + } + + private void buildTopology() { + builder = new TopologyBuilder(); + builder.setSpout("CustomizedKafkaSpout", new CustomizedKafkaSpout(), StreamAggregateConfig.SPOUT_PARALLELISM); + + builder.setBolt("ParseKvBolt", new ParseKvBolt(), StreamAggregateConfig.DATACENTER_BOLT_PARALLELISM) + .localOrShuffleGrouping("CustomizedKafkaSpout"); + + builder.setBolt("AggregateBolt", new AggregateBolt(), StreamAggregateConfig.DATACENTER_BOLT_PARALLELISM) + .fieldsGrouping("ParseKvBolt", new Fields("dimensions")); + + builder.setBolt("ResultSendBolt", new ResultSendBolt(), StreamAggregateConfig.KAFKA_BOLT_PARALLELISM) + .localOrShuffleGrouping("AggregateBolt"); +// builder.setBolt("PrintBolt", new PrintBolt(), 3).localOrShuffleGrouping("LogFlowWriteSpout"); + + } + + public static void main(String[] args) throws Exception { + StreamAggregateTopology csst = null; + boolean runLocally = true; + String parameter = "remote"; + int size = 2; + if (args.length >= size && parameter.equalsIgnoreCase(args[1])) { + runLocally = false; + csst = new StreamAggregateTopology(args[0]); + } else { + csst = new StreamAggregateTopology(); + } + + csst.buildTopology(); + + if (runLocally) { + logger.info("执行本地模式..."); + csst.runLocally(); + } else { + logger.info("执行远程部署模式..."); + csst.runRemotely(); + } + } +} diff --git a/src/main/java/cn/ac/iie/storm/utils/combine/AggregateUtils.java b/src/main/java/cn/ac/iie/storm/utils/combine/AggregateUtils.java new file mode 100644 index 0000000..d172f87 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/utils/combine/AggregateUtils.java @@ -0,0 +1,208 @@ +package cn.ac.iie.storm.utils.combine; + +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import cn.ac.iie.storm.utils.http.HttpClientUtil; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import org.apache.log4j.Logger; +import org.apache.storm.topology.BasicOutputCollector; +import org.apache.storm.tuple.Values; + + +import java.util.HashMap; + +/** + * @ClassNameAggregateUtils + * @Author lixkvip@126.com + * @Date2020/6/23 14:04 + * @Version V1.0 + **/ +public class AggregateUtils { + private final static Logger logger = Logger.getLogger(AggregateUtils.class); + + /** + * Long类型的数据求和 + * + * @param value1 第一个值 + * @param value2 第二个值 + * @return value1 + value2 + */ + public static Long longSum(Long value1, Long value2) { + + return value1 + value2; + } + + /** + * 计算Count + * + * @param count 当前count值 + * @return count+1 + */ + public static Long count(Long count) { + + count++; + return count; + } + + + /** + * 返回指标列的Map集合 + * + * @param schema 动态获取的schema + * @return 指标列集合 + * （c2s_byte_len, { "function" : "c2s_byte_sum", "name" : "c2s_byte_len", "fieldName" : "common_c2s_byte_num" }） + */ + public static HashMap getMetircsMap(String schema) { + HashMap metricsMap = new HashMap<>(16); + + JSONObject jsonObject = JSONObject.parseObject(schema); + String data = jsonObject.getString("data"); + JSONArray metrics = JSONObject.parseArray(JSONObject.parseObject(JSONObject.parseObject(data).getString("doc")).getString("metrics")); + + for (Object metric : metrics) { + JSONObject json = JSONObject.parseObject(metric.toString()); + String name = json.getString("name"); + metricsMap.put(name, json); + } + + return metricsMap; + } + + + /** + * 递归发送tuple + * + * @param headIndex ssss + * @param splitArr + * @param initStr + * @param collector + * @param message + * @param dimesionsObj + * @param name + */ + public static void reSend(int headIndex, String[] splitArr, String initStr, BasicOutputCollector collector, JSONObject message, JSONObject dimesionsObj, String name) { + +// //递归拼接字符串 +// if (splitArr.length == headIndex - 1) { +// //什么也不做 +// } else { +// //递归的核心代码 +// if ("".equals(initStr)) { +// initStr = splitArr[splitArr.length - headIndex]; +// } else { +// initStr = initStr + "/" + splitArr[splitArr.length - headIndex]; +// } +// dimesionsObj.put(name, initStr); +// +// collector.emit(new Values(splitArr[splitArr.length - headIndex], dimesionsObj.toString(), message.toString())); +// reSend(headIndex + 1, splitArr, initStr, collector, message, dimesionsObj, name); +// } + + //递归拼接字符串 + if (splitArr.length != headIndex - 1) { + //递归的核心代码 + if ("".equals(initStr)) { + initStr = splitArr[splitArr.length - headIndex]; + } else { + initStr = initStr + "/" + splitArr[splitArr.length - headIndex]; + } + dimesionsObj.put(name, initStr); + System.out.println("发送之前================================"); + collector.emit(new Values(splitArr[splitArr.length - headIndex], dimesionsObj.toString(), message.toString())); + System.out.println("send label ==============="+splitArr[splitArr.length - headIndex]); + System.out.println("send dimesion============="+dimesionsObj.toString()); + + + reSend(headIndex + 1, splitArr, initStr, collector, message, dimesionsObj, name); + } + } + + + /** + * 获取action模块的Map集合 + * + * @param schema 动态获取的schema + * @return （HTTP，metrics数组） + */ + public static HashMap getActionMap(String schema) { + JSONObject jsonObject = JSONObject.parseObject(schema); + String data = jsonObject.getString("data"); + JSONArray actions = JSONObject.parseArray(JSONObject.parseObject(JSONObject.parseObject(data).getString("doc")).getString("action")); + HashMap map = new HashMap<>(16); + + for (Object action : actions) { + JSONObject json = JSONObject.parseObject(action.toString()); + + String label = json.getString("label"); + String[] metrics = json.getString("metrics").split(","); + + map.put(label, metrics); + } + return map; + } + + + /** + * 获取时间列的集合 + * + * @param schema 动态获取的schema + * @return 时间列 + */ + public static String getTimeMetric(String schema) { + JSONObject jsonObject = JSONObject.parseObject(schema); + String data = jsonObject.getString("data"); + + return JSONObject.parseObject(JSONObject.parseObject(JSONObject.parseObject(data) + .getString("doc")) + .getString("timestamp")) + .getString("name"); + } + + /** + * 更新缓存中的对应关系map + * + * @param hashMap 当前缓存对应关系map + */ + public static void updateAppRelation(HashMap hashMap) { + try { + Long begin = System.currentTimeMillis(); + String schema = HttpClientUtil.requestByGetMethod(StreamAggregateConfig.APP_ID_HTTP); + String data = JSONObject.parseObject(schema).getString("data"); + JSONArray objects = JSONArray.parseArray(data); + for (Object object : objects) { + JSONArray jsonArray = JSONArray.parseArray(object.toString()); + Long key = jsonArray.getLong(0); + String value = jsonArray.getString(1); + if (hashMap.containsKey(key)) { + if (!value.equals(hashMap.get(key))) { + hashMap.put(key, value); + } + } else { + hashMap.put(key, value); + } + + } + System.out.println((System.currentTimeMillis() - begin)); + logger.warn("更新缓存中的对应的APP关系,拉取接口数据长度:[" + objects.size()); + } catch (Exception e) { + logger.error("更新缓存APP-ID失败,异常：" + e); + } + } + + /** + * 解析 dimensions 字段集 + * + * @param dimensions 维度集 + * @param message 原始日志 + * @return 结果维度集 + */ + public static JSONObject transDimensions(JSONArray dimensions, JSONObject message) { + JSONObject dimensionsObj = new JSONObject(); + for (Object dimension : dimensions) { + String fieldName = JSONObject.parseObject(dimension.toString()).getString("fieldName"); + String name = JSONObject.parseObject(dimension.toString()).getString("name"); + dimensionsObj.put(name, message.get(fieldName)); + } + return dimensionsObj; + } +} diff --git a/src/main/java/cn/ac/iie/storm/utils/common/LogSendKafka.java b/src/main/java/cn/ac/iie/storm/utils/common/LogSendKafka.java new file mode 100644 index 0000000..378fea6 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/utils/common/LogSendKafka.java @@ -0,0 +1,76 @@ +package cn.ac.iie.storm.utils.common; + + +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import org.apache.kafka.clients.producer.*; +import org.apache.log4j.Logger; + +import java.util.Properties; + +/** + * NTC系统配置产生日志写入数据中心类 + * + * @author Administrator + * @create 2018-08-13 15:11 + */ + +public class LogSendKafka { + + private static Logger logger = Logger.getLogger(LogSendKafka.class); + + /** + * kafka生产者，用于向kafka中发送消息 + */ + private static Producer kafkaProducer; + + /** + * kafka生产者适配器（单例），用来代理kafka生产者发送消息 + */ + private static LogSendKafka logSendKafka; + + private LogSendKafka() { + initKafkaProducer(); + } + + public static LogSendKafka getInstance() { + if (logSendKafka == null) { + logSendKafka = new LogSendKafka(); + } + return logSendKafka; + } + + + public void sendMessage(String message) { + final int[] errorSum = {0}; + kafkaProducer.send(new ProducerRecord<>(StreamAggregateConfig.RESULTS_OUTPUT_TOPIC, message), new Callback() { + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception != null) { + logger.error("写入" + StreamAggregateConfig.RESULTS_OUTPUT_TOPIC + "出现异常", exception); + errorSum[0]++; + } + } + }); + + kafkaProducer.flush(); + logger.debug("Log sent to National Center successfully!!!!!"); + } + + /** + * 根据kafka生产者配置信息初始化kafka消息生产者,只初始化一次 + */ + private void initKafkaProducer() { + Properties properties = new Properties(); + properties.put("bootstrap.servers", StreamAggregateConfig.RESULTS_BOOTSTRAP_SERVERS); + properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + properties.put("acks", "1"); + properties.put("linger.ms", "2"); + properties.put("request.timeout.ms", 30000); + properties.put("batch.size", 262144); + properties.put("buffer.memory", 33554432); + properties.put("compression.type", StreamAggregateConfig.KAFKA_COMPRESSION_TYPE); + kafkaProducer = new KafkaProducer<>(properties); + } + +} diff --git a/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfig.java b/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfig.java new file mode 100644 index 0000000..a768ce1 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfig.java @@ -0,0 +1,45 @@ +package cn.ac.iie.storm.utils.file; + + +/** + * @author Administrator + */ +public class StreamAggregateConfig { + + public static final String FORMAT_SPLITTER = ","; + /** + * System + */ + public static final Integer SPOUT_PARALLELISM = StreamAggregateConfigurations.getIntProperty(0, "spout.parallelism"); + public static final Integer DATACENTER_BOLT_PARALLELISM = StreamAggregateConfigurations.getIntProperty(0, "datacenter.bolt.parallelism"); + public static final Integer TOPOLOGY_WORKERS = StreamAggregateConfigurations.getIntProperty(0, "topology.workers"); + public static final Integer KAFKA_BOLT_PARALLELISM = StreamAggregateConfigurations.getIntProperty(0, "kafka.bolt.parallelism"); + public static final Integer TOPOLOGY_NUM_ACKS = StreamAggregateConfigurations.getIntProperty(0, "topology.num.acks"); + public static final Integer TOPOLOGY_SPOUT_SLEEP_TIME = StreamAggregateConfigurations.getIntProperty(0, "topology.spout.sleep.time"); + public static final Integer BATCH_INSERT_NUM = StreamAggregateConfigurations.getIntProperty(0, "batch.insert.num"); + public static final Integer DATA_CENTER_ID_NUM = StreamAggregateConfigurations.getIntProperty(0, "data.center.id.num"); + public static final Integer MAX_FAILURE_NUM = StreamAggregateConfigurations.getIntProperty(0, "max.failure.num"); + + + public static final Integer AGG_TIME = StreamAggregateConfigurations.getIntProperty(0, "agg.time"); + public static final Integer UPDATE_APP_ID_TIME = StreamAggregateConfigurations.getIntProperty(0, "update.app.id.time"); + + + /** + * kafka + */ + public static final String BOOTSTRAP_SERVERS = StreamAggregateConfigurations.getStringProperty(0, "bootstrap.servers"); + public static final String RESULTS_BOOTSTRAP_SERVERS = StreamAggregateConfigurations.getStringProperty(0, "results.bootstrap.servers"); + public static final String GROUP_ID = StreamAggregateConfigurations.getStringProperty(0, "group.id"); + public static final String RESULTS_OUTPUT_TOPIC = StreamAggregateConfigurations.getStringProperty(0, "results.output.topic"); + public static final String KAFKA_TOPIC = StreamAggregateConfigurations.getStringProperty(0, "kafka.topic"); + public static final String AUTO_OFFSET_RESET = StreamAggregateConfigurations.getStringProperty(0, "auto.offset.reset"); + public static final String KAFKA_COMPRESSION_TYPE = StreamAggregateConfigurations.getStringProperty(0, "kafka.compression.type"); + + /** + * http + */ + public static final String SCHEMA_HTTP = StreamAggregateConfigurations.getStringProperty(0, "schema.http"); + public static final String APP_ID_HTTP = StreamAggregateConfigurations.getStringProperty(0, "app.id.http"); + +} \ No newline at end of file diff --git a/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfigurations.java b/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfigurations.java new file mode 100644 index 0000000..03f67c0 --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/utils/file/StreamAggregateConfigurations.java @@ -0,0 +1,65 @@ +package cn.ac.iie.storm.utils.file; + +import java.util.Properties; + + +/** + * @author Administrator + */ + +public final class StreamAggregateConfigurations { + + // private static Properties propCommon = new Properties(); + private static Properties propService = new Properties(); + + + public static String getStringProperty(Integer type, String key) { + if (type == 0) { + return propService.getProperty(key); +// } else if (type == 1) { +// return propCommon.getProperty(key); + } else { + return null; + } + + } + + public static Integer getIntProperty(Integer type, String key) { + if (type == 0) { + return Integer.parseInt(propService.getProperty(key)); +// } else if (type == 1) { +// return Integer.parseInt(propCommon.getProperty(key)); + } else { + return null; + } + } + + public static Long getLongProperty(Integer type, String key) { + if (type == 0) { + return Long.parseLong(propService.getProperty(key)); +// } else if (type == 1) { +// return Long.parseLong(propCommon.getProperty(key)); + } else { + return null; + } + } + + public static Boolean getBooleanProperty(Integer type, String key) { + if (type == 0) { + return "true".equals(propService.getProperty(key).toLowerCase().trim()); +// } else if (type == 1) { +// return "true".equals(propCommon.getProperty(key).toLowerCase().trim()); + } else { + return null; + } + } + + static { + try { + propService.load(StreamAggregateConfigurations.class.getClassLoader().getResourceAsStream("service_flow_config.properties")); + } catch (Exception e) { +// propCommon = null; + propService = null; + } + } +} diff --git a/src/main/java/cn/ac/iie/storm/utils/http/HttpClientUtil.java b/src/main/java/cn/ac/iie/storm/utils/http/HttpClientUtil.java new file mode 100644 index 0000000..2e00efd --- /dev/null +++ b/src/main/java/cn/ac/iie/storm/utils/http/HttpClientUtil.java @@ -0,0 +1,55 @@ +package cn.ac.iie.storm.utils.http; + +import org.apache.http.HttpEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; + +/** + * 获取网关schema的工具类 + * + * @author qidaijie + */ +public class HttpClientUtil { + + /** + * 请求网关获取schema + * @param http 网关url + * @return schema + */ + public static String requestByGetMethod(String http) { + CloseableHttpClient httpClient = HttpClients.createDefault(); + StringBuilder entityStringBuilder = null; + try { + HttpGet get = new HttpGet(http); + try (CloseableHttpResponse httpResponse = httpClient.execute(get)) { + HttpEntity entity = httpResponse.getEntity(); + entityStringBuilder = new StringBuilder(); + if (null != entity) { + BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(httpResponse.getEntity().getContent(), "UTF-8"), 8 * 1024); + String line; + while ((line = bufferedReader.readLine()) != null) { + entityStringBuilder.append(line); + } + } + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + try { + if (httpClient != null) { + httpClient.close(); + } + } catch (IOException e) { + e.printStackTrace(); + } + } + return entityStringBuilder.toString(); + } + +} diff --git a/src/main/main.iml b/src/main/main.iml new file mode 100644 index 0000000..908ad4f --- /dev/null +++ b/src/main/main.iml @@ -0,0 +1,11 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/src/test/java/com/wp/AppIdTest.java b/src/test/java/com/wp/AppIdTest.java new file mode 100644 index 0000000..39c9348 --- /dev/null +++ b/src/test/java/com/wp/AppIdTest.java @@ -0,0 +1,54 @@ +package com.wp; + +import cn.ac.iie.storm.utils.file.StreamAggregateConfig; +import cn.ac.iie.storm.utils.http.HttpClientUtil; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; + +/** + * @author qidaijie + * @Package com.wp + * @Description: + * @date 2020/9/2215:09 + */ +public class AppIdTest { + + @Test + public void appTest() { + //http://192.168.44.12:9999/open-api/appDicList + String schema = HttpClientUtil.requestByGetMethod(StreamAggregateConfig.APP_ID_HTTP); + JSONObject jsonObject = JSONObject.parseObject(schema); + String data = jsonObject.getString("data"); + HashMap map = new HashMap<>(16); + JSONArray objects = JSONArray.parseArray(data); + for (Object object : objects) { + JSONArray jsonArray = JSONArray.parseArray(object.toString()); + map.put(jsonArray.getLong(0), jsonArray.getString(1)); +// System.out.println(object); + } + System.out.println(map.toString()); + + System.out.println(map.size()); + } + + @Test + public void changeApp() { + String a = "QQ"; + String[] alignmentPars = "0,/".split(StreamAggregateConfig.FORMAT_SPLITTER); + String data = "100/HTTP"; + int subscript = Integer.parseInt(alignmentPars[0]); + String[] fieldSplit = data.split(alignmentPars[1]); + Long appID = Long.valueOf(fieldSplit[subscript]); + int length = fieldSplit[subscript].length(); + StringBuilder sb = new StringBuilder(data); + + System.out.println(sb.replace(0, length, a)); + + + } +} diff --git a/src/test/java/com/wp/FilterBolt.java b/src/test/java/com/wp/FilterBolt.java new file mode 100644 index 0000000..a3a9ee7 --- /dev/null +++ b/src/test/java/com/wp/FilterBolt.java @@ -0,0 +1,133 @@ +package com.wp; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; + +/** + * @ClassNameFilterBolt + * @Author lixkvip@126.com + * @Date2020/7/1 14:53 + * @Version V1.0 + **/ +public class FilterBolt { + @SuppressWarnings("all") + public static void main(String[] args) { + JSONObject source = new JSONObject(); + + + String schema = "{\n" + + " \"task\": \"Application-Protocol-Distribution\",\n" + + " \"in\": \"CONNECTION-SKETCH-COMPLETED\",\n" + + " \"out\": \"TRAFFIC-PROTOCOL-STAT-LOG\",\n" + + " \"data\": {\n" + + " \"timestamp\": {\n" + + " \"name\": \"stat_time\"\n" + + " },\n" + + " \"dimensions\": [\n" + + " {\n" + + " \"name\": \"protocol_id\",\n" + + " \"fieldName\": \"common_protocol_label\",\n" + + " \"type\": \"String\"\n" + + " },\n" + + " {\n" + + " \"name\": \"device_id\",\n" + + " \"fieldName\": \"common_device_id\",\n" + + " \"type\": \"String\"\n" + + " },\n" + + " {\n" + + " \"name\": \"isp\",\n" + + " \"fieldName\": \"common_isp\",\n" + + " \"type\": \"String\"\n" + + " }\n" + + " ],\n" + + " \"metrics\": [\n" + + " { \"function\" : \"sessions_count\", \"name\" : \"sessions\"},\n" + + " { \"function\" : \"c2s_byte_sum\", \"name\" : \"c2s_byte_len\", \"fieldName\" : \"common_c2s_byte_num\" },\n" + + " { \"function\" : \"s2c_byte_sum\", \"name\" : \"s2c_byte_len\", \"fieldName\" : \"common_s2c_byte_num\" },\n" + + " { \"function\" : \"c2s_pkt_sum\", \"name\" : \"c2s_pkt_num\", \"fieldName\" : \"common_c2s_pkt_num\" },\n" + + " { \"function\" : \"s2c_pkt_sum\", \"name\" : \"s2c_pkt_num\", \"fieldName\" : \"common_s2c_pkt_num\" },\n" + + " { \"function\" : \"sip_disCount\", \"name\" : \"unique_sip_num\", \"fieldName\" : \"common_server_ip\" },\n" + + " { \"function\" : \"cip_disCount\", \"name\" : \"unique_cip_num\", \"fieldName\" : \"common_client_ip\" }\n" + + " ],\n" + + " \"filters\": {\n" + + " \"type\": \"and\",\n" + + " \"fields\": [\n" + + " {\n" + + " \"fieldName\": \"common_device_id\",\n" + + " \"type\": \"not\",\n" + + " \"values\": null\n" + + " },\n" + + " {\n" + + " \"fieldName\": \"common_protocol_label\",\n" + + " \"type\": \"not\",\n" + + " \"values\": null\n" + + " },\n" + + " {\n" + + " \"fieldName\": \"common_isp\",\n" + + " \"type\": \"not\",\n" + + " \"values\": null\n" + + " }\n" + + " ]\n" + + " },\n" + + " \"transforms\":[\n" + + " {\"function\":\"combination\",\"name\":\"protocol_id\",\"fieldName\":\"common_protocol_label\",\"parameters\": \"common_app_label,/\"},\n" + + " {\"function\":\"hierarchy\",\"name\":\"protocol_id\",\"fieldName\":\"common_protocol_label\",\"parameters\": \"/\"}\n" + + " ],\n" + + " \"action\":[\n" + + " {\"label\": \"Default\", \"metrics\": \"sessions,c2s_byte_len,s2c_byte_len,c2s_pkt_num,s2c_pkt_num\"},\n" + + " {\"label\": \"HTTP\", \"metrics\": \"sessions,c2s_byte_len,s2c_byte_len,c2s_pkt_num,s2c_pkt_num,unique_sip_num,unique_cip_num\"}\n" + + " ],\n" + + " \"granularity\":{\n" + + " \"type\": \"period\",\n" + + " \"period\": \"5M\"\n" + + " }\n" + + " }\n" + + "}"; + + + source.put("common_protocol_label", "HTTP"); + source.put("common_isp", "Unicom"); + source.put("common_device_id", "1"); + String data = JSONObject.parseObject(schema).getString("data"); + + String filters = JSONObject.parseObject(data).getString("filters"); + + boolean flag = true; + String type = JSONObject.parseObject(filters).getString("type"); + JSONArray fieldsArr = JSONObject.parseArray(JSONObject.parseObject(filters).getString("fields")); + if ("and".equals(type)) { + for (int i = 0; i < fieldsArr.size(); i++) { + + JSONObject field = JSONObject.parseObject(fieldsArr.get(i).toString()); + String name = field.getString("fieldName"); + String fieldType = field.getString("type"); + Object values = field.get("values"); + + Object nameValue = source.get(name); + + + if ("not".equals(fieldType)) { + + if (nameValue == values) { + //满足过滤条件 + flag = false; + } + + } else if ("in".equals(fieldType)) { + if (!values.toString().contains(nameValue.toString())) { + //满足过滤条件 + flag = false; + } + } + } + + if (flag){ + System.out.println("输出到下个Bolt"); + }else { + + System.out.println("此条消息被过滤掉"); + } + + } + } +} diff --git a/src/test/java/com/wp/SchemaTest.java b/src/test/java/com/wp/SchemaTest.java new file mode 100644 index 0000000..f275592 --- /dev/null +++ b/src/test/java/com/wp/SchemaTest.java @@ -0,0 +1,43 @@ +package com.wp; + +import com.alibaba.fastjson.JSONObject; +import com.zdjizhi.utils.StringUtil; + +/** + * @ClassNameSchemaTest + * @Author lixkvip@126.com + * @Date2020/6/28 10:11 + * @Version V1.0 + **/ +public class SchemaTest { + + static String str = ""; + + public static void main(String[] args) { + + + String str1 = null; + String str2 = " "; + + + System.out.println(StringUtil.isNotBlank(str1)); + System.out.println(StringUtil.isNotEmpty(str1)); + + System.out.println(StringUtil.isNotBlank(str2)); + System.out.println(StringUtil.isNotEmpty(str2)); + + } + + public static void reAdd(int m, String[] split, String str) { + + //递归拼接字符串 + if (0 == m) { + } else { + //递归的核心代码 + str = str + split[m - 1] + "/"; + reAdd(m - 1, split, str); + + } + + } +} diff --git a/src/test/test.iml b/src/test/test.iml new file mode 100644 index 0000000..5ebc6f4 --- /dev/null +++ b/src/test/test.iml @@ -0,0 +1,12 @@ + + + + + + + + + + + + \ No newline at end of file