diff options
Diffstat (limited to 'src/main/java')
4 files changed, 7 insertions, 10 deletions
diff --git a/src/main/java/com/zdjizhi/utils/functions/ResultFlatMapFunction.java b/src/main/java/com/zdjizhi/utils/functions/ResultFlatMapFunction.java index 8ff8839..57e8a2c 100644 --- a/src/main/java/com/zdjizhi/utils/functions/ResultFlatMapFunction.java +++ b/src/main/java/com/zdjizhi/utils/functions/ResultFlatMapFunction.java @@ -18,6 +18,7 @@ import java.util.Map; */ public class ResultFlatMapFunction implements FlatMapFunction<String, String> { private static String[] jobList = JsonParseUtil.getHierarchy(); + private static final String APP_NAME = "app_name"; @Override @SuppressWarnings("unchecked") @@ -26,6 +27,8 @@ public class ResultFlatMapFunction implements FlatMapFunction<String, String> { String name = jobList[0]; Map<String, Object> jsonObject = (Map<String, Object>) JsonMapper.fromJsonString(value, Map.class); String protocol = JsonParseUtil.getString(jsonObject, name); + String appName = JsonParseUtil.getString(jsonObject, APP_NAME); + jsonObject.remove(APP_NAME); if (StringUtil.isNotBlank(protocol)) { String[] protocolIds = protocol.split(StreamAggregateConfig.PROTOCOL_SPLITTER); for (String proto : protocolIds) { @@ -35,6 +38,9 @@ public class ResultFlatMapFunction implements FlatMapFunction<String, String> { out.collect(JsonMapper.toJsonString(jsonObject)); } else { stringBuffer.append(jobList[1]).append(proto); + if (proto.equals(appName)) { + jsonObject.put(APP_NAME, appName); + } jsonObject.put(name, stringBuffer.toString()); out.collect(JsonMapper.toJsonString(jsonObject)); } diff --git a/src/main/java/com/zdjizhi/utils/functions/parse/ParseMapFunction.java b/src/main/java/com/zdjizhi/utils/functions/parse/ParseMapFunction.java index d1fc36d..4d69c82 100644 --- a/src/main/java/com/zdjizhi/utils/functions/parse/ParseMapFunction.java +++ b/src/main/java/com/zdjizhi/utils/functions/parse/ParseMapFunction.java @@ -66,13 +66,6 @@ public class ParseMapFunction implements MapFunction<String, Tuple3<String, Stri case "hierarchy": String key = JsonParseUtil.getString(dimensionsObj, resultKeyName) + "@" + ThreadLocalRandom.current().nextInt(StreamAggregateConfig.RANDOM_RANGE_NUM); return new Tuple3<>(key, JsonMapper.toJsonString(dimensionsObj), ParseFunctions.getMetricsLog(originalLog)); -// case "dismantling": -// if (StringUtil.isNotBlank(parameters)) { -// if (logsValue != null) { -// JsonParseUtil.setValue(message, logsKeyName, ParseFunctions.dismantlingUtils(parameters, logsValue)); -// } -// } -// break; default: break; } diff --git a/src/main/java/com/zdjizhi/utils/functions/statistics/FirstCountWindowFunction.java b/src/main/java/com/zdjizhi/utils/functions/statistics/FirstCountWindowFunction.java index 92f6697..93844be 100644 --- a/src/main/java/com/zdjizhi/utils/functions/statistics/FirstCountWindowFunction.java +++ b/src/main/java/com/zdjizhi/utils/functions/statistics/FirstCountWindowFunction.java @@ -59,7 +59,6 @@ public class FirstCountWindowFunction extends ProcessWindowFunction<Tuple3<Strin if (!cacheMap.isEmpty()) { for (String dimensions : cacheMap.keySet()) { Map<String, Object> resultMap = cacheMap.get(dimensions); - System.out.println("resultMap"+resultMap.toString()); output.collect(new Tuple2<>(dimensions, JsonMapper.toJsonString(resultMap))); } } diff --git a/src/main/java/com/zdjizhi/utils/general/ParseFunctions.java b/src/main/java/com/zdjizhi/utils/general/ParseFunctions.java index e634a02..0c8d7df 100644 --- a/src/main/java/com/zdjizhi/utils/general/ParseFunctions.java +++ b/src/main/java/com/zdjizhi/utils/general/ParseFunctions.java @@ -118,13 +118,12 @@ public class ParseFunctions { /** * 根据表达式解析json * <p> - * // * @param message json + * //* @param message json * * @param expr 解析表达式 * @return 解析结果 */ public static void flattenSpec(Map<String, Object> dimensions, String expr, String resultKeyName, String logsKeyValue) { - try { if (StringUtil.isNotBlank(expr)) { ArrayList<Object> read = JsonPath.parse(logsKeyValue).read(expr); |