Commit b28f2373 authored by sheteng's avatar sheteng

调整数据

parent 7a7d70c3
...@@ -81,7 +81,7 @@ public class AppsflyerAnalyze { ...@@ -81,7 +81,7 @@ public class AppsflyerAnalyze {
properties.setProperty("auto.offset.reset", "earliest"); properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties); FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
// TODO: 2021/8/10 设置消费开始的时间戳1628524944000 // TODO: 2021/8/10 设置消费开始的时间戳1628524944000
// dataConsumer.setStartFromTimestamp(1629821205000L); // dataConsumer.setStartFromTimestamp(1630598733000L);
Properties props2 = new Properties(); Properties props2 = new Properties();
props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers()); props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers());
...@@ -101,20 +101,20 @@ public class AppsflyerAnalyze { ...@@ -101,20 +101,20 @@ public class AppsflyerAnalyze {
.name("activition-event"); .name("activition-event");
DataStream<ActivationMetrics> createRoleMetricStream = adEventDataStream // DataStream<ActivationMetrics> createRoleMetricStream = adEventDataStream
.filter(EventTypeEnum::createRoleEvent) // .filter(EventTypeEnum::createRoleEvent)
.flatMap(new CreateRoleProcess(appConfig)) // .flatMap(new CreateRoleProcess(appConfig))
.uid("create-role-event") // .uid("create-role-event")
.name("create-role-event"); // .name("create-role-event");
//注册\留存 //注册\留存 由于可能出现创角和角色登录是一瞬间的,所以他两合并计算
DataStream<ActivationMetrics> loginRoleMetricStream = adEventDataStream DataStream<ActivationMetrics> loginRoleMetricStream = adEventDataStream
.filter(EventTypeEnum::roleLoginEvent) .filter(EventTypeEnum::roleLoginEvent)
.flatMap(new RoleLoginProcess(appConfig)) .flatMap(new RoleLoginProcess(appConfig))
.uid("role-login-event") .uid("role-login-event")
.name("role-login-event"); .name("role-login-event");
activitionMetricStream.union(loginRoleMetricStream, createRoleMetricStream) activitionMetricStream.union(loginRoleMetricStream)
.flatMap(new ActivitionMetricToStringProcess(appConfig)) .flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("sink-metric-event") .uid("sink-metric-event")
...@@ -136,7 +136,7 @@ public class AppsflyerAnalyze { ...@@ -136,7 +136,7 @@ public class AppsflyerAnalyze {
.uid("activation-event") .uid("activation-event")
.name("activation-event"); .name("activation-event");
//事件统计激活
adEventDataStream adEventDataStream
.filter(EventTypeEnum::notHeartBeatEvent) .filter(EventTypeEnum::notHeartBeatEvent)
.flatMap(new EventMectrisProcessor(appConfig)) .flatMap(new EventMectrisProcessor(appConfig))
......
...@@ -84,7 +84,7 @@ public class AppsflyerMetricStore { ...@@ -84,7 +84,7 @@ public class AppsflyerMetricStore {
} }
}).uid("parse-event").name("parse-event"); }).uid("parse-event").name("parse-event");
activationMetricsDataStream.print(); // activationMetricsDataStream.print();
//角色注册激活预聚合 //角色注册激活预聚合
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor()) .flatMap(new MetricJsonToObProcessor())
...@@ -166,7 +166,7 @@ public class AppsflyerMetricStore { ...@@ -166,7 +166,7 @@ public class AppsflyerMetricStore {
.name("server-remindsink-event"); .name("server-remindsink-event");
/** /**
* 时间写入 * 事件写入
*/ */
activationMetricsDataStream.filter(SinkMetricEvent::isEvent) activationMetricsDataStream.filter(SinkMetricEvent::isEvent)
.addSink(new EventTimesSink(appConfig)) .addSink(new EventTimesSink(appConfig))
......
...@@ -45,4 +45,8 @@ public interface GameAccountDao { ...@@ -45,4 +45,8 @@ public interface GameAccountDao {
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " + @SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'purchase',:event_time ) ") " values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'purchase',:event_time ) ")
void insertPauchase(@BindBean AppsFlyerEvent value); void insertPauchase(@BindBean AppsFlyerEvent value);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'createRole',:event_time ) ")
void insertCreateRole(@BindBean AppsFlyerEvent value);
} }
...@@ -46,6 +46,7 @@ public enum EventTypeEnum { ...@@ -46,6 +46,7 @@ public enum EventTypeEnum {
public static boolean roleLoginEvent(AppsFlyerEvent event) { public static boolean roleLoginEvent(AppsFlyerEvent event) {
return (ROLE_LOGIN.getName().equals(event.getEvent()) return (ROLE_LOGIN.getName().equals(event.getEvent())
|| FAST_LOGIN.getName().equals(event.getEvent()) || FAST_LOGIN.getName().equals(event.getEvent())
|| CREATE_ROLE.getName().equals(event.getEvent())
|| GAME_LOGIN.getName().equals(event.getEvent()) || GAME_LOGIN.getName().equals(event.getEvent())
|| SDK_LOGIN.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE); || SDK_LOGIN.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
} }
......
...@@ -47,30 +47,34 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -47,30 +47,34 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time))); activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time))); activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
if (!account.isPresent()) { // 设备未激活 if (!account.isPresent()) { // 设备未激活
// activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
// activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
gameAccountDao.insertActivition(value, event_time); gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L); activationMetrics.setData_activation(1L);
activition = value.getEvent_time(); activition = value.getEvent_time();
// return;
} else { } else {
GameAccount gameAccount = account.get(); GameAccount gameAccount = account.get();
String activationTime = gameAccount.getEvent_time(); String activationTime = gameAccount.getEvent_time();
// activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
// activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
activition = DateUtils.formatAsDate(activationTime); activition = DateUtils.formatAsDate(activationTime);
} }
boolean isRole = false;
Optional<GameAccount> accountCreateRole = gameAccountDao.findByPrimary(value, Constants.TYPE_CREATEROLE);
if (!accountCreateRole.isPresent()) {
activationMetrics.setData_create_role(1L);
gameAccountDao.insertCreateRole(value);
isRole = true;
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER); Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { if (!accountRegister.isPresent()) {
activationMetrics.setData_register(1L); activationMetrics.setData_register(1L);
activationMetrics.setData_create_role(1L);
activationMetrics.setData_device_register(1L); activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_register_create(1L);//当日注册并创角 activationMetrics.setData_day_register_create(1L);//当日注册并创角
if (DateUtils.daysBetween(activition, value.getEvent_time()) == 0) { //当日激活并且注册 if (DateUtils.daysBetween(activition, value.getEvent_time()) == 0) { //当日激活并且注册
activationMetrics.setData_day_act_register(1L); activationMetrics.setData_day_act_register(1L);
} }
} else { } else {
if (DateUtils.daysBetween(DateUtils.formatAsDate(accountRegister.get().getEvent_time()), value.getEvent_time()) == 0) { //当日注册创角 if (DateUtils.daysBetween(DateUtils.formatAsDate(accountRegister.get().getEvent_time()), value.getEvent_time()) == 0 && isRole) { //当日注册创角
activationMetrics.setData_day_register_create(1L); activationMetrics.setData_day_register_create(1L);
} }
} }
......
...@@ -8,21 +8,16 @@ import com.fshark.overseas.advert.modle.EventTypeEnum; ...@@ -8,21 +8,16 @@ import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics; import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*; import com.fshark.overseas.advert.util.*;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector; import org.apache.flink.util.Collector;
import java.time.LocalDate;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale; import java.util.Locale;
import java.util.Optional; import java.util.Optional;
public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> { public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
private AppConfig appConfig; private final AppConfig appConfig;
private JdbiContext jdbiContext; private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao; private GameAccountDao gameAccountDao;
...@@ -53,11 +48,37 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -53,11 +48,37 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
if (!account.isPresent()) { // 设备未激活 if (!account.isPresent()) { // 设备未激活
gameAccountDao.insertActivition(value, event_time); gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L); activationMetrics.setData_activation(1L);
// return;
} else { } else {
GameAccount gameAccount = account.get(); GameAccount gameAccount = account.get();
activationTime = gameAccount.getEvent_time(); activationTime = gameAccount.getEvent_time();
} }
if (EventTypeEnum.createRoleEvent(value)) {
boolean isRole = false;
Optional<GameAccount> accountCreateRole = gameAccountDao.findByPrimary(value, Constants.TYPE_CREATEROLE);
if (!accountCreateRole.isPresent()) {
activationMetrics.setData_create_role(1L);
gameAccountDao.insertCreateRole(value);
isRole = true;
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) {
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_register_create(1L);//当日注册并创角
if (DateUtils.daysBetween(DateUtils.formatAsDate(activationTime), value.getEvent_time()) == 0) { //当日激活并且注册
activationMetrics.setData_day_act_register(1L);
}
} else {
if (DateUtils.daysBetween(DateUtils.formatAsDate(accountRegister.get().getEvent_time()), value.getEvent_time()) == 0 && isRole) { //当日注册创角
activationMetrics.setData_day_register_create(1L);
}
}
gameAccountDao.insertRegister(value, event_time);
} else {
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER); Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { //设备未注册 if (!accountRegister.isPresent()) { //设备未注册
gameAccountDao.insertRegister(value, event_time); gameAccountDao.insertRegister(value, event_time);
...@@ -91,6 +112,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -91,6 +112,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
} }
} }
gameAccountDao.insertRoleLogin(value); gameAccountDao.insertRoleLogin(value);
}
out.collect(activationMetrics); out.collect(activationMetrics);
out.collect(retentMetrics); out.collect(retentMetrics);
} }
......
...@@ -12,6 +12,7 @@ import com.google.gson.Gson; ...@@ -12,6 +12,7 @@ import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.util.Optional; import java.util.Optional;
public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> { public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
...@@ -28,7 +29,7 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> { ...@@ -28,7 +29,7 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){ if (appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else { } else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
...@@ -39,11 +40,16 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> { ...@@ -39,11 +40,16 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
@Override @Override
public void invoke(SinkMetricEvent value, Context context) throws Exception { public void invoke(SinkMetricEvent value, Context context) throws Exception {
super.invoke(value, context); super.invoke(value, context);
if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())){ if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())) {
OrderData eventData = value.getOrderData(); OrderData eventData = value.getOrderData();
orderDao.insertOrder(eventData); orderDao.insertOrder(eventData);
if (!StringUtils.isBlank(eventData.getReg_server_time())) {
eventData.setServer_reg_date(DateUtils.formatAsYMD(eventData.getReg_server_time())); eventData.setServer_reg_date(DateUtils.formatAsYMD(eventData.getReg_server_time()));
eventData.setServer_reg_hour(DateUtils.formatAsHH(eventData.getReg_server_time())); eventData.setServer_reg_hour(DateUtils.formatAsHH(eventData.getReg_server_time()));
} else {
eventData.setServer_reg_date("-1");
eventData.setServer_reg_hour("-1");
}
orderDao.insertServerOrder(eventData); orderDao.insertServerOrder(eventData);
} }
} }
......
...@@ -71,6 +71,7 @@ public class Constants { ...@@ -71,6 +71,7 @@ public class Constants {
public static final String TYPE_ACTIVIYION = "activation"; public static final String TYPE_ACTIVIYION = "activation";
public static final String TYPE_REGISTER = "register"; public static final String TYPE_REGISTER = "register";
public static final String TYPE_CREATEROLE = "createRole";
public static final String TYPE_ROLELOGIN = "rolelogin"; public static final String TYPE_ROLELOGIN = "rolelogin";
public static final String TYPE_PAUCHASE = "purchase"; public static final String TYPE_PAUCHASE = "purchase";
......
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