Commit 10a88757 authored by sheteng's avatar sheteng

新ltv

parent 72a25268
/build/
/.gradle
/.gradle/
/.idea/
......@@ -24,7 +24,7 @@ version = '0.3'
mainClassName = 'com.fshark.overseas.advert.AppsflyerAnalyze'
ext {
javaVersion = '1.8'
flinkVersion = '1.11.1'
flinkVersion = '1.12.0'
scalaBinaryVersion = '2.12'
slf4jVersion = '1.7.7'
log4jVersion = '1.2.17'
......@@ -52,8 +52,6 @@ dependencies {
compile 'com.zaxxer:HikariCP:3.4.5'
compile group: 'org.jdbi', name: 'jdbi3-core', version: '3.18.0'
compile group: 'org.jdbi', name: 'jdbi3-sqlobject', version: '3.18.0'
compile 'redis.clients:jedis:3.5.1'
compile 'com.squareup.okhttp3:okhttp:3.14.1'
implementation 'com.google.code.gson:gson:2.8.4'
compile 'com.alibaba:fastjson:1.2.62'
compile "org.apache.flink:flink-connector-kafka_${scalaBinaryVersion}:${flinkVersion}"
......
......@@ -2,7 +2,6 @@ package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerRemainMetrics;
import com.fshark.overseas.advert.processor.*;
......@@ -10,7 +9,7 @@ import com.fshark.overseas.advert.processor.server.ServerActivationAggregateFunc
import com.fshark.overseas.advert.processor.server.ServerActivationResultWindows;
import com.fshark.overseas.advert.processor.server.ServerRemainProcessor;
import com.fshark.overseas.advert.sink.ActicitionSink;
import com.fshark.overseas.advert.sink.EventTimesSink;
import com.fshark.overseas.advert.sink.EventTimesSinkProcess;
import com.fshark.overseas.advert.sink.PurchaseSink;
import com.fshark.overseas.advert.sink.RemainSink;
import com.fshark.overseas.advert.sink.server.ServerActivationSink;
......@@ -83,38 +82,29 @@ public class AppsflyerMetricStore {
})
.uid("parse-event")
.name("parse-event");
// activationMetricsDataStream.print();
//角色注册激活预聚合
activationMetricsDataStream.print();
//角色注册
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
// .keyBy(ActivationMetrics::getKey)
// .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
// .aggregate(new ActivationAggregateFunction(), new ActivationResultWindows())
.uid("activation-agg-event")
.name("activation-agg-event");
// 角色注册激活写入
activitionResult.addSink(new ActicitionSink(appConfig))
.uid("login-sink-event")
.name("login-sink-event");
//
// //指标详细记录
activationMetricsDataStream.timeWindowAll(Time.minutes(1)).process(new StoreMetricProcess(appConfig))
//指标详细记录
activitionResult.timeWindowAll(Time.seconds(1)).process(new StoreMetricProcess(appConfig))
.uid("metric-detail")
.name("metric-detail");
// 留存预聚合
DataStream<RemainMetrics> remainResult = activationMetricsDataStream.flatMap(new RemainProcess())
.keyBy(RemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("user_num")
.uid("remain-event")
.name("remain-event");
//留存写入
remainResult
.addSink(new RemainSink(appConfig))
.uid("remain-sink-event")
.name("remain-sink-event");
// 角色注册激活写入
// activitionResult.addSink(new ActicitionSink(appConfig))
// .uid("login-sink-event")
// .name("login-sink-event");
// 留存写入
// activationMetricsDataStream
// .flatMap(new RemainProcess())
// .timeWindowAll(Time.seconds(30)).process()
// .addSink(new RemainSink(appConfig))
// .uid("remain-sink-event")
// .name("remain-sink-event");
//订单写入
activationMetricsDataStream
......@@ -122,8 +112,8 @@ public class AppsflyerMetricStore {
.addSink(new PurchaseSink(appConfig))
.uid("order-sink-event")
.name("order-sink-event");
//
//
//角色注册激活预聚合
DataStream<ServerActivationMetrics> activationMetricsList = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
......@@ -139,12 +129,12 @@ public class AppsflyerMetricStore {
.aggregate(new ServerActivationAggregateFunction(), new ServerActivationResultWindows())
.uid("server-activation-agg-event")
.name("server-activation-agg-event");
//
activationMetricsList.addSink(new ServerActivationSink(appConfig))
.uid("server-activationsink-event")
.name("server-activationsink-event");
//留存预聚合
//
// //留存预聚合
DataStream<ServerRemainMetrics> remainMetrics = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override
......@@ -168,7 +158,8 @@ public class AppsflyerMetricStore {
* 事件写入
*/
activationMetricsDataStream.filter(SinkMetricEvent::isEvent)
.addSink(new EventTimesSink(appConfig))
.timeWindowAll(Time.seconds(30))
.process(new EventTimesSinkProcess(appConfig))
.uid("event-sink-event")
.name("event-sink-event");
......
package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.processor.*;
import com.fshark.overseas.advert.sink.EventInfoSinkProcess;
import com.fshark.overseas.advert.util.AppConfig;
import org.apache.flink.api.common.functions.FilterFunction;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
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.util.Collector;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class AppsflyerOuidAnalyze {
static AppConfig loadConfigFromArgs(String[] args) {
ParameterTool parameters = ParameterTool.fromArgs(args);
AppConfig appConfig = new AppConfig();
try {
appConfig.setSourceKafkaBootstrapServers(parameters.get("source-kafka-bootstrap-servers"));
appConfig.setSourceKafkaGroupId(parameters.get("source-kafka-group-id"));
appConfig.setSourceKafkaTopic(parameters.get("source-kafka-topic"));
appConfig.setDbHost(parameters.get("jdbc-host"));
appConfig.setDbUser(parameters.get("jdbc-user"));
appConfig.setDbPassword(parameters.get("jdbc-password"));
appConfig.setSinkType(parameters.get("sink-type"));
} catch (Exception e) {
System.err.println("Place set args: " +
" --source-kafka-bootstrap-servers <source-kafka-bootstrap-servers>" +
" --source-kafka-group-id <source-kafka-group-id>" +
" --source-kafka-topic <source-kafka-topic>" +
" --sink-kafka-bootstrap-servers <sink-kafka-bootstrap-servers>" +
" --sink-kafka-group-id <sink-kafka-group-id>" +
" --sink-kafka-topic <sink-kafka-topic>" +
" --jdbc-host <jdbc-host> " +
" --jdbc-user <jdbc-user> " +
" --jdbc-password <jdbc-password> ");
throw e;
}
return appConfig;
}
public static void main(String[] args) throws Exception {
AppConfig appConfig = loadConfigFromArgs(args);
final StreamExecutionEnvironment sse = StreamExecutionEnvironment.getExecutionEnvironment();
sse.setParallelism(1);
sse.enableCheckpointing(1000 * 60 * 10);
sse.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS)));
//source设置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers());
properties.setProperty("group.id", appConfig.getSourceKafkaGroupId());
properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer)
.flatMap(new ParseEventProcessor())
.filter(new FilterFunction<AppsFlyerEvent>() {
@Override
public boolean filter(AppsFlyerEvent value) throws Exception {
if (value.getMedia_source().contains("rank") || value.getCampaign_id().contains("rank")) {
return false;
}
return true;
}
})
.uid("parse-event")
.name("parse-event")
.shuffle();
adEventDataStream.print();
//ouid注册
DataStream<ActivationMetrics> ouidMetrics = adEventDataStream
.filter(EventTypeEnum::loginEvent)
.flatMap(new OuidLoginProcess(appConfig))
.uid("sdk-event")
.name("sdk-event");
//写入指标
ouidMetrics.print();
ouidMetrics.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new StoreMetricProcess(appConfig))
.uid("store-metrics")
.name("store-metrics");
// ouidMetrics.timeWindowAll(Time.minutes(1));
//订单信息
adEventDataStream.filter(EventTypeEnum::orderEvent).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new OuidOrderDataProcessor(appConfig))
.uid("order-metrics")
.name("order-metrics");
sse.execute("appsflyer-analyze");
}
}
......@@ -19,6 +19,12 @@ public class GameAccount {
private String mediaSource;
private String campaign_id;
private String ad_set_id;
private String ad_id;
private String event_type;
private String event_time;
......
......@@ -24,6 +24,13 @@ public interface GameAccountDao {
@RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findByPrimary(@BindBean AppsFlyerEvent event, @Bind("event_type") String event_type);
@SqlQuery("SELECT * FROM appsflyer_activation_devices WHERE game_id=:game_id AND ouid=:ouid And platform =:platform AND event_type ='ouidRegister' order by event_time desc limit 1")
@RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findByOuid(@BindBean AppsFlyerEvent event,@Bind("ouid") String ouid);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`,`ouid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
"values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'ouidRegister',:reg_time) ")
void insertOuidRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time,@Bind("ouid") String ouid);
@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,'activation',:reg_time) ")
......@@ -34,9 +41,9 @@ public interface GameAccountDao {
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 ) ")
void insertRegister(@BindBean AppsFlyerEvent value, @Bind("reg_time") String reg_time);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`,`ouid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'register',:reg_time ) ")
void insertRegister(@BindBean AppsFlyerEvent value, @Bind("reg_time") String reg_time,@Bind("ouid") String ouid);
@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`) " +
......@@ -51,10 +58,6 @@ public interface GameAccountDao {
" 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);
@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,6 +19,12 @@ public interface OrderDao {
" :pay_time, :revenue,:currency,:orderno,:time_zone,:channel,:server_reg_date,:server_reg_hour,:reg_server_id)")
void insertServerOrder(@BindBean OrderData eventData);
@SqlUpdate("insert into appsflyer_pay_record_ouid (region, did, uid,ouid,platform, game_id, media_source, campaign_id, ad_set_id,ad_id, " +
" pay_time, money, currency, order_no, ouid_reg_time, time_zone ,channel,data_type)" +
" values(:region, :did, :uid, :ouid,:platform, :game_id, :media_source, :campaign_id, :ad_set_id , :ad_id," +
" :pay_time, :revenue,:currency,:orderno,:ouid_reg_time,:time_zone,:channel,:data_type)")
void insertOuidOrder(@BindBean OrderData orderData);
// @SqlUpdate("insert into appsflyer_order_accumulative" +
// " (`ds`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,`%s`" +
// " values()")
......
......@@ -16,6 +16,9 @@ public class GameAccountMapper implements RowMapper<GameAccount> {
.did(rs.getString("did"))
.platform(rs.getString("platform"))
.mediaSource(rs.getString("media_source"))
.campaign_id(rs.getString("campaign_id"))
.ad_set_id(rs.getString("ad_set_id"))
.ad_id(rs.getString("ad_id"))
.event_type(rs.getString("event_type"))
.event_time(rs.getString("event_time"))
.build();
......
......@@ -448,6 +448,9 @@ public class AppsFlyerEvent implements Serializable {
if (StringUtils.isBlank(af_channel)) {
af_channel = Constants.DEFAULT_FIELD_VALUE;
}
if(StringUtils.equals(af_channel,"Pangle")){
appsFlyerEvent.setMedia_source("Pangle");
}
appsFlyerEvent.setChannel(af_channel);
appsFlyerEvent.setApp_name(
......@@ -514,8 +517,4 @@ public class AppsFlyerEvent implements Serializable {
jsonObject.put(v1Col, jsonObject.get(v2Col));
}
public boolean isInstallEvent(){
return EventTypeEnum.ROLE_LOGIN.equals(event);
}
}
\ No newline at end of file
......@@ -16,12 +16,13 @@ public enum EventTypeEnum {
第一次创角作为注册
*/
// REGISTER("createrole"),
CREATE_ROLE("createrole"),
CREATE_ROLE("rolelogin"),
SDK_ACTIVATION("activation"),
SDK_REGISTER("login"),
SDK_REGISTER("register"),
FAST_LOGIN("fast login"),
GAME_LOGIN("gamelogin"),
ROLE_LOGIN("rolelogin"),
ROLE_LOGIN("login"),
LOGIN("login"),
ACTIVATION("install"),
PURCHASE("af_purchase"),
RE_ENGAGEMENT("re-engagement"),
......@@ -44,8 +45,7 @@ public enum EventTypeEnum {
}
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);
return (SDK_REGISTER.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean roleLoginEvent(AppsFlyerEvent event) {
......@@ -71,6 +71,10 @@ public enum EventTypeEnum {
return PURCHASE.getName().equals(event.getEvent()) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean loginEvent(AppsFlyerEvent event) {
return LOGIN.getName().equals(event.getEvent()) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean notHeartBeatEvent(AppsFlyerEvent event) {
return !HEARTBEAT.getName().equals(event.getEvent()) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
......
......@@ -41,6 +41,8 @@ public class OrderData implements Serializable {
*/
private String uid;
private String ouid;
/**
* 设备唯一标识
* ios --> idfa
......@@ -151,6 +153,8 @@ public class OrderData implements Serializable {
private String server_reg_hour;
private String ouid_reg_time;
private int data_type;
......
......@@ -53,4 +53,8 @@ public class ActivationMetrics extends BaseMetrics implements Serializable {
* 留存天数
*/
private Long dateRemind = 0L;
/**
* ouid 注册
*/
private long data_ouid_register = 0L;
}
package com.fshark.overseas.advert.processor;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
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 org.apache.http.util.TextUtils;
import java.util.Locale;
import java.util.Optional;
/**
* 设备激活
*/
public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
public OuidLoginProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
String ouid = jsonObject.getString("ouid");
if (TextUtils.isEmpty(ouid)) {
return;
}
ActivationMetrics activationMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid());
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
Optional<GameAccount> account = gameAccountDao.findByOuid(value,ouid);
if (!account.isPresent()) {
gameAccountDao.insertOuidRegister(value, event_time,ouid);
activationMetrics.setData_ouid_register(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) {
metrics.setTime_zone(value.getTime_zone());
metrics.setPlatform(value.getPlatform());
metrics.setMedia_source(value.getMedia_source());
metrics.setGame_id(value.getGame_id());
metrics.setCampaign_id(value.getCampaign_id());
metrics.setAd_set_id(value.getAd_set_id());
metrics.setAd_id(value.getAd_id());
metrics.setRegion(value.getRegion());
metrics.setChannel(value.getChannel());
metrics.setData_type(1);
if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2);
}
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据
metrics.setData_type(3);
}
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
package com.fshark.overseas.advert.processor;
import com.alibaba.fastjson.JSONObject;
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;
import com.fshark.overseas.advert.modle.OrderData;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.http.util.TextUtils;
import java.util.Optional;
public class OuidOrderDataProcessor extends ProcessAllWindowFunction<AppsFlyerEvent, Object, TimeWindow> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
private OrderDao orderDao;
public OuidOrderDataProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
orderDao = jdbiContext.getJdbi().onDemand(OrderDao.class);
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
@Override
public void process(Context context, Iterable<AppsFlyerEvent> elements, Collector<Object> out) throws Exception {
for (AppsFlyerEvent value : elements) {
try {
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
String ouid = jsonObject.getString("ouid");
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null || TextUtils.isEmpty(ouid)) { //不是我们上报的格式
return;
}
if ("ios".equalsIgnoreCase(value.getPlatform())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
OrderData orderData = new OrderData(value);
orderData.setOuid(ouid);
Optional<GameAccount> ouidRegister = gameAccountDao.findByOuid(value, ouid);
if (ouidRegister.isPresent()) {
GameAccount gameAccount = ouidRegister.get();
orderData.setMedia_source(gameAccount.getMediaSource());
orderData.setCampaign_id(gameAccount.getCampaign_id());
orderData.setAd_set_id(gameAccount.getAd_set_id());
orderData.setAd_id(gameAccount.getAd_id());
orderData.setOuid_reg_time(gameAccount.getEvent_time());
orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())), orderData.getMedia_source(), orderData.getData_type()));
orderDao.insertOuidOrder(orderData);
}
} catch (Exception e) {
return;
}
}
}
}
package com.fshark.overseas.advert.processor;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
......@@ -8,9 +9,11 @@ import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.http.util.TextUtils;
import java.time.LocalDateTime;
import java.util.Locale;
......@@ -35,6 +38,11 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
String ouid = jsonObject.getString("ouid");
if (TextUtils.isEmpty(ouid)) {
return;
}
ActivationMetrics activationMetrics = new ActivationMetrics();
ActivationMetrics retentMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics);
......@@ -66,7 +74,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);
gameAccountDao.insertRegister(value, event_time,ouid);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_register_create(1L);//当日注册并创角
......@@ -83,7 +91,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);
gameAccountDao.insertRegister(value, event_time,ouid);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
if (DateUtils.daysBetween(DateUtils.formatAsDate(activationTime), value.getEvent_time()) == 0) { //当日激活并且注册
......
......@@ -41,16 +41,6 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid());
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
if (EventTypeEnum.SDK_ACTIVATION.getName().equals(value.getEvent())){
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_ACTIVATION);
if (!account.isPresent()){
gameAccountDao.insertSdkActivition(value, event_time);
activationMetrics.setData_sdk_activation(1L);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
out.collect(activationMetrics);
}
}
if (EventTypeEnum.SDK_REGISTER.getName().equals(value.getEvent())){
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_REGISTER);
......
package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.EventInfoDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DatabaseUtils;
import com.fshark.overseas.advert.util.JdbiContext;
import com.fshark.overseas.advert.util.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
public class EventTimesSink extends RichSinkFunction<SinkMetricEvent> {
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
public class EventTimesSinkProcess extends ProcessAllWindowFunction<SinkMetricEvent, Object, TimeWindow> {
private AppConfig appConfig;
public EventTimesSink(AppConfig appConfig) {
public EventTimesSinkProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
private JdbiContext jdbiContext;
private EventInfoDao eventInfoDao;
//定义连接、预编译语句
private Connection conn;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
Class.forName("com.mysql.cj.jdbc.Driver");
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
}
eventInfoDao = jdbiContext.getJdbi().onDemand(EventInfoDao.class);
conn = DriverManager.getConnection(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
if (conn != null) {
conn.close();
}
}
@Override
public void invoke(SinkMetricEvent value, Context context) throws Exception {
super.invoke(value, context);
if (value.isEvent()){
eventInfoDao.insertEventTimes(value.getEventMetrics());
public void process(Context context, Iterable<SinkMetricEvent> elements, Collector<Object> out) throws Exception {
try {
PreparedStatement insertStmt = conn.prepareStatement("insert into appsflyer_event_times (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`event_name`,`event_date`,`times`) " +
"values (?,?,?,?,?,?,?,?,?)");
for (SinkMetricEvent sinkMetricEvent : elements) {
EventMetrics element = sinkMetricEvent.getEventMetrics();
insertStmt.setLong(1, element.getDataDate());
insertStmt.setString(2, element.getRegion());
insertStmt.setString(3, element.getPlatform());
insertStmt.setString(4, element.getGameId());
insertStmt.setString(5, element.getMediaSource());
insertStmt.setString(6, element.getCampaignId());
insertStmt.setString(7, element.getEvent_name());
insertStmt.setLong(8, element.getEvent_date());
insertStmt.setLong(9, element.getTimes());
insertStmt.addBatch();
}
insertStmt.executeBatch(); //批量执行
conn.commit();
insertStmt.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
package com.fshark.overseas.advert.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.exceptions.JedisException;
import java.util.function.Function;
/**
* @author libing
* @version 1.0
* @date 2019-08-02 16:06
*/
public class RedisClientUtils {
private static Logger LOG = LoggerFactory.getLogger(RedisClientUtils.class);
public static <R> R using(JedisPool jedisPool, int maxRetry, Function<Jedis, R> apply) {
do {
try (Jedis client = jedisPool.getResource()) {
return apply.apply(client);
} catch (JedisException e) {
LOG.error("retry " + maxRetry, e);
}
} while (--maxRetry >= 0);
return null;
}
public static JedisPoolConfig getGeneralRedisPoolConfig() {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(2);
jedisPoolConfig.setMaxIdle(2);
jedisPoolConfig.setTestWhileIdle(true);
return jedisPoolConfig;
}
}
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