Commit dcbd035e authored by she's avatar she

ouid创角

parent e4f6b39d
......@@ -4,8 +4,11 @@
<component name="GradleSettings">
<option name="linkedExternalProjectsSettings">
<GradleProjectSettings>
<option name="delegatedBuild" value="true" />
<option name="testRunner" value="GRADLE" />
<option name="distributionType" value="DEFAULT_WRAPPED" />
<option name="externalProjectPath" value="$PROJECT_DIR$" />
<option name="gradleJvm" value="1.8" />
<option name="modules">
<set>
<option value="$PROJECT_DIR$" />
......
......@@ -55,10 +55,10 @@ dependencies {
// https://mvnrepository.com/artifact/mysql/mysql-connector-java
implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.25'
compile "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-java:${flinkVersion}"
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-java:${flinkVersion}"
compileOnly "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52'
// --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g.
......
......@@ -95,6 +95,12 @@ public class AppsflyerAnalyze {
}
return true;
}
}).process(new ProcessFunction<AppsFlyerEvent, AppsFlyerEvent>() {
@Override
public void processElement(AppsFlyerEvent value, ProcessFunction<AppsFlyerEvent, AppsFlyerEvent>.Context ctx, Collector<AppsFlyerEvent> out) throws Exception {
out.collect(value);
// if ()
}
})
.uid("parse-event")
.name("parse-event");
......@@ -127,7 +133,9 @@ public class AppsflyerAnalyze {
out.collect(value);
}
}
}).uid("shuffle-event").name("shuffle-event");
})
.uid("shuffle-event")
.name("shuffle-event");
DataStream<AppsFlyerEvent> installStream = mainStream.getSideOutput(installTag);
DataStream<AppsFlyerEvent> loginStream = mainStream.getSideOutput(loginTag);
......@@ -135,8 +143,6 @@ public class AppsflyerAnalyze {
DataStream<AppsFlyerEvent> orderStream = mainStream.getSideOutput(orderTag);
DataStream<AppsFlyerEvent> sdkRegisterStream = mainStream.getSideOutput(sdkRegisterTag);
// loginStream.print();
orderStream.print();
//广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = installStream
.flatMap(new ActivitionProcess(appConfig))
......@@ -162,25 +168,24 @@ public class AppsflyerAnalyze {
.name("sink-metric-event");
/**
* ouid相关指标计算 注册留存日活
* ouid相关指标计算 注册留存日活 新增创角指标
*/
DataStream<ActivationMetrics> ouidMetrics = loginStream
DataStream<ActivationMetrics> ouidMetrics = loginStream.union(roleloginStream)
.flatMap(new OuidLoginProcess(appConfig))
.uid("ouid-event")
.name("ouid-event");
// ouidMetrics.print("ouid");
ouidMetrics.flatMap(new RichFlatMapFunction<ActivationMetrics, String>() {
@Override
public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception {
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_OUID_METRIC);
out.collect(new Gson().toJson(sinkMetricEvent));
}
})
@Override
public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception {
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_OUID_METRIC);
out.collect(new Gson().toJson(sinkMetricEvent));
}
})
.addSink(kafkaSink)
.uid("sink_ouid-event")
.name("sink_ouid-event");
......@@ -200,14 +205,6 @@ public class AppsflyerAnalyze {
.uid("ouid_order-metrics")
.name("ouid_order-metrics");
//区服 角色注册,激活 暂时删除
// roleloginStream
// .flatMap(new ServerActivationProcessor(appConfig))
// .flatMap(new ServerMtricToStringProcessor())
// .addSink(kafkaSink)
// .uid("activation-event")
// .name("activation-event");
//事件统计激活
mainStream
.filter(EventTypeEnum::notHeartBeatEvent)
......
......@@ -34,7 +34,7 @@ public interface GameAccountDao {
void insertOuidRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time, @Bind("ouid") String ouid);
@SqlUpdate("insert into appsflyer_ouid (`platform`,`did`, `uid`,`ouid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`,`channel`) " +
" values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'ouidLogin',:event_time,:channel) " +
" values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,:event_type,:event_time,:channel) " +
" on duplicate key update event_time=values(event_time)")
void insertOuidEvent(@BindBean AppsFlyerEvent event, @Bind("ouid") String ouid, @Bind("event_type") String event_type);
......
......@@ -256,6 +256,8 @@ public class AppsFlyerEvent implements Serializable {
did = nvl(idfa, Constants.DEFAULT_FIELD_VALUE);
} else if (PlatformEnum.Android.getName().equals(platform)) {
did = nvl(advertisingId, Constants.DEFAULT_FIELD_VALUE);
} else if(PlatformEnum.PC.getName().equals(platform)) {
did = nvl(advertisingId, Constants.DEFAULT_FIELD_VALUE);
} else {
did = Constants.DEFAULT_FIELD_VALUE;
}
......
......@@ -33,7 +33,7 @@ public class ActivationMetrics extends BaseMetrics implements Serializable {
*/
private Long data_device_register = 0L;
/**
* 创角数 (不去重的)
* 创角数
*/
private Long data_create_role = 0L;
/**
......
......@@ -42,6 +42,7 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
if (jsonObject == null) return;
String ouid = jsonObject.getString("ouid");
if (TextUtils.isEmpty(ouid)) {
return;
......@@ -85,6 +86,18 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
activationMetrics.setData_active_day_dau(1L);
}
}
if (EventTypeEnum.createRoleEvent(value)) {
boolean isRole = false;
Optional<GameAccount> accountCreateRole = gameAccountDao.findByOuid(value, ouid,Constants.TYPE_CREATEROLE);
if (!accountCreateRole.isPresent()) {
activationMetrics.setData_create_role(1L);
gameAccountDao.insertOuidEvent(value,ouid,Constants.TYPE_CREATEROLE);
isRole = true;
}
if (DateUtils.daysBetween(DateUtils.formatAsDate(gameAccount.getEvent_time()), value.getEvent_time()) == 0 && isRole) { //当日注册创角
activationMetrics.setData_day_register_create(1L);
}
}
}
out.collect(retentMetrics);
out.collect(activationMetrics);
......
......@@ -12,6 +12,7 @@ import org.apache.http.util.TextUtils;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationMetrics, Object, TimeWindow> {
......@@ -47,26 +48,7 @@ public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationM
" (`uid`,`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,`metric_name`,`metric_value`)" +
" values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
for (ActivationMetrics element : elements) {
String name = dealMetrics(element);
if (!TextUtils.isEmpty(name)) {
insertStmt.setString(1, element.getUid());
insertStmt.setLong(2, element.getData_date());
insertStmt.setLong(3, element.getData_hour());
insertStmt.setString(4, element.getTime_zone());
insertStmt.setString(5, element.getRegion());
insertStmt.setString(6, element.getPlatform());
insertStmt.setString(7, element.getGame_id());
insertStmt.setString(8, element.getMedia_source());
insertStmt.setString(9, element.getCampaign_id());
insertStmt.setString(10, element.getAd_set_id());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getChannel());
insertStmt.setLong(13, element.getData_type());
insertStmt.setString(14, name);
insertStmt.setLong(15, 1l);
insertStmt.addBatch();
}
dealMetrics(element, insertStmt);
}
insertStmt.executeBatch(); //批量执行
conn.commit();
......@@ -76,50 +58,72 @@ public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationM
}
}
private String dealMetrics(ActivationMetrics element) {
private void createBatch(String name, PreparedStatement insertStmt, ActivationMetrics element) throws SQLException {
if (!TextUtils.isEmpty(name)) {
insertStmt.setString(1, element.getUid());
insertStmt.setLong(2, element.getData_date());
insertStmt.setLong(3, element.getData_hour());
insertStmt.setString(4, element.getTime_zone());
insertStmt.setString(5, element.getRegion());
insertStmt.setString(6, element.getPlatform());
insertStmt.setString(7, element.getGame_id());
insertStmt.setString(8, element.getMedia_source());
insertStmt.setString(9, element.getCampaign_id());
insertStmt.setString(10, element.getAd_set_id());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getChannel());
insertStmt.setLong(13, element.getData_type());
insertStmt.setString(14, name);
insertStmt.setLong(15, 1l);
insertStmt.addBatch();
}
}
private String dealMetrics(ActivationMetrics element, PreparedStatement insertStmt) throws SQLException {
if (element.getData_activation() > 0) {
return "data_activation";
createBatch("data_activation", insertStmt, element);
}
if (element.getData_register() > 0) {
return "data_register";
createBatch("data_register", insertStmt, element);
}
if (element.getData_device_register() > 0) {
return "data_device_register";
createBatch("data_device_register", insertStmt, element);
}
if (element.getData_create_role() > 0) {
return "data_create_role";
createBatch("data_create_role", insertStmt, element);
}
if (element.getData_day_act_register() > 0) {
return "data_day_act_register";
createBatch("data_day_act_register", insertStmt, element);
}
if (element.getData_day_register_create() > 0) {
return "data_day_register_create";
createBatch("data_day_register_create", insertStmt, element);
}
if (element.getData_active_day_dau() > 0) {
return "data_active_day_dau";
createBatch("data_active_day_dau", insertStmt, element);
}
if (element.getData_sdk_activation() > 0) {
return "data_sdk_activition";
createBatch("data_sdk_activition", insertStmt, element);
}
if (element.getData_sdk_register() > 0) {
return "data_sdk_register";
createBatch("data_sdk_register", insertStmt, element);
}
if (element.getDateRemind() > 0) {
return "retention_day" + element.getDateRemind();
createBatch("retention_day" + element.getDateRemind(), insertStmt, element);
}
if (element.getData_ouid_register() > 0) {
return "ouid_role";
createBatch("ouid_role", insertStmt, element);
}
return null;
}
......
......@@ -39,6 +39,7 @@ 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());
if (jsonObject == null) return;
String ouid = jsonObject.getString("ouid");
if (TextUtils.isEmpty(ouid)) {
return;
......
......@@ -11,6 +11,7 @@ import org.apache.http.util.TextUtils;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetrics, Object, TimeWindow> {
......@@ -53,9 +54,9 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
" values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ");
for (ActivationMetrics value : elements) {
Long days = value.getDateRemind() + 1;
if ( days == 2 || days == 3 || days == 7 || days == 15 || days == 30){
if (days == 2 || days == 3 || days == 7 || days == 15 || days == 30) {
String column = "data_active_" + days + "day";
insertStmt.setString(1,value.getData_date().toString());
insertStmt.setString(1, value.getData_date().toString());
insertStmt.setString(2, value.getData_hour().toString());
insertStmt.setString(3, value.getTime_zone());
insertStmt.setString(4, value.getRegion());
......@@ -155,26 +156,7 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
" (`uid`,`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,`metric_name`,`metric_value`)" +
" values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
for (ActivationMetrics element : elements) {
String name = dealMetrics(element);
if (!TextUtils.isEmpty(name)) {
insertStmt.setString(1, element.getUid());
insertStmt.setLong(2, element.getData_date());
insertStmt.setLong(3, element.getData_hour());
insertStmt.setString(4, element.getTime_zone());
insertStmt.setString(5, element.getRegion());
insertStmt.setString(6, element.getPlatform());
insertStmt.setString(7, element.getGame_id());
insertStmt.setString(8, element.getMedia_source());
insertStmt.setString(9, element.getCampaign_id());
insertStmt.setString(10, element.getAd_set_id());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getChannel());
insertStmt.setLong(13, element.getData_type());
insertStmt.setString(14, name);
insertStmt.setLong(15, 1l);
insertStmt.addBatch();
}
dealMetrics(element, insertStmt);
}
insertStmt.executeBatch(); //批量执行
conn.commit();
......@@ -184,50 +166,71 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
}
}
private String dealMetrics(ActivationMetrics element) {
private void createBatch(String name, PreparedStatement insertStmt, ActivationMetrics element) throws SQLException {
if (!TextUtils.isEmpty(name)) {
insertStmt.setString(1, element.getUid());
insertStmt.setLong(2, element.getData_date());
insertStmt.setLong(3, element.getData_hour());
insertStmt.setString(4, element.getTime_zone());
insertStmt.setString(5, element.getRegion());
insertStmt.setString(6, element.getPlatform());
insertStmt.setString(7, element.getGame_id());
insertStmt.setString(8, element.getMedia_source());
insertStmt.setString(9, element.getCampaign_id());
insertStmt.setString(10, element.getAd_set_id());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getChannel());
insertStmt.setLong(13, element.getData_type());
insertStmt.setString(14, name);
insertStmt.setLong(15, 1l);
insertStmt.addBatch();
}
}
private String dealMetrics(ActivationMetrics element, PreparedStatement insertStmt) throws SQLException {
if (element.getData_activation() > 0) {
return "data_activation";
createBatch("data_activation", insertStmt, element);
}
if (element.getData_register() > 0) {
return "data_register";
createBatch("data_register", insertStmt, element);
}
if (element.getData_device_register() > 0) {
return "data_device_register";
createBatch("data_device_register", insertStmt, element);
}
if (element.getData_create_role() > 0) {
return "data_create_role";
createBatch("data_create_role", insertStmt, element);
}
if (element.getData_day_act_register() > 0) {
return "data_day_act_register";
createBatch("data_day_act_register", insertStmt, element);
}
if (element.getData_day_register_create() > 0) {
return "data_day_register_create";
createBatch("data_day_register_create", insertStmt, element);
}
if (element.getData_active_day_dau() > 0) {
return "data_active_day_dau";
createBatch("data_active_day_dau", insertStmt, element);
}
if (element.getData_sdk_activation() > 0) {
return "data_sdk_activition";
createBatch("data_sdk_activition", insertStmt, element);
}
if (element.getData_sdk_register() > 0) {
return "data_sdk_register";
createBatch("data_sdk_register", insertStmt, element);
}
if (element.getDateRemind() > 0) {
return "retention_day" + element.getDateRemind();
createBatch("retention_day" + element.getDateRemind(), insertStmt, element);
}
if (element.getData_ouid_register() > 0) {
return "ouid_role";
createBatch("ouid_role", insertStmt, element);
}
return null;
}
......
......@@ -50,8 +50,10 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
} else if (PlatformEnum.PC.getName().equals(eventData.getPlatform())){
platform_id = 3;
}
Optional<Game> gameByPrimary = gameMappingDao.findGameByPrimary(eventData.getGame_id(),platform_id);
gameByPrimary.ifPresent(game -> eventData.setRate(game.getRate()));
if (!appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
Optional<Game> gameByPrimary = gameMappingDao.findGameByPrimary(eventData.getGame_id(), platform_id);
gameByPrimary.ifPresent(game -> eventData.setRate(game.getRate()));
}
orderDao.insertOrder(eventData);
} else if (Constants.TYPE_SINK_OUID_ORDER.equals(value.getEventType())){
OrderData eventData = value.getOrderData();
......@@ -63,8 +65,10 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
} else if (PlatformEnum.PC.getName().equals(eventData.getPlatform())){
platform_id = 3;
}
Optional<Game> gameByPrimary = gameMappingDao.findGameByPrimary(eventData.getGame_id(),platform_id);
gameByPrimary.ifPresent(game -> eventData.setRate(game.getRate()));
if (!appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
Optional<Game> gameByPrimary = gameMappingDao.findGameByPrimary(eventData.getGame_id(), platform_id);
gameByPrimary.ifPresent(game -> eventData.setRate(game.getRate()));
}
orderDao.insertOuidOrder(eventData);
}
}
......
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