Commit b4277acc authored by sheteng's avatar sheteng

增加sdk激活注册

parent 949152e9
<component name="ProjectCodeStyleConfiguration">
<code_scheme name="Project" version="173">
<JetCodeStyleSettings>
<option name="CODE_STYLE_DEFAULTS" value="KOTLIN_OFFICIAL" />
</JetCodeStyleSettings>
<codeStyleSettings language="kotlin">
<option name="CALL_PARAMETERS_WRAP" value="5" />
<option name="CALL_PARAMETERS_LPAREN_ON_NEXT_LINE" value="true" />
<option name="CALL_PARAMETERS_RPAREN_ON_NEXT_LINE" value="true" />
<option name="METHOD_PARAMETERS_WRAP" value="5" />
<option name="METHOD_PARAMETERS_LPAREN_ON_NEXT_LINE" value="true" />
<option name="METHOD_PARAMETERS_RPAREN_ON_NEXT_LINE" value="true" />
<option name="EXTENDS_LIST_WRAP" value="1" />
<option name="METHOD_CALL_CHAIN_WRAP" value="1" />
<option name="ASSIGNMENT_WRAP" value="1" />
</codeStyleSettings>
</code_scheme>
</component>
\ No newline at end of file
...@@ -12,6 +12,7 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies; ...@@ -12,6 +12,7 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.time.Time;
...@@ -89,7 +90,7 @@ public class AppsflyerAnalyze { ...@@ -89,7 +90,7 @@ public class AppsflyerAnalyze {
.uid("parse-event") .uid("parse-event")
.name("parse-event") .name("parse-event")
.shuffle(); .shuffle();
// adEventDataStream.print();
//广告指标(激活) //广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream
.filter(EventTypeEnum::installEvent) .filter(EventTypeEnum::installEvent)
...@@ -97,13 +98,6 @@ public class AppsflyerAnalyze { ...@@ -97,13 +98,6 @@ public class AppsflyerAnalyze {
.uid("activition-event") .uid("activition-event")
.name("activition-event"); .name("activition-event");
// DataStream<ActivationMetrics> createRoleMetricStream = adEventDataStream
// .filter(EventTypeEnum::createRoleEvent)
// .flatMap(new CreateRoleProcess(appConfig))
// .uid("create-role-event")
// .name("create-role-event");
//注册\留存 由于可能出现创角和角色登录是一瞬间的,所以他两合并计算 //注册\留存 由于可能出现创角和角色登录是一瞬间的,所以他两合并计算
DataStream<ActivationMetrics> loginRoleMetricStream = adEventDataStream DataStream<ActivationMetrics> loginRoleMetricStream = adEventDataStream
.filter(EventTypeEnum::roleLoginEvent) .filter(EventTypeEnum::roleLoginEvent)
...@@ -111,7 +105,15 @@ public class AppsflyerAnalyze { ...@@ -111,7 +105,15 @@ public class AppsflyerAnalyze {
.uid("role-login-event") .uid("role-login-event")
.name("role-login-event"); .name("role-login-event");
activitionMetricStream.union(loginRoleMetricStream) //sdk激活注册 当天
DataStream<ActivationMetrics> sdkMetrics = adEventDataStream
.filter(EventTypeEnum::sdkEvent)
.flatMap(new SdkProcess(appConfig))
.uid("sdk-event")
.name("sdk-event");
activitionMetricStream.union(loginRoleMetricStream).union(sdkMetrics)
.flatMap(new ActivitionMetricToStringProcess(appConfig)) .flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("sink-metric-event") .uid("sink-metric-event")
...@@ -136,6 +138,7 @@ public class AppsflyerAnalyze { ...@@ -136,6 +138,7 @@ public class AppsflyerAnalyze {
//事件统计激活 //事件统计激活
adEventDataStream adEventDataStream
.filter(EventTypeEnum::notHeartBeatEvent) .filter(EventTypeEnum::notHeartBeatEvent)
.keyBy(AppsFlyerEvent::getUid)
.flatMap(new EventMectrisProcessor(appConfig)) .flatMap(new EventMectrisProcessor(appConfig))
.keyBy(EventMetrics::getKeys) .keyBy(EventMetrics::getKeys)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1))) .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
......
...@@ -83,14 +83,13 @@ public class AppsflyerMetricStore { ...@@ -83,14 +83,13 @@ public class AppsflyerMetricStore {
}) })
.uid("parse-event") .uid("parse-event")
.name("parse-event"); .name("parse-event");
// activationMetricsDataStream.print(); // activationMetricsDataStream.print();
//角色注册激活预聚合 //角色注册激活预聚合
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor()) .flatMap(new MetricJsonToObProcessor())
.keyBy(ActivationMetrics::getKey) // .keyBy(ActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1))) // .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.aggregate(new ActivationAggregateFunction(), new ActivationResultWindows()) // .aggregate(new ActivationAggregateFunction(), new ActivationResultWindows())
.uid("activation-agg-event") .uid("activation-agg-event")
.name("activation-agg-event"); .name("activation-agg-event");
......
...@@ -50,4 +50,12 @@ public interface GameAccountDao { ...@@ -50,4 +50,12 @@ 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,'createRole',: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); void insertCreateRole(@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,'sdkActivation',:reg_time) ")
void insertSdkActivition(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_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,'sdkRegister',:reg_time) ")
void insertSdkRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time);
} }
...@@ -19,8 +19,8 @@ public interface MetricsDao { ...@@ -19,8 +19,8 @@ public interface MetricsDao {
* @param event * @param event
*/ */
@SqlUpdate("insert into appsflyer_metrics" + @SqlUpdate("insert into appsflyer_metrics" +
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_activation`,`data_register`,`data_device_register`,`data_create_role`,`data_day_act_register`,`data_day_register_create`,`data_active_day_dau`,`data_type`)" + " (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_activation`,`data_sdk_activation`,`data_sdk_register`,`data_register`,`data_device_register`,`data_create_role`,`data_day_act_register`,`data_day_register_create`,`data_active_day_dau`,`data_type`)" +
" values (:data_date,:data_hour,:time_zone,:region,:platform,:game_id,:media_source,:campaign_id,:ad_set_id,:ad_id,:channel,:data_activation,:data_register,:data_device_register,:data_create_role,:data_day_act_register,:data_day_register_create,:data_active_day_dau,:data_type) ") " values (:data_date,:data_hour,:time_zone,:region,:platform,:game_id,:media_source,:campaign_id,:ad_set_id,:ad_id,:channel,:data_activation,:data_sdk_activation,:data_sdk_register,:data_register,:data_device_register,:data_create_role,:data_day_act_register,:data_day_register_create,:data_active_day_dau,:data_type) ")
void insertActivationMetrics(@BindBean ActivationMetrics event); void insertActivationMetrics(@BindBean ActivationMetrics event);
......
...@@ -15,9 +15,10 @@ public enum EventTypeEnum { ...@@ -15,9 +15,10 @@ public enum EventTypeEnum {
/* /*
第一次创角作为注册 第一次创角作为注册
*/ */
REGISTER("createrole"), // REGISTER("createrole"),
CREATE_ROLE("createrole"), CREATE_ROLE("createrole"),
SDK_LOGIN("sdk login"), SDK_ACTIVATION("activation"),
SDK_REGISTER("register"),
FAST_LOGIN("fast login"), FAST_LOGIN("fast login"),
GAME_LOGIN("gamelogin"), GAME_LOGIN("gamelogin"),
ROLE_LOGIN("rolelogin"), ROLE_LOGIN("rolelogin"),
...@@ -42,6 +43,11 @@ public enum EventTypeEnum { ...@@ -42,6 +43,11 @@ public enum EventTypeEnum {
return Arrays.stream(values()).map(EventTypeEnum::getName).distinct().collect(Collectors.toList()); return Arrays.stream(values()).map(EventTypeEnum::getName).distinct().collect(Collectors.toList());
} }
public static boolean sdkEvent(AppsFlyerEvent event){
return (SDK_ACTIVATION.getName().equals(event.getEvent())
|| SDK_REGISTER.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
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())
|| CREATE_ROLE.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE); || CREATE_ROLE.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
......
...@@ -15,10 +15,19 @@ public class ActivationMetrics extends BaseMetrics implements Serializable { ...@@ -15,10 +15,19 @@ public class ActivationMetrics extends BaseMetrics implements Serializable {
* 激活数 按唯一设备去重 * 激活数 按唯一设备去重
*/ */
private Long data_activation =0L; private Long data_activation =0L;
/**
* sdk激活数 按唯一设备去重
*/
private Long data_sdk_activation =0L;
/** /**
* 注册数 按唯一用户去重 * 注册数 按唯一用户去重
*/ */
private Long data_register = 0L; private Long data_register = 0L;
/**
* sdk注册数 按唯一用户去重
*/
private Long data_sdk_register = 0L;
/** /**
* 设备注册数 * 设备注册数
*/ */
......
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import org.apache.flink.api.common.functions.AggregateFunction;
public class ActivationAggregateFunction implements AggregateFunction<ActivationMetrics,ActivationMetrics,ActivationMetrics> {
@Override
public ActivationMetrics createAccumulator() {
return new ActivationMetrics();
}
@Override
public ActivationMetrics add(ActivationMetrics value, ActivationMetrics accumulator) {
value.setData_activation(value.getData_activation() + accumulator.getData_activation());
value.setData_register(accumulator.getData_register() + value.getData_register());
value.setData_device_register(accumulator.getData_device_register() + value.getData_device_register());
value.setData_create_role(accumulator.getData_create_role() + value.getData_create_role());
value.setData_day_act_register(accumulator.getData_day_act_register() + value.getData_day_act_register());
value.setData_day_register_create(accumulator.getData_day_register_create() + value.getData_day_register_create());
value.setData_active_day_dau(accumulator.getData_active_day_dau() + value.getData_active_day_dau());
return value;
}
@Override
public ActivationMetrics getResult(ActivationMetrics accumulator) {
return accumulator;
}
@Override
public ActivationMetrics merge(ActivationMetrics value, ActivationMetrics accumulator) {
value.setData_activation(accumulator.getData_activation() + value.getData_activation());
value.setData_register(accumulator.getData_register() + value.getData_register());
value.setData_device_register(accumulator.getData_device_register() + value.getData_device_register());
value.setData_create_role(accumulator.getData_create_role() + value.getData_create_role());
value.setData_day_act_register(accumulator.getData_day_act_register() + value.getData_day_act_register());
value.setData_day_register_create(accumulator.getData_day_register_create() + value.getData_day_register_create());
value.setData_active_day_dau(accumulator.getData_active_day_dau() + value.getData_active_day_dau());
return value;
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.google.gson.Gson;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
public class ActivationResultWindows implements WindowFunction<ActivationMetrics, ActivationMetrics, String, TimeWindow> {
@Override
public void apply(String s, TimeWindow window, Iterable<ActivationMetrics> input, Collector<ActivationMetrics> out) throws Exception {
BaseMetrics baseMetrics = BaseMetrics.setKey(s);
ActivationMetrics activationMetrics = copyMetrics(baseMetrics);
ActivationMetrics accumulator = input.iterator().next();
activationMetrics.setData_activation(accumulator.getData_activation());
activationMetrics.setData_register(accumulator.getData_register());
activationMetrics.setData_device_register(accumulator.getData_device_register());
activationMetrics.setData_create_role(accumulator.getData_create_role());
activationMetrics.setData_day_act_register(accumulator.getData_day_act_register());
activationMetrics.setData_day_register_create(accumulator.getData_day_register_create());
activationMetrics.setData_active_day_dau(accumulator.getData_active_day_dau());
out.collect(activationMetrics);
}
private ActivationMetrics copyMetrics(BaseMetrics metrics) {
Gson gson = new Gson();
return gson.fromJson(gson.toJson(metrics), ActivationMetrics.class);
}
}
...@@ -10,9 +10,15 @@ import com.fshark.overseas.advert.modle.metrics.EventMetrics; ...@@ -10,9 +10,15 @@ import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.util.*; import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson; import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
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.LocalDateTime;
import java.util.Optional; import java.util.Optional;
public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, EventMetrics> { public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, EventMetrics> {
...@@ -24,6 +30,8 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E ...@@ -24,6 +30,8 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E
private GameAccountDao gameAccountDao; private GameAccountDao gameAccountDao;
private transient ValueState<Long> userTime;
public EventMectrisProcessor(AppConfig appConfig) { public EventMectrisProcessor(AppConfig appConfig) {
this.appConfig = appConfig; this.appConfig = appConfig;
...@@ -34,6 +42,12 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E ...@@ -34,6 +42,12 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>(
"user_time", // the state name
TypeInformation.of(new TypeHint<Long>() {}), // type information
null); // default value of the state, if nothing was set
userTime = getRuntimeContext().getState(descriptor);
} }
...@@ -56,7 +70,6 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E ...@@ -56,7 +70,6 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E
return; return;
} }
} }
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
EventMetrics metrics = new EventMetrics(); EventMetrics metrics = new EventMetrics();
metrics.setTimeZone(value.getTime_zone()); metrics.setTimeZone(value.getTime_zone());
metrics.setRegion(value.getRegion()); metrics.setRegion(value.getRegion());
...@@ -65,27 +78,30 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E ...@@ -65,27 +78,30 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E
metrics.setCampaignId(value.getCampaign_id()); metrics.setCampaignId(value.getCampaign_id());
metrics.setEvent_name(value.getEvent()); metrics.setEvent_name(value.getEvent());
metrics.setMediaSource(value.getMedia_source()); metrics.setMediaSource(value.getMedia_source());
if (account.isPresent()) { Long date = userTime.value();
GameAccount gameAccount = account.get(); if (date == null){
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(gameAccount.getEvent_time()))); Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (account.isPresent()) {
} else if (EventTypeEnum.ACTIVATION.getName().equals(value.getEvent())) { GameAccount gameAccount = account.get();
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time()))); metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(gameAccount.getEvent_time())));
metrics.setEvent_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time()))); userTime.update(metrics.getDataDate());
metrics.setTimes(1L); } else if (EventTypeEnum.ACTIVATION.getName().equals(value.getEvent())) {
out.collect(metrics); metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
} else { userTime.update(metrics.getDataDate());
Optional<GameAccount> byAccountPrimary = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER); } else {
if (byAccountPrimary.isPresent()) { Optional<GameAccount> byAccountPrimary = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
GameAccount registerAccount = byAccountPrimary.get(); if (byAccountPrimary.isPresent()) {
if (null != registerAccount.getEvent_time()) { GameAccount registerAccount = byAccountPrimary.get();
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(registerAccount.getEvent_time()))); if (null != registerAccount.getEvent_time()) {
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(registerAccount.getEvent_time())));
userTime.update(metrics.getDataDate());
}
} }
} }
} else {
metrics.setDataDate(date);
} }
metrics.setEvent_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time()))); metrics.setEvent_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
metrics.setTimes(1L); metrics.setTimes(1L);
out.collect(metrics); out.collect(metrics);
} }
......
package com.fshark.overseas.advert.processor; package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.GameAccount; import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao; import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; import com.fshark.overseas.advert.modle.AppsFlyerEvent;
...@@ -8,24 +7,24 @@ import com.fshark.overseas.advert.modle.EventTypeEnum; ...@@ -8,24 +7,24 @@ 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.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.util.Date;
import java.util.Locale; import java.util.Locale;
import java.util.Optional; import java.util.Optional;
public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> { /**
* 设备激活
*/
public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
private AppConfig appConfig; private AppConfig appConfig;
private JdbiContext jdbiContext; private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao; private GameAccountDao gameAccountDao;
public CreateRoleProcess(AppConfig appConfig) { public SdkProcess(AppConfig appConfig) {
this.appConfig = appConfig; this.appConfig = appConfig;
} }
...@@ -41,47 +40,30 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -41,47 +40,30 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
ActivationMetrics activationMetrics = new ActivationMetrics(); ActivationMetrics activationMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics); copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid()); activationMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time()); String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
LocalDateTime activition; if (EventTypeEnum.SDK_ACTIVATION.getName().equals(value.getEvent())){
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time))); Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_ACTIVATION);
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time))); if (!account.isPresent()){
if (!account.isPresent()) { // 设备未激活 gameAccountDao.insertSdkActivition(value, event_time);
gameAccountDao.insertActivition(value, event_time); activationMetrics.setData_sdk_activation(1L);
activationMetrics.setData_activation(1L); activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activition = value.getEvent_time(); activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
// return; out.collect(activationMetrics);
} else { }
GameAccount gameAccount = account.get();
String activationTime = gameAccount.getEvent_time();
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); if (EventTypeEnum.SDK_REGISTER.getName().equals(value.getEvent())){
if (!accountRegister.isPresent()) { Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_REGISTER);
activationMetrics.setData_register(1L); if (!account.isPresent()){
activationMetrics.setData_device_register(1L); gameAccountDao.insertSdkRegister(value, event_time);
activationMetrics.setData_day_register_create(1L);//当日注册并创角 activationMetrics.setData_sdk_register(1L);
if (DateUtils.daysBetween(activition, value.getEvent_time()) == 0) { //当日激活并且注册 activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_day_act_register(1L); activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
} out.collect(activationMetrics);
} 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);
out.collect(activationMetrics);
}
}
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) { private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
metrics.setTime_zone(value.getTime_zone()); metrics.setTime_zone(value.getTime_zone());
...@@ -93,7 +75,6 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -93,7 +75,6 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
metrics.setAd_id(value.getAd_id()); metrics.setAd_id(value.getAd_id());
metrics.setRegion(value.getRegion()); metrics.setRegion(value.getRegion());
metrics.setChannel(value.getChannel()); metrics.setChannel(value.getChannel());
metrics.setData_type(1); metrics.setData_type(1);
if (value.getIs_retargeting()) { //再营销数据 if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2); metrics.setData_type(2);
...@@ -101,6 +82,7 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -101,6 +82,7 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据 if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据
metrics.setData_type(3); metrics.setData_type(3);
} }
} }
@Override @Override
......
...@@ -70,6 +70,14 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<SinkMetricEvent ...@@ -70,6 +70,14 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<SinkMetricEvent
metricsDetailDao.insertMetricDetail(element, "data_active_day_dau"); metricsDetailDao.insertMetricDetail(element, "data_active_day_dau");
} }
if (element.getData_sdk_activation() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_sdk_activition");
}
if (element.getData_sdk_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_sdk_register");
}
if (element.getDateRemind() > 0) { if (element.getDateRemind() > 0) {
String column = "retention_day" + element.getDateRemind(); String column = "retention_day" + element.getDateRemind();
metricsDetailDao.insertMetricDetail(element, column); metricsDetailDao.insertMetricDetail(element, column);
......
...@@ -25,7 +25,8 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> { ...@@ -25,7 +25,8 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> {
@Override @Override
public void invoke(ActivationMetrics value, Context context) throws Exception { public void invoke(ActivationMetrics value, Context context) throws Exception {
super.invoke(value, context); super.invoke(value, context);
if (value.getData_activation() + value.getData_register() + value.getData_device_register() + value.getData_create_role() + value.getData_day_act_register() + value.getData_day_register_create() + value.getData_active_day_dau() > 0) { if (value.getData_activation() + value.getData_register() + value.getData_device_register() + value.getData_create_role() + value.getData_day_act_register()
+ value.getData_day_register_create() + value.getData_active_day_dau() + value.getData_sdk_activation() + value.getData_sdk_register() > 0) {
metricsDao.insertActivationMetrics(value); metricsDao.insertActivationMetrics(value);
} }
if (value.getData_register() > 0){ if (value.getData_register() > 0){
......
...@@ -75,6 +75,8 @@ public class Constants { ...@@ -75,6 +75,8 @@ public class Constants {
public static final String TYPE_CREATEROLE = "createRole"; 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";
public static final String TYPE_SDK_ACTIVATION = "sdkActivation";
public static final String TYPE_SDK_REGISTER = "sdkRegister";
public static final String TYPE_SINK="mysql"; public static final String TYPE_SINK="mysql";
......
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