Commit f4660a13 authored by sheteng's avatar sheteng

ouid ltv

parent 10a88757
......@@ -90,7 +90,7 @@ public class AppsflyerAnalyze {
.uid("parse-event")
.name("parse-event")
.shuffle();
// adEventDataStream.print();
adEventDataStream.print();
//广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream
.filter(EventTypeEnum::installEvent)
......
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.processor.ParseEventProcessor;
import com.fshark.overseas.advert.sink.EventInfoSink;
import com.fshark.overseas.advert.sink.EventInfoSinkProcess;
......@@ -11,6 +12,7 @@ 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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
......@@ -77,7 +79,7 @@ public class AppsflyerInfoDmi {
.name("parse-event")
.shuffle();
adEventDataStream.timeWindowAll(Time.seconds(30)).process(new EventInfoSinkProcess(appConfig))
adEventDataStream.filter(EventTypeEnum::notHeartBeatEvent).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).process(new EventInfoSinkProcess(appConfig))
.uid("event")
.name("event");
......
......@@ -82,14 +82,15 @@ public class AppsflyerMetricStore {
})
.uid("parse-event")
.name("parse-event");
activationMetricsDataStream.print();
//角色注册
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
.uid("activation-agg-event")
.name("activation-agg-event");
activitionResult.print();
//指标详细记录
activitionResult.timeWindowAll(Time.seconds(1)).process(new StoreMetricProcess(appConfig))
activitionResult.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).process(new StoreMetricProcess(appConfig))
.uid("metric-detail")
.name("metric-detail");
......@@ -125,16 +126,16 @@ public class AppsflyerMetricStore {
}
})
.keyBy(ServerActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.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
......@@ -146,7 +147,7 @@ public class AppsflyerMetricStore {
})
.flatMap(new ServerRemainProcessor(appConfig))
.keyBy(ServerRemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum("user_num")
.uid("server-remain-event")
.name("server-remain-event");
......@@ -158,7 +159,7 @@ public class AppsflyerMetricStore {
* 事件写入
*/
activationMetricsDataStream.filter(SinkMetricEvent::isEvent)
.timeWindowAll(Time.seconds(30))
.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new EventTimesSinkProcess(appConfig))
.uid("event-sink-event")
.name("event-sink-event");
......
......@@ -72,7 +72,7 @@ public class AppsflyerOuidAnalyze {
.filter(new FilterFunction<AppsFlyerEvent>() {
@Override
public boolean filter(AppsFlyerEvent value) throws Exception {
if (value.getMedia_source().contains("rank") || value.getCampaign_id().contains("rank")) {
if (value.getMedia_source().contains("rank") || value.getCampaign_id().contains("rank")){
return false;
}
return true;
......@@ -97,8 +97,8 @@ public class AppsflyerOuidAnalyze {
// ouidMetrics.timeWindowAll(Time.minutes(1));
//订单信息
adEventDataStream.filter(EventTypeEnum::orderEvent).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new OuidOrderDataProcessor(appConfig))
adEventDataStream.filter(EventTypeEnum::orderEvent)
.addSink(new OuidOrderDataProcessor(appConfig))
.uid("order-metrics")
.name("order-metrics");
......
......@@ -24,11 +24,11 @@ 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")
@SqlQuery("SELECT * FROM appsflyer_ouid_role 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`) " +
@SqlUpdate("insert into appsflyer_ouid_role (`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);
......
package com.fshark.overseas.advert.entity.dao;
import com.fshark.overseas.advert.entity.mapper.AdMappingMapper;
import com.fshark.overseas.advert.entity.mapper.MediaCampaignMapper;
import com.fshark.overseas.advert.entity.mapper.MediaMapper;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.mapping.AdMapping;
import com.fshark.overseas.advert.modle.mapping.Media;
import com.fshark.overseas.advert.modle.mapping.MediaCampaign;
import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
import org.jdbi.v3.sqlobject.customizer.Bind;
import org.jdbi.v3.sqlobject.customizer.BindBean;
......@@ -31,16 +35,23 @@ public interface GameMappingDao {
Optional<Media> findMediaByPrimary(@BindBean AppsFlyerEvent value);
@SqlUpdate("insert into media (`af_media_id`) values (:media_source)")
@RegisterRowMapper(MediaMapper.class)
void saveMediaSource(@BindBean AppsFlyerEvent value);
@SqlQuery("select count(1) from adgroup_media_mapping where af_media_id = :media_source and af_group_id = :campaign_id ")
boolean existsMediaCampaign(@BindBean AppsFlyerEvent appsFlyerEvent);
@SqlQuery("select * from adgroup_media_mapping where af_media_id = :media_source and af_group_id = :campaign_id ")
@RegisterRowMapper(MediaCampaignMapper.class)
Optional<MediaCampaign> findMediaCampaign(@BindBean AppsFlyerEvent appsFlyerEvent);
@SqlUpdate("insert into adgroup_media_mapping (`media_id`, `af_media_id`, `af_group_id`, `af_group_name`) values (:id, :media_source, :campaign_id, :campaign)")
@SqlUpdate("insert into adgroup_media_mapping (`media_id`, `af_media_id`, `af_group_id`, `af_group_name`) values (:id, :media_source, :campaign_id, :campaign) on duplicate key update af_group_name = values(af_group_name)")
void saveMediaCampaign(@BindBean AppsFlyerEvent appsFlyerEvent, @BindBean Media media);
@SqlUpdate("insert into ad_mapping (`media_id`, `af_media_id`, `af_ad_group_id`, `af_ad_group_name`, `af_ad_plan_id`, `af_ad_plan_name`, `af_ad_id`, `af_ad_name`) " +
" values (:id, :media_source , :campaign_id, :campaign, :ad_set_id, :ad_set, :ad_id, :ad) on duplicate key update af_ad_group_name = values(af_ad_group_name), af_ad_plan_name = values(af_ad_plan_name), af_ad_name = values(af_ad_name)")
void saveAdMapping(@BindBean AppsFlyerEvent appsFlyerEvent, @BindBean Media media);
@SqlQuery("select * from ad_mapping where af_media_id = :media_source and af_ad_group_id = :campaign_id and af_ad_plan_id = :ad_set_id and af_ad_id = :ad_id")
@RegisterRowMapper(AdMappingMapper.class)
Optional<AdMapping> findAdMapping(@BindBean AppsFlyerEvent value);
}
package com.fshark.overseas.advert.entity.mapper;
import com.fshark.overseas.advert.modle.mapping.AdMapping;
import org.jdbi.v3.core.mapper.RowMapper;
import org.jdbi.v3.core.statement.StatementContext;
import java.sql.ResultSet;
import java.sql.SQLException;
public class AdMappingMapper implements RowMapper<AdMapping> {
@Override
public AdMapping map(ResultSet rs, StatementContext ctx) throws SQLException {
return AdMapping.builder()
.af_ad_group_id(rs.getString("af_ad_group_id"))
.af_ad_group_name(rs.getString("af_ad_group_name"))
.af_ad_plan_id(rs.getString("af_ad_plan_id"))
.af_ad_plan_name(rs.getString("af_ad_plan_name"))
.af_ad_id(rs.getString("af_ad_id"))
.af_ad_name(rs.getString("af_ad_name"))
.build();
}
}
package com.fshark.overseas.advert.entity.mapper;
import com.fshark.overseas.advert.modle.mapping.MediaCampaign;
import org.jdbi.v3.core.mapper.RowMapper;
import org.jdbi.v3.core.statement.StatementContext;
import java.sql.ResultSet;
import java.sql.SQLException;
public class MediaCampaignMapper implements RowMapper<MediaCampaign> {
@Override
public MediaCampaign map(ResultSet rs, StatementContext ctx) throws SQLException {
return MediaCampaign.builder()
.campaign(rs.getString("af_group_id"))
.media_source(rs.getString("af_media_id"))
.af_group_name(rs.getString("af_group_name"))
.build();
}
}
package com.fshark.overseas.advert.modle.mapping;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
@Builder
@Data
@AllArgsConstructor
public class AdMapping implements Serializable {
private String af_ad_group_id;
private String af_ad_group_name;
private String af_ad_plan_id;
private String af_ad_plan_name;
private String af_ad_id;
private String af_ad_name;
}
package com.fshark.overseas.advert.modle.mapping;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@Builder
@Data
@AllArgsConstructor
public class MediaCampaign {
private String media_source;
private String campaign;
private String af_group_name;
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.dao.GameMappingDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.PlatformEnum;
import com.fshark.overseas.advert.modle.mapping.AdMapping;
import com.fshark.overseas.advert.modle.mapping.Game;
import com.fshark.overseas.advert.modle.mapping.Media;
import com.fshark.overseas.advert.modle.mapping.MediaCampaign;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
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.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
......@@ -26,13 +23,6 @@ import java.util.Optional;
*/
public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
private MapStateDescriptor<String, Game> stateDescriptor = new MapStateDescriptor("game_list_state", String.class, Game.class);
private MapStateDescriptor<String, Media> mediasourceStateDescriptor = new MapStateDescriptor("mediasource_list_state", String.class, Media.class);
private MapStateDescriptor<String, MediaCampaign> mediaCampaignStateDescriptor = new MapStateDescriptor("media_campaign_list_state", String.class, MediaCampaign.class);
private MapStateDescriptor<String, AdMapping> adMappingMapStateDescriptor = new MapStateDescriptor("ad_mapping_list_state", String.class, AdMapping.class);
private AppConfig appConfig;
......@@ -40,13 +30,6 @@ public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent,
private GameMappingDao gameMappingDao;
private MapState<String, Game> gameState;
private MapState<String, Media> mediaSourceState;
private MapState<String, MediaCampaign> mediaCampaignState;
private MapState<String, AdMapping> adMappingMapState;
public GameAdMappingProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
......@@ -58,10 +41,6 @@ public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent,
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameMappingDao = jdbiContext.getJdbi().onDemand(GameMappingDao.class);
gameState = getRuntimeContext().getMapState(stateDescriptor);
mediaSourceState = getRuntimeContext().getMapState(mediasourceStateDescriptor);
mediaCampaignState = getRuntimeContext().getMapState(mediaCampaignStateDescriptor);
adMappingMapState = getRuntimeContext().getMapState(adMappingMapStateDescriptor);
}
@Override
......@@ -72,48 +51,29 @@ public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent,
}
//更新game
Game game = gameState.get(value.getGame_id() + value.getPlatform());
if (game == null) {
boolean gameByPrimary = gameMappingDao.findGameByPrimary(value, platform_id);
if (!gameByPrimary) {
gameMappingDao.saveGame(value, platform_id);
}
gameState.put(value.getGame_id() + value.getPlatform(), new Game(platform_id, value.getGame_id()));
}
//mediaSource更新
String mediaSource = value.getMedia_source();
if (!Constants.DEFAULT_FIELD_VALUE.equals(mediaSource)) {
Media media = mediaSourceState.get(mediaSource);
if (media == null) {
Optional<Media> mediaByPrimary = gameMappingDao.findMediaByPrimary(value);
if (!mediaByPrimary.isPresent()) {
gameMappingDao.saveMediaSource(value);
} else {
mediaSourceState.put(mediaSource, mediaByPrimary.get());
}
}
}
//MediaCampaign更新
MediaCampaign mediaCampaign = mediaCampaignState.get(value.getMedia_source() + value.getCampaign());
if (null == mediaCampaign) {
Media media = mediaSourceState.get(mediaSource);
if (null != media) {
if (!gameMappingDao.existsMediaCampaign(value)) {
Media media = mediaByPrimary.get();
Optional<MediaCampaign> mediaCampaign = gameMappingDao.findMediaCampaign(value);
if (!mediaCampaign.isPresent() || !mediaCampaign.get().getAf_group_name().equals(value.getCampaign())) {
gameMappingDao.saveMediaCampaign(value, media);
mediaCampaignState.put(value.getMedia_source() + value.getCampaign(), new MediaCampaign(value.getMedia_source(), value.getCampaign()));
}
}
}
//adMapping 更新
AdMapping adMapping = adMappingMapState.get(value.getCampaign_id() + value.getAd_set_id() + value.getAd_id());
if (null == adMapping) {
Media media = mediaSourceState.get(mediaSource);
if (null != media) {
// adMapping 更新
Optional<AdMapping> adMapping = gameMappingDao.findAdMapping(value);
if (!adMapping.isPresent() ||
!adMapping.get().getAf_ad_group_name().equals(value.getCampaign()) ||
!adMapping.get().getAf_ad_plan_name().equals(value.getAd_set()) ||
!adMapping.get().getAf_ad_name().equals(value.getAd())) {
gameMappingDao.saveAdMapping(value, media);
adMappingMapState.put(value.getCampaign_id() + value.getAd_set_id() + value.getAd_id(),new AdMapping(value.getCampaign_id(),value.getAd_set_id(),value.getAd_id()));
}
}
......
......@@ -15,6 +15,7 @@ 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.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;
......@@ -22,7 +23,7 @@ import org.apache.http.util.TextUtils;
import java.util.Optional;
public class OuidOrderDataProcessor extends ProcessAllWindowFunction<AppsFlyerEvent, Object, TimeWindow> {
public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> {
private AppConfig appConfig;
......@@ -53,9 +54,8 @@ public class OuidOrderDataProcessor extends ProcessAllWindowFunction<AppsFlyerEv
}
@Override
public void process(Context context, Iterable<AppsFlyerEvent> elements, Collector<Object> out) throws Exception {
for (AppsFlyerEvent value : elements) {
try {
public void invoke(AppsFlyerEvent value, Context context) throws Exception {
super.invoke(value, context);
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
String ouid = jsonObject.getString("ouid");
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
......@@ -77,12 +77,7 @@ public class OuidOrderDataProcessor extends ProcessAllWindowFunction<AppsFlyerEv
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;
}
}
orderDao.insertOuidOrder(orderData);
}
}
......@@ -46,31 +46,64 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
}
private void insertRemindMetrics(Iterable<ActivationMetrics> elements) {
// days = days + 1;
// 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" +
// " (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,`%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.getData_type(), value.getUser_num()
// );
// handle.createUpdate(sql).execute();
// return null;
// });
// }
try {
PreparedStatement insertStmt = conn.prepareStatement("insert into appsflyer_metrics " +
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,?) " +
" values (?,?,?,?,?,?,?,?,?,?,?,?,?) ");
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,`data_active_2day`,`data_active_3day`,`data_active_7day`,`data_active_15day`,`data_active_30day`) " +
" values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ");
for (ActivationMetrics value : elements) {
Long days = value.getDateRemind() + 1;
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(2, value.getData_hour().toString());
insertStmt.setString(3, value.getTime_zone());
insertStmt.setString(4, value.getRegion());
insertStmt.setString(5, value.getPlatform());
insertStmt.setString(6, value.getGame_id());
insertStmt.setString(7, value.getMedia_source());
insertStmt.setString(8, value.getCampaign_id());
insertStmt.setString(9, value.getAd_set_id());
insertStmt.setString(10, value.getAd_id());
insertStmt.setString(11, value.getChannel());
insertStmt.setLong(12, value.getData_type());
if (days == 2) {
insertStmt.setLong(13, 1L);
} else {
insertStmt.setLong(13, 0L);
}
if (days == 3) {
insertStmt.setLong(14, 1L);
} else {
insertStmt.setLong(14, 0L);
}
if (days == 7) {
insertStmt.setLong(15, 1L);
} else {
insertStmt.setLong(15, 0L);
}
if (days == 15) {
insertStmt.setLong(16, 1L);
} else {
insertStmt.setLong(16, 0L);
}
} catch (Exception e) {
if (days == 30) {
insertStmt.setLong(17, 1L);
} else {
insertStmt.setLong(17, 0L);
}
insertStmt.addBatch();
}
}
insertStmt.executeBatch(); //批量执行
conn.commit();
insertStmt.close();
} catch (Exception e) {
e.printStackTrace();
}
......@@ -194,7 +227,7 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
}
if (element.getData_ouid_register() > 0) {
return "ouid_register";
return "ouid_role";
}
return null;
}
......
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