Commit bb95876a authored by sheteng's avatar sheteng

调整数据

parent b28f2373
......@@ -3,31 +3,19 @@ package com.fshark.overseas.advert;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.fshark.overseas.advert.processor.*;
import com.fshark.overseas.advert.sink.ActicitionSink;
import com.fshark.overseas.advert.sink.PurchaseSink;
import com.fshark.overseas.advert.sink.RemainSink;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
......@@ -36,7 +24,7 @@ import java.util.concurrent.TimeUnit;
* 广告计算
*/
public class AppsflyerAnalyze {
static AppConfig loadConfigFromArgs(String[] args) throws Exception {
static AppConfig loadConfigFromArgs(String[] args) {
ParameterTool parameters = ParameterTool.fromArgs(args);
AppConfig appConfig = new AppConfig();
try {
......@@ -81,7 +69,7 @@ public class AppsflyerAnalyze {
properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
// TODO: 2021/8/10 设置消费开始的时间戳1628524944000
// dataConsumer.setStartFromTimestamp(1630598733000L);
// dataConsumer.setStartFromTimestamp(1631548945000L);
Properties props2 = new Properties();
props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers());
......
......@@ -17,14 +17,11 @@ import com.fshark.overseas.advert.sink.server.ServerActivationSink;
import com.fshark.overseas.advert.sink.server.ServerRemindSink;
import com.fshark.overseas.advert.util.AppConfig;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
......@@ -35,7 +32,7 @@ import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class AppsflyerMetricStore {
static AppConfig loadConfigFromArgs(String[] args) throws Exception {
static AppConfig loadConfigFromArgs(String[] args) {
ParameterTool parameters = ParameterTool.fromArgs(args);
AppConfig appConfig = new AppConfig();
try {
......@@ -77,12 +74,15 @@ public class AppsflyerMetricStore {
properties.setProperty("auto.offset.reset", "latest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
DataStream<SinkMetricEvent> activationMetricsDataStream = sse.addSource(dataConsumer).flatMap(new FlatMapFunction<String, SinkMetricEvent>() {
@Override
public void flatMap(String value, Collector<SinkMetricEvent> out) throws Exception {
out.collect(new Gson().fromJson(value, SinkMetricEvent.class));
}
}).uid("parse-event").name("parse-event");
DataStream<SinkMetricEvent> activationMetricsDataStream = sse.addSource(dataConsumer)
.flatMap(new RichFlatMapFunction<String, SinkMetricEvent>() {
@Override
public void flatMap(String value, Collector<SinkMetricEvent> out) throws Exception {
out.collect(new Gson().fromJson(value, SinkMetricEvent.class));
}
})
.uid("parse-event")
.name("parse-event");
// activationMetricsDataStream.print();
//角色注册激活预聚合
......@@ -129,7 +129,7 @@ public class AppsflyerMetricStore {
DataStream<ServerActivationMetrics> activationMetricsList = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) throws Exception {
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) {
if (value.isServer()) {
out.collect(value.getServerActivationMetrics());
}
......@@ -149,7 +149,7 @@ public class AppsflyerMetricStore {
DataStream<ServerRemainMetrics> remainMetrics = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) throws Exception {
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) {
if (value.isServer()) {
out.collect(value.getServerActivationMetrics());
}
......
......@@ -2,11 +2,8 @@ package com.fshark.overseas.advert.entity.dao;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.RegisterAccount;
import com.fshark.overseas.advert.entity.mapper.GameAccountMapper;
import com.fshark.overseas.advert.entity.mapper.RegisterAccountMapper;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import org.apache.kafka.common.protocol.types.Field;
import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
import org.jdbi.v3.sqlobject.customizer.Bind;
import org.jdbi.v3.sqlobject.customizer.BindBean;
......@@ -32,6 +29,10 @@ public interface GameAccountDao {
"values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'activation',:reg_time) ")
void insertActivition(@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,'retargeting',:reg_time) ")
void insertRetargeting(@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,'register',:reg_time ) ")
......
......@@ -8,9 +8,9 @@ public interface OrderDao {
@SqlUpdate("insert into appsflyer_pay_record (region, did, uid,platform, game_id, media_source, campaign_id, ad_set_id,ad_id, " +
" pay_time, money, currency, order_no, reg_time, activation_time, time_zone ,channel,data_type)" +
" pay_time, money, currency, order_no, reg_time, activation_time, time_zone ,channel,data_type,retargeting_time)" +
" values(:region, :did, :uid, :platform, :game_id, :media_source, :campaign_id, :ad_set_id , :ad_id," +
" :pay_time, :revenue,:currency,:orderno,:reg_time,:activation_time,:time_zone,:channel,:data_type)")
" :pay_time, :revenue,:currency,:orderno,:reg_time,:activation_time,:time_zone,:channel,:data_type,:retargeting_time)")
void insertOrder(@BindBean OrderData orderData);
@SqlUpdate("insert into appsflyer_pay_record_server (region, did, uid,platform, game_id, media_source, campaign_id, ad_set_id,ad_id, " +
......
......@@ -2,7 +2,6 @@ package com.fshark.overseas.advert.modle;
import com.fshark.overseas.advert.util.Constants;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
......@@ -45,10 +44,7 @@ public enum EventTypeEnum {
public static boolean roleLoginEvent(AppsFlyerEvent event) {
return (ROLE_LOGIN.getName().equals(event.getEvent())
|| FAST_LOGIN.getName().equals(event.getEvent())
|| CREATE_ROLE.getName().equals(event.getEvent())
|| GAME_LOGIN.getName().equals(event.getEvent())
|| SDK_LOGIN.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
|| CREATE_ROLE.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean createRoleEvent(AppsFlyerEvent event) {
......
......@@ -2,14 +2,11 @@ package com.fshark.overseas.advert.modle;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.util.DateUtils;
import lombok.Data;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
/**
......@@ -140,6 +137,11 @@ public class OrderData implements Serializable {
*/
private String activation_time;
/**
* 归因时间
*/
private String retargeting_time;
private String reg_server_id;
......
......@@ -42,13 +42,26 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
if (!account.isPresent() || value.getIs_retargeting()) { //
if (!account.isPresent()) { //
gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
out.collect(activationMetrics);
}
if (value.getIs_retargeting()) {
Optional<GameAccount> byPrimary = gameAccountDao.findByPrimary(value, Constants.TYPE_RETARGETING);
if (!byPrimary.isPresent()) {
gameAccountDao.insertRetargeting(value, event_time);
activationMetrics.setData_activation(1L);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
out.collect(activationMetrics);
}
}
}
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
......
......@@ -3,7 +3,6 @@ package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.ServerRoleAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.entity.dao.OrderDao;
import com.fshark.overseas.advert.entity.dao.ServerRoleDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsflyerPurchase;
......@@ -61,6 +60,15 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE
if (accountRegister.isPresent()) {
GameAccount gameAccount = accountRegister.get();
orderData.setReg_time(gameAccount.getEvent_time());
if (orderData.getActivation_time() == null) {
orderData.setActivation_time(gameAccount.getEvent_time());
}
}
Optional<GameAccount> accountRetarget = gameAccountDao.findByPrimary(value, Constants.TYPE_RETARGETING);
if (accountRetarget.isPresent()) {
GameAccount gameAccount = accountRetarget.get();
orderData.setRetargeting_time(gameAccount.getEvent_time());
}
Optional<ServerRoleAccount> serverAccount = serverRoleDao.findByPrimary(value, appsflyerPurchase.getServerid(), Constants.TYPE_ACTIVIYION);
......
......@@ -11,6 +11,7 @@ import com.fshark.overseas.advert.util.*;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.time.LocalDateTime;
import java.util.Locale;
import java.util.Optional;
......@@ -46,8 +47,8 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
String activationTime = event_time;
if (!account.isPresent()) { // 设备未激活
gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L);
// gameAccountDao.insertActivition(value, event_time);
// activationMetrics.setData_activation(1L);
// return;
} else {
GameAccount gameAccount = account.get();
......@@ -65,6 +66,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) {
gameAccountDao.insertRegister(value, event_time);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_register_create(1L);//当日注册并创角
......@@ -76,7 +78,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
activationMetrics.setData_day_register_create(1L);
}
}
gameAccountDao.insertRegister(value, event_time);
} else {
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
......@@ -84,32 +86,35 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
gameAccountDao.insertRegister(value, event_time);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
if (DateUtils.daysBetween(DateUtils.formatAsDate(activationTime), value.getEvent_time()) == 0) { //当日激活并且注册
activationMetrics.setData_day_act_register(1L);
}
} else {
activationTime = accountRegister.get().getEvent_time();
}
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
retentMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
retentMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
CommonUtils.setType(activationMetrics);
CommonUtils.setType(retentMetrics);
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_time();
LocalDateTime date = DateUtils.formatAsDate(last_login_time);
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
retentMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
retentMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
CommonUtils.setType(activationMetrics);
CommonUtils.setType(retentMetrics);
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_time();
LocalDateTime date = DateUtils.formatAsDate(last_login_time);
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
gameAccountDao.insertRoleLogin(value);
}
......
package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.entity.dao.OrderDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsflyerPurchase;
import com.fshark.overseas.advert.modle.OrderData;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.util.Optional;
public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
private AppConfig appConfig;
......
......@@ -30,10 +30,10 @@ public class RemainSink extends RichSinkFunction<RemainMetrics> {
jdbiContext.getJdbi().withHandle(handle -> {
String sql = String.format("insert into appsflyer_retention" +
" (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,%s)" +
" values ('%s','%s','%s','%s','%s','%s','%s','%s','KRW',%s) ",
" (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,`data_type`,%s)" +
" values ('%s','%s','%s','%s','%s','%s','%s','%s','KRW',%s,%s) ",
column, value.getData_date(), value.getRegion(), value.getPlatform(), value.getGame_id(), value.getMedia_source(), value.getCampaign_id()
, value.getAd_set_id(), value.getAd_id(), value.getUser_num()
, value.getAd_set_id(), value.getAd_id(), value.getData_type(), value.getUser_num()
);
handle.createUpdate(sql).execute();
return null;
......@@ -60,7 +60,7 @@ public class RemainSink extends RichSinkFunction<RemainMetrics> {
@Override
public void open(Configuration parameters) throws Exception {
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());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
......
......@@ -70,6 +70,7 @@ public class Constants {
public static final String REMATCHED = "rematched";
public static final String TYPE_ACTIVIYION = "activation";
public static final String TYPE_RETARGETING = "retargeting";
public static final String TYPE_REGISTER = "register";
public static final String TYPE_CREATEROLE = "createRole";
public static final String TYPE_ROLELOGIN = "rolelogin";
......
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