Commit 7a7d70c3 authored by sheteng's avatar sheteng

加入区服数据

parent 7abb7c42
......@@ -3,13 +3,17 @@ 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;
......@@ -77,7 +81,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(1629129773000L);
// dataConsumer.setStartFromTimestamp(1629821205000L);
Properties props2 = new Properties();
props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers());
......@@ -110,14 +114,7 @@ public class AppsflyerAnalyze {
.uid("role-login-event")
.name("role-login-event");
//日活
DataStream<ActivationMetrics> dauMetricStream = adEventDataStream
.filter(EventTypeEnum::roleLoginEvent)
.flatMap(new DauProcess(appConfig))
.uid("dau-event")
.name("dau-event");
activitionMetricStream.union(loginRoleMetricStream, createRoleMetricStream,dauMetricStream)
activitionMetricStream.union(loginRoleMetricStream, createRoleMetricStream)
.flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink)
.uid("sink-metric-event")
......@@ -130,6 +127,27 @@ public class AppsflyerAnalyze {
.uid("order-event")
.name("order-event");
//区服 角色注册,激活
adEventDataStream
.filter(EventTypeEnum::roleEvent)
.flatMap(new ServerActivationProcessor(appConfig))
.flatMap(new ServerMtricToStringProcessor())
.addSink(kafkaSink)
.uid("activation-event")
.name("activation-event");
adEventDataStream
.filter(EventTypeEnum::notHeartBeatEvent)
.flatMap(new EventMectrisProcessor(appConfig))
.keyBy(EventMetrics::getKeys)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("times")
.flatMap(new EventMetricsToStringProcessor())
.addSink(kafkaSink)
.uid("aggregate-event")
.name("aggregate-event");
sse.execute("appsflyer-analyze");
}
......
......@@ -3,13 +3,22 @@ package com.fshark.overseas.advert;
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.*;
import com.fshark.overseas.advert.processor.server.ServerActivationAggregateFunction;
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.PurchaseSink;
import com.fshark.overseas.advert.sink.RemainSink;
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;
......@@ -75,7 +84,7 @@ public class AppsflyerMetricStore {
}
}).uid("parse-event").name("parse-event");
// activationMetricsDataStream.print();
activationMetricsDataStream.print();
//角色注册激活预聚合
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
......@@ -115,6 +124,55 @@ public class AppsflyerMetricStore {
.uid("order-sink-event")
.name("order-sink-event");
//角色注册激活预聚合
DataStream<ServerActivationMetrics> activationMetricsList = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) throws Exception {
if (value.isServer()) {
out.collect(value.getServerActivationMetrics());
}
}
})
.keyBy(ServerActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.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
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) throws Exception {
if (value.isServer()) {
out.collect(value.getServerActivationMetrics());
}
}
})
.flatMap(new ServerRemainProcessor(appConfig))
.keyBy(ServerRemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("user_num")
.uid("server-remain-event")
.name("server-remain-event");
remainMetrics.addSink(new ServerRemindSink(appConfig))
.uid("server-remindsink-event")
.name("server-remindsink-event");
/**
* 时间写入
*/
activationMetricsDataStream.filter(SinkMetricEvent::isEvent)
.addSink(new EventTimesSink(appConfig))
.uid("event-sink-event")
.name("event-sink-event");
sse.execute("AppsflyerMetricStore-analyze");
}
}
package com.fshark.overseas.advert.entity;
import lombok.Builder;
import lombok.Data;
@Builder
@Data
public class ServerRoleAccount {
private String did;
private String uid;
private String gameId;
private String platform;
private String mediaSource;
private String region;
private String time_zone;
private String campaign_id;
private String ad_set_id;
private String ad_id;
private String register_server_id;
private String role_id;
private String event_time;
}
package com.fshark.overseas.advert.entity.dao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import org.jdbi.v3.sqlobject.customizer.BindBean;
import org.jdbi.v3.sqlobject.statement.SqlUpdate;
public interface EventInfoDao {
@SqlUpdate("insert into appsflyer_event_times (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`event_name`,`event_date`,`times`) " +
"values (:dataDate,:region,:platform,:gameId,:mediaSource,:campaignId,:event_name,:event_date,:times)")
void insertEventTimes(@BindBean EventMetrics value);
@SqlUpdate("insert into appsflyer_event_info (`region`, `uid`, `did`, `platform`, `game_id`, `media_source`, `campaign_id`,`campaign_name`, `ad_set_id`,`ad_set_name`, `ad_id`,`ad_name`, `event_name`, `event_time`, `event_value`,`channel`,`time_zone`) values \n" +
"(:region, :uid, :did, :platform, :game_id, :media_source, :campaign_id,:campaign, :ad_set_id ,:ad_set, :ad_id ,:ad, :event, :event_time, :event_value,:channel,:time_zone)")
Boolean insertEventInfo(@BindBean AppsFlyerEvent event);
}
......@@ -4,6 +4,7 @@ package com.fshark.overseas.advert.entity.dao;
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 org.jdbi.v3.sqlobject.customizer.Bind;
import org.jdbi.v3.sqlobject.customizer.BindBean;
import org.jdbi.v3.sqlobject.statement.SqlUpdate;
......@@ -29,5 +30,15 @@ public interface MetricsDao {
void insertRegisterNum(@BindBean ActivationMetrics event);
/**
* 更新角色指标表的注册创角等
* @param event
*/
@SqlUpdate("insert into appsflyer_metrics_server" +
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`server_id`,`data_register`,`data_create_role`,`data_day_register_create`,`data_active_day_dau`)" +
" values (:data_date,:data_hour,:time_zone,:region,:platform,:game_id,:media_source,:campaign_id,:ad_set_id,:ad_id,:channel,:server_id,:data_register,:data_create_role,:data_day_register_create,:data_active_day_dau) ")
void updateServerActivationMetrics(@BindBean ServerActivationMetrics event);
}
......@@ -13,6 +13,12 @@ public interface OrderDao {
" :pay_time, :revenue,:currency,:orderno,:reg_time,:activation_time,:time_zone,:channel,:data_type)")
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, " +
" pay_time, money, currency, order_no, time_zone ,channel,server_reg_date,server_reg_hour,reg_server_id)" +
" values(:region, :did, :uid, :platform, :game_id, :media_source, :campaign_id, :ad_set_id , :ad_id," +
" :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_order_accumulative" +
// " (`ds`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,`%s`" +
// " values()")
......
package com.fshark.overseas.advert.entity.dao;
import com.fshark.overseas.advert.entity.ServerRoleAccount;
import com.fshark.overseas.advert.entity.mapper.ServerRoleAccountMapper;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsFlyerWithAccountEvent;
import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
import org.jdbi.v3.sqlobject.customizer.Bind;
import org.jdbi.v3.sqlobject.customizer.BindBean;
import org.jdbi.v3.sqlobject.statement.SqlQuery;
import org.jdbi.v3.sqlobject.statement.SqlUpdate;
import java.util.Optional;
public interface ServerRoleDao {
@SqlQuery("SELECT * FROM appsflyer_server_role WHERE game_id=:game_id AND uid=:uid AND platform =:platform AND register_server_id = :serverid AND event_type =:event_type order by event_time desc limit 1")
@RegisterRowMapper(ServerRoleAccountMapper.class)
Optional<ServerRoleAccount> findByPrimary(@BindBean AppsFlyerEvent event, @Bind("serverid") String serverid, @Bind("event_type") String event_type);
@SqlUpdate("insert into appsflyer_server_role (region, did, uid, platform, game_id, media_source, campaign_id, ad_set_id, ad_id ,time_zone,register_server_id,role_id,event_type,event_time)" +
" values (:region, :did, :uid, :platform, :game_id, :media_source, :campaign_id, :ad_set_id , :ad_id ,:time_zone, :register_server_id ,:role_id,'activation',:reg_event_time)")
void insertServerRole(@BindBean AppsFlyerWithAccountEvent event);
@SqlUpdate("insert into appsflyer_server_role (region, did, uid, platform, game_id, media_source, campaign_id, ad_set_id, ad_id ,time_zone,register_server_id,role_id,event_type,event_time)" +
" values (:region, :did, :uid, :platform, :game_id, :media_source, :campaign_id, :ad_set_id , :ad_id ,:time_zone, :register_server_id ,:role_id,'rolelogin',:reg_event_time)")
void inserttServerLogin(@BindBean AppsFlyerWithAccountEvent event);
}
package com.fshark.overseas.advert.entity.mapper;
import com.fshark.overseas.advert.entity.ServerRoleAccount;
import org.jdbi.v3.core.mapper.RowMapper;
import org.jdbi.v3.core.statement.StatementContext;
import java.sql.ResultSet;
import java.sql.SQLException;
public class ServerRoleAccountMapper implements RowMapper<ServerRoleAccount> {
@Override
public ServerRoleAccount map(ResultSet rs, StatementContext ctx) throws SQLException {
return ServerRoleAccount.builder()
.uid(rs.getString("uid"))
.did(rs.getString("did"))
.gameId(rs.getString("game_id"))
.platform(rs.getString("platform"))
.time_zone(rs.getString("time_zone"))
.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"))
.register_server_id(rs.getString("register_server_id"))
.role_id(rs.getString("role_id"))
.event_time(rs.getString("event_time"))
.build();
}
}
......@@ -7,20 +7,12 @@ import lombok.EqualsAndHashCode;
@Data
@EqualsAndHashCode(callSuper = true)
public class AppsFlyerWithAccountEvent extends AppsFlyerEvent {
/**
* 服务注册时间
*/
private String register_server_time;
/**
* 区服
*/
private String register_server_id;
/**
* 上次区服登录时间
*/
private String last_role_login_time;
private String reg_event_time;
/**
* 数据类型 1.新增 2.再营销 3.预注册
......
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;
......@@ -24,6 +26,7 @@ public enum EventTypeEnum {
PURCHASE("af_purchase"),
RE_ENGAGEMENT("re-engagement"),
RE_ATTRIBUTION("re-attribution"),
HEARTBEAT("heartbeat"),
LV_2("lv 2");
private String name;
......@@ -41,22 +44,32 @@ public enum EventTypeEnum {
}
public static boolean roleLoginEvent(AppsFlyerEvent event) {
return ROLE_LOGIN.getName().equals(event.getEvent())
return (ROLE_LOGIN.getName().equals(event.getEvent())
|| FAST_LOGIN.getName().equals(event.getEvent())
|| GAME_LOGIN.getName().equals(event.getEvent())
|| SDK_LOGIN.getName().equals(event.getEvent());
|| SDK_LOGIN.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean createRoleEvent(AppsFlyerEvent event) {
return CREATE_ROLE.getName().equals(event.getEvent());
return CREATE_ROLE.getName().equals(event.getEvent()) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean roleEvent(AppsFlyerEvent event) {
return (CREATE_ROLE.getName().equals(event.getEvent()) || ROLE_LOGIN.getName().equals(event.getEvent())) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean installEvent(AppsFlyerEvent event) {
return ACTIVATION.getName().equals(event.getEvent()) || RE_ENGAGEMENT.getName().equals(event.getEvent()) || RE_ATTRIBUTION.getName().equals(event.getEvent());
return (ACTIVATION.getName().equals(event.getEvent()) || RE_ENGAGEMENT.getName().equals(event.getEvent()) || RE_ATTRIBUTION.getName().equals(event.getEvent()))&& !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE);
}
public static boolean orderEvent(AppsFlyerEvent event) {
return PURCHASE.getName().equals(event.getEvent());
return PURCHASE.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);
}
}
......@@ -145,6 +145,9 @@ public class OrderData implements Serializable {
private String reg_server_time;
private String server_reg_date;
private String server_reg_hour;
private int data_type;
......
package com.fshark.overseas.advert.modle;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.fshark.overseas.advert.util.Constants;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class SinkMetricEvent {
private String eventType;
private ActivationMetrics eventData;
private OrderData orderData;
private EventMetrics eventMetrics;
private ServerActivationMetrics serverActivationMetrics;
public boolean isMetricEvent() {
return Constants.TYPE_SINK_EVENT.equals(eventType);
}
public boolean isOrderEvent() {
return Constants.TYPE_SINK_ORDER.equals(eventType);
}
public boolean isEvent() {
return Constants.TYPE_SINK_EVENT.equals(eventType);
}
public boolean isServer() {
return Constants.TYPE_SINK_SERVER.equals(eventType);
}
}
package com.fshark.overseas.advert.modle.metrics;
import com.google.gson.Gson;
import lombok.Data;
import java.io.Serializable;
@Data
public class BaseServerMetrics implements Serializable {
/**
* 日期 20190228
*/
private Long data_date;
/**
* 小时 13
*/
private Integer data_hour;
/**
* 时区 +0900
*/
private String time_zone;
/**
* 媒体平台 googleadwords_int
*/
private String media_source;
/**
* 平台 Android|IOS
*/
private String platform;
/**
* 游戏ID bundle_id, com.edgames.new19gp
*/
private String game_id;
/**
* 广告系列ID 23843308705120641
*/
private String campaign_id;
/**
* 广告组|计划ID 23843308705230641
*/
private String ad_set_id;
/**
* 广告ID 23843308708070641
*/
private String ad_id;
/**
* 韩国KR, 香港 HK, 台湾 TW, 其它:-1,或者其它值
*/
private String region;
/**
* 渠道
*/
private String channel;
/**
* 区服id
*/
private String server_id;
/**
* 数据类型 1.新增 2.再营销 3.预注册
*/
private int data_type = 1;
public String getKey() {
return new Gson().toJson(this);
}
public static BaseServerMetrics setKey(String keys) {
return new Gson().fromJson(keys,BaseServerMetrics.class);
}
}
package com.fshark.overseas.advert.modle.metrics;
import lombok.Data;
@Data
public class EventMetrics {
public EventMetrics(){}
/**
* 日期 20190228
*/
private Long dataDate = -1L;
/**
* 时区 +0900
*/
private String timeZone;
/**
* 韩国KR, 香港 HK, 台湾 TW, 其它:-1,或者其它值
*/
private String region;
/**
* 媒体平台 googleadwords_int
*/
private String mediaSource;
/**
* 平台 Android|IOS
*/
private String platform;
/**
* 游戏ID bundle_id, com.edgames.new19gp
*/
private String gameId;
/**
* 广告系列ID 23843308705120641
*/
private String campaignId;
/**
* eventname
*/
private String event_name;
private Long event_date;
/**
* 发生次数
*/
private Long times;
private String keys;
public String getKeys() {
return String.format("%s|%s|%s|%s|%s|%s|%s|%s",dataDate,timeZone,region,mediaSource,platform,gameId,campaignId,event_name);
}
}
package com.fshark.overseas.advert.modle.metrics;
import lombok.Data;
import lombok.EqualsAndHashCode;
@Data
@EqualsAndHashCode(callSuper = true)
public class ServerActivationMetrics extends BaseServerMetrics {
/**
* 激活数 按唯一设备去重
*/
// private Integer dataActivation;
/**
* 区服注册数 按唯一用户去重
*/
private Long data_register = 0L;
/**
* 当日激活 并且当日注册数
*/
// private Integer dataDayActRegister;
/**
* 设备注册数
*/
// private Integer dataDeviceRegister;
/**
* 区服创角数
*/
private Long data_create_role = 0L;
/**
* 当日注册 并且当日创角数
*/
private Long data_day_register_create = 0L;
/**
* 当日登录用户,排除当日注册用户
*/
private Long data_active_day_dau = 0L;
/**
* 留存天数
*/
private Long dateRemind = 0L;
}
package com.fshark.overseas.advert.modle.metrics;
import lombok.Data;
import lombok.EqualsAndHashCode;
@Data
@EqualsAndHashCode(callSuper = true)
public class ServerRemainMetrics extends BaseServerMetrics {
/**
* 留存时间
*/
private Long days;
/**
* 留存人数
*/
private Long user_num;
public String getRemind_key() {
return String.format("%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s",getData_date(),getTime_zone(),getRegion(),getMedia_source(),getPlatform(),getGame_id(),getCampaign_id(),getAd_set_id(),getAd_id(),getChannel(),getServer_id());
}
}
......@@ -17,7 +17,7 @@ public class ActivitionMetricToStringProcess extends RichFlatMapFunction<Activat
Gson gson = new Gson();
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_EVENT);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_METRIC);
out.collect(gson.toJson(sinkMetricEvent));
}
}
......@@ -44,23 +44,26 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
LocalDateTime activition;
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
if (!account.isPresent()) { // 设备未激活
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
// activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
// activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L);
activition = value.getEvent_time();
} else {
GameAccount gameAccount = account.get();
String activationTime = gameAccount.getEvent_time();
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
// activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
// activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
activition = DateUtils.formatAsDate(activationTime);
}
activationMetrics.setData_create_role(1L);
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) {
activationMetrics.setData_register(1L);
activationMetrics.setData_create_role(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_register_create(1L);//当日注册并创角
if (DateUtils.daysBetween(activition, value.getEvent_time()) == 0) { //当日激活并且注册
......
......@@ -4,27 +4,28 @@ package com.fshark.overseas.advert.processor;
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.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.modle.AppsflyerPurchase;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.util.*;
import org.apache.commons.lang3.StringUtils;
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 java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
import java.util.Optional;
public class DauProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, EventMetrics> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
public DauProcess(AppConfig appConfig) {
public EventMectrisProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
}
......@@ -33,68 +34,60 @@ public class DauProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
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 {
ActivationMetrics activationMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
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);
} else {
GameAccount gameAccount = account.get();
activationTime = gameAccount.getEvent_time();
public void flatMap(AppsFlyerEvent value, Collector<EventMetrics> out) throws Exception {
if (value == null || value.getEvent_time() == null) {
return;
}
CommonUtils.setType(activationMetrics);
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_time();
if (!StringUtils.isBlank(last_login_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) {
activationMetrics.setData_active_day_dau(1L);
}
if (EventTypeEnum.PURCHASE.getName().equals(value.getEvent())) {
try {
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getServerid() == null) { //不是我们上报的格式
return;
}
if ("ios".equals(value.getPlatform().toLowerCase())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
} catch (Exception e) {
return;
}
}
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
EventMetrics metrics = new EventMetrics();
metrics.setTimeZone(value.getTime_zone());
metrics.setRegion(value.getRegion());
metrics.setPlatform(value.getPlatform());
metrics.setGameId(value.getGame_id());
metrics.setCampaignId(value.getCampaign_id());
metrics.setEvent_name(value.getEvent());
metrics.setMediaSource(value.getMedia_source());
if (account.isPresent()) {
GameAccount gameAccount = account.get();
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(gameAccount.getEvent_time())));
} else if (EventTypeEnum.ACTIVATION.getName().equals(value.getEvent())) {
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
metrics.setEvent_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
metrics.setTimes(1L);
out.collect(metrics);
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setData_active_day_dau(1L);
Optional<GameAccount> byAccountPrimary = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (byAccountPrimary.isPresent()) {
GameAccount registerAccount = byAccountPrimary.get();
if (null != registerAccount.getEvent_time()) {
metrics.setDataDate(Long.valueOf(DateUtils.formatAsYMD(registerAccount.getEvent_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.setEvent_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
metrics.setData_type(1);
if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2);
}
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据
metrics.setData_type(3);
}
metrics.setTimes(1L);
out.collect(metrics);
}
@Override
......@@ -103,6 +96,6 @@ public class DauProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;
public class EventMetricsToStringProcessor extends RichFlatMapFunction<EventMetrics, String> {
@Override
public void flatMap(EventMetrics value, Collector<String> out) throws Exception {
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_EVENT);
sinkMetricEvent.setEventMetrics(value);
Gson gson = new Gson();
out.collect(gson.toJson(sinkMetricEvent));
}
}
......@@ -14,7 +14,7 @@ public class MetricJsonToObProcessor extends RichFlatMapFunction<SinkMetricEvent
@Override
public void flatMap(SinkMetricEvent value, Collector<ActivationMetrics> out) throws Exception {
if (Constants.TYPE_SINK_EVENT.equals(value.getEventType()) && value.getEventData() != null) {
if (Constants.TYPE_SINK_METRIC.equals(value.getEventType()) && value.getEventData() != null) {
out.collect(value.getEventData());
}
}
......
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;
import com.fshark.overseas.advert.modle.OrderData;
......@@ -21,6 +23,7 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
private ServerRoleDao serverRoleDao;
public PurchaseToOrderDataProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
......@@ -31,38 +34,51 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
serverRoleDao = jdbiContext.getJdbi().onDemand(ServerRoleDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<String> out) throws Exception {
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null) { //不是我们上报的格式
return;
}
try {
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null) { //不是我们上报的格式
return;
}
if ("ios".equalsIgnoreCase(value.getPlatform())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
if ("ios".equalsIgnoreCase(value.getPlatform())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
OrderData orderData = new OrderData(value);
Optional<GameAccount> accountActivition = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (accountActivition.isPresent()) {
GameAccount gameAccount = accountActivition.get();
orderData.setActivation_time(gameAccount.getEvent_time());
orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())),orderData.getMedia_source(),orderData.getData_type()));
}
OrderData orderData = new OrderData(value);
Optional<GameAccount> accountActivition = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (accountActivition.isPresent()) {
GameAccount gameAccount = accountActivition.get();
orderData.setActivation_time(gameAccount.getEvent_time());
orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())), orderData.getMedia_source(), orderData.getData_type()));
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (accountRegister.isPresent()) {
GameAccount gameAccount = accountRegister.get();
orderData.setReg_time(gameAccount.getEvent_time());
}
gameAccountDao.insertPauchase(value);
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (accountRegister.isPresent()) {
GameAccount gameAccount = accountRegister.get();
orderData.setReg_time(gameAccount.getEvent_time());
}
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_ORDER) ;
sinkMetricEvent.setOrderData(orderData);
out.collect(new Gson().toJson(sinkMetricEvent));
Optional<ServerRoleAccount> serverAccount = serverRoleDao.findByPrimary(value, appsflyerPurchase.getServerid(), Constants.TYPE_ACTIVIYION);
if (serverAccount.isPresent()) {
String event_time = serverAccount.get().getEvent_time();
orderData.setReg_server_id(appsflyerPurchase.getServerid());
orderData.setReg_server_time(event_time);
}
gameAccountDao.insertPauchase(value);
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_ORDER);
sinkMetricEvent.setOrderData(orderData);
out.collect(new Gson().toJson(sinkMetricEvent));
} catch (Exception e) {
return;
}
}
......
......@@ -17,7 +17,7 @@ public class RemainProcess extends RichFlatMapFunction<SinkMetricEvent, RemainMe
@Override
public void flatMap(SinkMetricEvent value, Collector<RemainMetrics> out) throws Exception {
if (Constants.TYPE_SINK_EVENT.equals(value.getEventType())) {
if (Constants.TYPE_SINK_METRIC.equals(value.getEventType())) {
ActivationMetrics eventData = value.getEventData();
if (eventData.getDateRemind() == 0L) {
return;
......
......@@ -8,6 +8,7 @@ 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.commons.cli.CommandLine;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
......@@ -39,10 +40,15 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
ActivationMetrics activationMetrics = new ActivationMetrics();
ActivationMetrics retentMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics);
copyBaseMetrics(value, retentMetrics);
activationMetrics.setUid(value.getUid());
retentMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
String activationTime = event_time;
if (!account.isPresent()) { // 设备未激活
gameAccountDao.insertActivition(value, event_time);
......@@ -61,9 +67,10 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
activationTime = accountRegister.get().getEvent_time();
}
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
// CommonUtils.setType(activationMetrics);
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();
......@@ -71,18 +78,21 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween); //留存天数
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween); //留存天数
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
gameAccountDao.insertRoleLogin(value);
out.collect(activationMetrics);
out.collect(retentMetrics);
}
......
package com.fshark.overseas.advert.processor;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.entity.ServerRoleAccount;
import com.fshark.overseas.advert.entity.dao.ServerRoleDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsFlyerWithAccountEvent;
import com.fshark.overseas.advert.modle.metrics.BaseServerMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
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 java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
import java.util.Optional;
/**
* 转化成激活注册指标数据
*/
public class ServerActivationProcessor extends RichFlatMapFunction<AppsFlyerEvent, ServerActivationMetrics> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private ServerRoleDao serverRoleDao;
public ServerActivationProcessor(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());
serverRoleDao = jdbiContext.getJdbi().onDemand(ServerRoleDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<ServerActivationMetrics> out) throws Exception {
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
String serverid = jsonObject.getString("serverid");
String roleId = jsonObject.getString("roleid");
if (serverid == null) {
return;
}
Optional<ServerRoleAccount> serverAccount = serverRoleDao.findByPrimary(value, serverid, Constants.TYPE_ACTIVIYION);
AppsFlyerWithAccountEvent appsFlyerWithAccountEvent = copyEvent(value);
ServerActivationMetrics activationMetrics = new ServerActivationMetrics();
copyBaseMetrics(appsFlyerWithAccountEvent, activationMetrics);
activationMetrics.setServer_id(serverid);
appsFlyerWithAccountEvent.setRegister_server_id(serverid);
String time = DateUtils.formatAsYMDHMS(value.getEvent_time());
if (serverAccount.isPresent()) {
ServerRoleAccount serverRole = serverAccount.get();
String register_server_time = serverRole.getEvent_time();
Optional<ServerRoleAccount> loginRole = serverRoleDao.findByPrimary(value, serverid, Constants.TYPE_ROLELOGIN);
if (loginRole.isPresent()) {
String last_role_login_time = loginRole.get().getEvent_time();
activationMetrics.setData_date(Long.valueOf(DateUtils.formatAsYMD(register_server_time)));
activationMetrics.setData_hour(Integer.valueOf(DateUtils.formatAsHH(register_server_time)));
//当日登录排除当日注册用户 dau
LocalDateTime last_login_date = DateUtils.formatAsDate(last_role_login_time);
if (DateUtils.daysBetween(last_login_date, value.getEvent_time()) > 0) {
long daysBetween = DateUtils.daysBetween(DateUtils.formatAsDate(register_server_time), value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween);
out.collect(activationMetrics);
ServerActivationMetrics activationMetrics1 = new ServerActivationMetrics();
copyBaseMetrics(appsFlyerWithAccountEvent, activationMetrics1);
activationMetrics1.setData_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
activationMetrics1.setData_hour(Integer.valueOf(DateUtils.formatAsHH(value.getEvent_time())));
activationMetrics1.setData_active_day_dau(1L);
activationMetrics1.setServer_id(serverid);
out.collect(activationMetrics1);
appsFlyerWithAccountEvent.setReg_event_time(time);
}
}
}
} else {
activationMetrics.setData_date(Long.valueOf(DateUtils.formatAsYMD(time)));
activationMetrics.setData_hour(Integer.valueOf(DateUtils.formatAsHH(time)));
activationMetrics.setData_register(1L);
activationMetrics.setData_create_role(1L);
activationMetrics.setData_day_register_create(1L);
appsFlyerWithAccountEvent.setRole_id(roleId);
appsFlyerWithAccountEvent.setReg_event_time(time);
serverRoleDao.insertServerRole(appsFlyerWithAccountEvent);
out.collect(activationMetrics);
}
serverRoleDao.inserttServerLogin(appsFlyerWithAccountEvent);
}
private void copyBaseMetrics(AppsFlyerWithAccountEvent value, BaseServerMetrics metrics) {
metrics.setTime_zone(value.getTime_zone());
metrics.setPlatform(value.getPlatform());
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.setData_type(value.getData_type());
metrics.setChannel(value.getChannel());
metrics.setMedia_source(value.getMedia_source());
value.setData_type(1);
if (value.getIs_retargeting()) { //再营销数据
value.setData_type(2);
}
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据
value.setData_type(3);
}
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
private AppsFlyerWithAccountEvent copyEvent(AppsFlyerEvent value) {
Gson gson = new Gson();
return gson.fromJson(gson.toJson(value), AppsFlyerWithAccountEvent.class);
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;
public class ServerMtricToStringProcessor extends RichFlatMapFunction<ServerActivationMetrics, String> {
@Override
public void flatMap(ServerActivationMetrics value, Collector<String> out) throws Exception {
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_SERVER);
sinkMetricEvent.setServerActivationMetrics(value);
Gson gson = new Gson();
out.collect(gson.toJson(sinkMetricEvent));
}
}
......@@ -40,7 +40,7 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<SinkMetricEvent
@Override
public void process(Context context, Iterable<SinkMetricEvent> elements, Collector<Object> out) throws Exception {
for (SinkMetricEvent sinkMetricEvent : elements) {
if (Constants.TYPE_SINK_EVENT.equals(sinkMetricEvent.getEventType())){
if (Constants.TYPE_SINK_METRIC.equals(sinkMetricEvent.getEventType())){
ActivationMetrics element = sinkMetricEvent.getEventData();
if (element.getData_activation() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_activation");
......
package com.fshark.overseas.advert.processor.server;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple4;
public class ServerActivationAggregateFunction implements AggregateFunction<ServerActivationMetrics, Tuple4<Long,Long,Long,Long>, Tuple4<Long,Long,Long,Long>> {
@Override
public Tuple4<Long, Long, Long, Long> createAccumulator() {
Tuple4<Long, Long, Long, Long> objectObjectObjectObjectTuple4 = new Tuple4<>(0L,0L,0L,0L);
return objectObjectObjectObjectTuple4;
}
@Override
public Tuple4<Long, Long, Long, Long> add(ServerActivationMetrics value, Tuple4<Long, Long, Long, Long> accumulator) {
return new Tuple4<Long, Long, Long, Long>(accumulator.f0 + value.getData_register(),accumulator.f1 + value.getData_create_role(),accumulator.f2 + value.getData_day_register_create(),accumulator.f3 + value.getData_active_day_dau());
}
@Override
public Tuple4<Long, Long, Long, Long> getResult(Tuple4<Long, Long, Long, Long> accumulator) {
return accumulator;
}
@Override
public Tuple4<Long, Long, Long, Long> merge(Tuple4<Long, Long, Long, Long> a, Tuple4<Long, Long, Long, Long> b) {
return new Tuple4<Long, Long, Long, Long>(a.f0 + b.f0,a.f1 + b.f1,a.f2 + b.f2,a.f3 + b.f3);
}
}
package com.fshark.overseas.advert.processor.server;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseServerMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.google.gson.Gson;
import org.apache.flink.api.java.tuple.Tuple4;
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 ServerActivationResultWindows implements WindowFunction<Tuple4<Long, Long, Long, Long>, ServerActivationMetrics, String, TimeWindow> {
@Override
public void apply(String s, TimeWindow window, Iterable<Tuple4<Long, Long, Long, Long>> input, Collector<ServerActivationMetrics> out) throws Exception {
BaseServerMetrics baseMetrics = BaseServerMetrics.setKey(s);
ServerActivationMetrics activationMetrics = copyMetrics(baseMetrics);
Tuple4<Long, Long, Long, Long> next = input.iterator().next();
activationMetrics.setData_register(next.f0);
activationMetrics.setData_create_role(next.f1);
activationMetrics.setData_day_register_create(next.f2);
activationMetrics.setData_active_day_dau(next.f3);
out.collect(activationMetrics);
}
private ServerActivationMetrics copyMetrics(BaseServerMetrics metrics) {
Gson gson = new Gson();
return gson.fromJson(gson.toJson(metrics), ServerActivationMetrics.class);
}
}
package com.fshark.overseas.advert.processor.server;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseServerMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.ServerRemainMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
;
/**
* 留存指标
*/
public class ServerRemainProcessor implements FlatMapFunction<ServerActivationMetrics, ServerRemainMetrics> {
private AppConfig appConfig;
public ServerRemainProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void flatMap(ServerActivationMetrics value, Collector<ServerRemainMetrics> out) throws Exception {
if (value.getDateRemind() == 0L) {
return;
}
ServerRemainMetrics remainMetrics = new ServerRemainMetrics();
remainMetrics.setData_date(value.getData_date());
remainMetrics.setData_hour(value.getData_hour());
copyBaseMetrics(value, remainMetrics);
remainMetrics.setServer_id(value.getServer_id());
remainMetrics.setDays(value.getDateRemind() + 1);
remainMetrics.setUser_num(1L);
remainMetrics.setMedia_source(value.getMedia_source());
out.collect(remainMetrics);
}
private void copyBaseMetrics(ServerActivationMetrics value, BaseServerMetrics metrics) {
metrics.setTime_zone(value.getTime_zone());
metrics.setPlatform(value.getPlatform());
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.setData_type(value.getData_type());
metrics.setChannel(value.getChannel());
}
}
package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.EventInfoDao;
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 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> {
private AppConfig appConfig;
public EventTimesSink(AppConfig appConfig) {
this.appConfig = appConfig;
}
private JdbiContext jdbiContext;
private EventInfoDao eventInfoDao;
@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());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
eventInfoDao = jdbiContext.getJdbi().onDemand(EventInfoDao.class);
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
@Override
public void invoke(SinkMetricEvent value, Context context) throws Exception {
super.invoke(value, context);
if (value.isEvent()){
eventInfoDao.insertEventTimes(value.getEventMetrics());
}
}
}
......@@ -20,7 +20,6 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
private JdbiContext jdbiContext;
private OrderDao orderDao;
private GameAccountDao gameAccountDao;
public PurchaseSink(AppConfig appConfig) {
this.appConfig = appConfig;
......@@ -35,7 +34,6 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
orderDao = jdbiContext.getJdbi().onDemand(OrderDao.class);
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
}
@Override
......@@ -44,6 +42,9 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())){
OrderData eventData = value.getOrderData();
orderDao.insertOrder(eventData);
eventData.setServer_reg_date(DateUtils.formatAsYMD(eventData.getReg_server_time()));
eventData.setServer_reg_hour(DateUtils.formatAsHH(eventData.getReg_server_time()));
orderDao.insertServerOrder(eventData);
}
}
......
package com.fshark.overseas.advert.sink.server;
import com.fshark.overseas.advert.entity.dao.MetricsDao;
import com.fshark.overseas.advert.modle.metrics.ServerActivationMetrics;
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 org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class ServerActivationSink extends RichSinkFunction<ServerActivationMetrics> {
private AppConfig appConfig;
public ServerActivationSink(AppConfig appConfig) {
this.appConfig = appConfig;
}
private JdbiContext jdbiContext;
private MetricsDao metricsDao;
@Override
public void invoke(ServerActivationMetrics value, Context context) throws Exception {
super.invoke(value, context);
if (value.getData_active_day_dau() + value.getData_register() + value.getData_day_register_create() + value.getData_create_role() > 0) {
metricsDao.updateServerActivationMetrics(value);
}
}
@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());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
metricsDao = jdbiContext.getJdbi().onDemand(MetricsDao.class);
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
package com.fshark.overseas.advert.sink.server;
import com.fshark.overseas.advert.entity.dao.MetricsDao;
import com.fshark.overseas.advert.modle.metrics.ServerRemainMetrics;
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 org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class ServerRemindSink extends RichSinkFunction<ServerRemainMetrics> {
private AppConfig appConfig;
public ServerRemindSink(AppConfig appConfig) {
this.appConfig = appConfig;
}
private JdbiContext jdbiContext;
private MetricsDao metricsDao;
@Override
public void invoke(ServerRemainMetrics value, Context context) throws Exception {
super.invoke(value, context);
//短线留存
Long days = value.getDays();
if (days == 2 || days == 3 || days == 7 || days == 15 || days == 30) {
String column = "data_active_" + days + "day";
jdbiContext.getJdbi().withHandle(handle -> {
String sql = String.format("insert into appsflyer_metrics_server" +
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`server_id`,`%s`)" +
" values ('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s') ",
column, value.getData_date(), value.getData_hour(), value.getTime_zone(), value.getRegion(), value.getPlatform(), value.getGame_id(), value.getMedia_source(), value.getCampaign_id()
, value.getAd_set_id(), value.getAd_id(), value.getChannel(), value.getServer_id(), value.getUser_num(), column, column, column
);
handle.createUpdate(sql).execute();
return null;
});
}
}
@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());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
metricsDao = jdbiContext.getJdbi().onDemand(MetricsDao.class);
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
......@@ -77,8 +77,10 @@ public class Constants {
public static final String TYPE_SINK="mysql";
public static final String TYPE_SINK_EVENT="metric";
public static final String TYPE_SINK_METRIC="metric";
public static final String TYPE_SINK_ORDER="order";
public static final String TYPE_SINK_EVENT="event";
public static final String TYPE_SINK_SERVER="server";
......
......@@ -45,6 +45,13 @@ public class DateUtils {
return parse.format(YMD_FORMATTER);
}
public static String formatAsYMD(LocalDateTime ts) {
return ts.format(YMD_FORMATTER);
}
public static String formatAsHH(LocalDateTime ts) {
return ts.format(HH_FORMATTER);
}
public static String formatAsHH(String ts) {
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_FORMATTER);
......
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