Skip to content

F
flink-monitor
Commit 80f547e9 authored by 赵建伟's avatar 赵建伟
Browse files
Options

update codes

parent fba1845d
Hide whitespace changes
Inline Side-by-side
Showing with 81 additions and 24 deletions
src/main/java/com/gmei/data/monitor/operator/PortraitMonitorShdOperator.java
View file @ 80f547e9
...@@ -60,25 +60,7 @@ public class PortraitMonitorShdOperator implements BaseOperator{ ...@@ -60,25 +60,7 @@ public class PortraitMonitorShdOperator implements BaseOperator{
public boolean filter(String value) throws Exception { public boolean filter(String value) throws Exception {
return JSON.isValid(value); return JSON.isValid(value);
} }
}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() { }).map(new MapFunction<String, JSONObject>() {
@Override
public long extractAscendingTimestamp(String element) {
JSONObject jsonObject = JSON.parseObject(element);
long logTime = 0;
String maidianEventTime = jsonObject.getString("create_at");
if (StringUtils.isNotBlank(maidianEventTime)) {
logTime = Long.valueOf(maidianEventTime) * 1000;
}
JSONObject appObject = jsonObject.getJSONObject("APP");
if (null != appObject) {
Long time = appObject.getLong("time");
if(null != time && time != 0){
logTime = time;
}
}
return logTime;
}
}).map(new MapFunction<String, JSONObject>() {
@Override @Override
public JSONObject map(String value) throws Exception { public JSONObject map(String value) throws Exception {
return JSONObject.parseObject(value); return JSONObject.parseObject(value);
...@@ -116,11 +98,11 @@ public class PortraitMonitorShdOperator implements BaseOperator{ ...@@ -116,11 +98,11 @@ public class PortraitMonitorShdOperator implements BaseOperator{
} }
String appAction = appObject.getString("action"); String appAction = appObject.getString("action");
if (null != appAction) { if (null != appAction) {
String[] edits = {"create", "update", "answer"}; // String[] edits = {"create", "update", "answer"};
if (Arrays.asList(edits).contains(appAction)) { // if (Arrays.asList(edits).contains(appAction)) {
jsonObject.put("statistics_action", appAction); // jsonObject.put("statistics_action", appAction);
return true; // return true;
} // }
String[] interact = {"like", "comment", "collect"}; String[] interact = {"like", "comment", "collect"};
if (Arrays.asList(interact).contains(appAction)) { if (Arrays.asList(interact).contains(appAction)) {
jsonObject.put("statistics_action", appAction); jsonObject.put("statistics_action", appAction);
...@@ -146,6 +128,81 @@ public class PortraitMonitorShdOperator implements BaseOperator{ ...@@ -146,6 +128,81 @@ public class PortraitMonitorShdOperator implements BaseOperator{
} }
return false; return false;
} }
}).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<JSONObject>() {
@Override
public long extractAscendingTimestamp(JSONObject jsonObject) {
long logTime = 0;
JSONObject sysObject = jsonObject.getJSONObject("SYS");
if (null != sysObject) {
String action = sysObject.getString("action");
if (null != action) {
if ("/api/private_conversation/".equals(action)
//|| "/api/initiate/interest_record".equals(action)
//|| "/api/one_image/share/v3".equals(action)
//|| "/gm_ai/face_app/test_skin".equals(action)
) {
JSONObject appObject = jsonObject.getJSONObject("APP");
if (null != appObject) {
Long time = appObject.getLong("time");
if(null != time && time != 0){
logTime = time;
}
}
}
}
}
JSONObject appObject = jsonObject.getJSONObject("APP");
if (null != appObject) {
String eventType = appObject.getString("event_type");
if (null != eventType) {
if ("validate_order".equals(eventType)
|| "paid_success".equals(eventType)
|| "add_shopcart".equals(eventType)) {
String backendEventTime = jsonObject.getString("TIME");
if (StringUtils.isNotBlank(backendEventTime)) {
try {
logTime = dateTimeFormat.parseMillis(backendEventTime);
} catch (IllegalArgumentException e) {
try {
logTime = dateTimeNoMillisFormat.parseMillis(backendEventTime);
} catch (IllegalArgumentException e2) {
e2.printStackTrace();
}
}
}
}
}
String appAction = appObject.getString("action");
if (null != appAction) {
// String[] edits = {"create", "update", "answer"};
// if (Arrays.asList(edits).contains(appAction)) {
// jsonObject.put("statistics_action", appAction);
// return true;
// }
String[] interact = {"like", "comment", "collect"};
if (Arrays.asList(interact).contains(appAction)) {
String timestamp = appObject.getString("timestamp");
logTime = Math.round(Double.valueOf(timestamp) * 1000);
}
}
}
String type = jsonObject.getString("type");
String device = jsonObject.getString("device");
if (null != type && null != device) {
if ("do_search".equals(type)
|| "goto_welfare_detail".equals(type)
|| "on_click_card".equals(type)
//|| "home_click_section".equals(type)
) {
String maidianEventTime = jsonObject.getString("create_at");
if (StringUtils.isNotBlank(maidianEventTime)) {
logTime = Long.valueOf(maidianEventTime) * 1000;
}
}
}
return logTime;
}
}); });
//filter01.print(); //filter01.print();
SingleOutputStreamOperator map02 = filter01.map(new MapFunction<JSONObject, Tuple2<String, JSONObject>>() { SingleOutputStreamOperator map02 = filter01.map(new MapFunction<JSONObject, Tuple2<String, JSONObject>>() {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment