Commit bbbdc8ac authored by she's avatar she

sdk注册数

parent 3d588d75
...@@ -55,10 +55,10 @@ dependencies { ...@@ -55,10 +55,10 @@ dependencies {
// https://mvnrepository.com/artifact/mysql/mysql-connector-java // https://mvnrepository.com/artifact/mysql/mysql-connector-java
implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.25' implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.25'
compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}" compile "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-java:${flinkVersion}" compile "org.apache.flink:flink-java:${flinkVersion}"
compileOnly "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}" compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}" compile "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52' implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52'
// -------------------------------------------------------------- // --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g. // Dependencies that should be part of the shadow jar, e.g.
......
...@@ -94,7 +94,7 @@ public class AppsflyerAnalyze { ...@@ -94,7 +94,7 @@ public class AppsflyerAnalyze {
.uid("parse-event") .uid("parse-event")
.name("parse-event"); .name("parse-event");
adEventDataStream.print();
//测流提取出来激活,登录,创角角色登录,订单事件 //测流提取出来激活,登录,创角角色登录,订单事件
OutputTag<AppsFlyerEvent> installTag = new OutputTag<AppsFlyerEvent>("install") { OutputTag<AppsFlyerEvent> installTag = new OutputTag<AppsFlyerEvent>("install") {
}; };
...@@ -146,12 +146,12 @@ public class AppsflyerAnalyze { ...@@ -146,12 +146,12 @@ public class AppsflyerAnalyze {
.name("role-login-event"); .name("role-login-event");
//sdk激活注册 当天 //sdk激活注册 当天
DataStream<ActivationMetrics> sdkMetrics = sdkRegisterStream // DataStream<ActivationMetrics> sdkMetrics = sdkRegisterStream
.flatMap(new SdkProcess(appConfig)) // .flatMap(new SdkProcess(appConfig))
.uid("sdk-event") // .uid("sdk-event")
.name("sdk-event"); // .name("sdk-event");
activitionMetricStream.union(loginMetricStream).union(sdkMetrics) activitionMetricStream.union(loginMetricStream)
.flatMap(new ActivitionMetricToStringProcess(appConfig)) .flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("sink-metric-event") .uid("sink-metric-event")
...@@ -180,6 +180,7 @@ public class AppsflyerAnalyze { ...@@ -180,6 +180,7 @@ public class AppsflyerAnalyze {
.uid("sink_ouid-event") .uid("sink_ouid-event")
.name("sink_ouid-event"); .name("sink_ouid-event");
// af订单信息 // af订单信息
orderStream orderStream
.flatMap(new PurchaseToOrderDataProcessor(appConfig)) .flatMap(new PurchaseToOrderDataProcessor(appConfig))
......
...@@ -3,7 +3,6 @@ package com.fshark.overseas.advert.app; ...@@ -3,7 +3,6 @@ package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum; import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.processor.ParseEventProcessor; import com.fshark.overseas.advert.processor.ParseEventProcessor;
import com.fshark.overseas.advert.sink.EventInfoSink;
import com.fshark.overseas.advert.sink.EventInfoSinkProcess; import com.fshark.overseas.advert.sink.EventInfoSinkProcess;
import com.fshark.overseas.advert.util.AppConfig; import com.fshark.overseas.advert.util.AppConfig;
import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.FilterFunction;
......
...@@ -69,4 +69,8 @@ public interface GameAccountDao { ...@@ -69,4 +69,8 @@ public interface GameAccountDao {
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " + @SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
"values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'sdkRegister',:reg_time) ") "values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'sdkRegister',:reg_time) ")
void insertSdkRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time); void insertSdkRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time);
@SqlQuery("SELECT * FROM appsflyer_activation_devices WHERE game_id=:game_id AND ouid=:ouid And platform =:platform AND event_type =:event_type order by event_time desc limit 1")
@RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findRegisterByouid(AppsFlyerEvent value, String ouid, String typeRegister);
} }
...@@ -4,10 +4,6 @@ import com.fshark.overseas.advert.entity.GameAccount; ...@@ -4,10 +4,6 @@ import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao; import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.entity.dao.GameMappingDao; import com.fshark.overseas.advert.entity.dao.GameMappingDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; 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.Media;
import com.fshark.overseas.advert.modle.mapping.MediaCampaign;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics; import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*; import com.fshark.overseas.advert.util.*;
...@@ -83,7 +79,7 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -83,7 +79,7 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
if (value.getIs_retargeting()) { //再营销数据 if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2); metrics.setData_type(2);
} }
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据 if (value.getCampaign().toLowerCase(Locale.ROOT).contains("acp") || value.getCampaign().toLowerCase(Locale.ROOT).contains("pre")) { //预注册数据
metrics.setData_type(3); metrics.setData_type(3);
} }
......
...@@ -12,7 +12,6 @@ import org.apache.flink.api.common.functions.RichFlatMapFunction; ...@@ -12,7 +12,6 @@ import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector; import org.apache.flink.util.Collector;
import org.apache.http.util.TextUtils; import org.apache.http.util.TextUtils;
import org.gradle.internal.impldep.org.joda.time.DateTime;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Locale; import java.util.Locale;
...@@ -137,7 +136,7 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -137,7 +136,7 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
if (value.getIs_retargeting()) { //再营销数据 if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2); metrics.setData_type(2);
} }
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据 if (value.getCampaign().toLowerCase(Locale.ROOT).contains("acp") || value.getCampaign().toLowerCase(Locale.ROOT).contains("pre")) { //预注册数据
metrics.setData_type(3); metrics.setData_type(3);
} }
......
...@@ -9,7 +9,6 @@ import com.fshark.overseas.advert.modle.EventTypeEnum; ...@@ -9,7 +9,6 @@ import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics; import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*; import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector; import org.apache.flink.util.Collector;
...@@ -90,6 +89,11 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -90,6 +89,11 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
} else { } else {
Optional<GameAccount> sdkaAccount = gameAccountDao.findRegisterByouid(value,ouid, Constants.TYPE_REGISTER);
if (!sdkaAccount.isPresent()) {
activationMetrics.setData_sdk_register(1L);
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER); Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { //设备未注册 if (!accountRegister.isPresent()) { //设备未注册
gameAccountDao.insertRegister(value, event_time,ouid); gameAccountDao.insertRegister(value, event_time,ouid);
...@@ -147,7 +151,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -147,7 +151,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
if (value.getIs_retargeting()) { //再营销数据 if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2); metrics.setData_type(2);
} }
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据 if (value.getCampaign().toLowerCase(Locale.ROOT).contains("acp") || value.getCampaign().toLowerCase(Locale.ROOT).contains("pre")) { //预注册数据
metrics.setData_type(3); metrics.setData_type(3);
} }
} }
......
...@@ -3,7 +3,6 @@ package com.fshark.overseas.advert.processor; ...@@ -3,7 +3,6 @@ package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.GameAccount; import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao; import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics; import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*; import com.fshark.overseas.advert.util.*;
...@@ -65,7 +64,7 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe ...@@ -65,7 +64,7 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
if (value.getIs_retargeting()) { //再营销数据 if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2); metrics.setData_type(2);
} }
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据 if (value.getCampaign().toLowerCase(Locale.ROOT).contains("acp") || value.getCampaign().toLowerCase(Locale.ROOT).contains("pre")) { //预注册数据
metrics.setData_type(3); metrics.setData_type(3);
} }
......
...@@ -4,7 +4,6 @@ import com.fshark.overseas.advert.modle.AppsFlyerEvent; ...@@ -4,7 +4,6 @@ import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.util.AppConfig; import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants; import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DateUtils; import com.fshark.overseas.advert.util.DateUtils;
import lombok.val;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction; import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
...@@ -64,6 +63,7 @@ public class EventInfoSinkProcess extends ProcessAllWindowFunction<AppsFlyerEven ...@@ -64,6 +63,7 @@ public class EventInfoSinkProcess extends ProcessAllWindowFunction<AppsFlyerEven
insertStmt.setString(11, element.getAd_id()); insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getAd()); insertStmt.setString(12, element.getAd());
insertStmt.setString(13, element.getEvent()); insertStmt.setString(13, element.getEvent());
// DatePattern.ISO8601_FORMAT.format()
insertStmt.setString(14, DateUtils.formatAsYMDHMS(element.getEvent_time())); insertStmt.setString(14, DateUtils.formatAsYMDHMS(element.getEvent_time()));
insertStmt.setString(15, element.getEvent_value()); insertStmt.setString(15, element.getEvent_value());
insertStmt.setString(16, element.getChannel()); insertStmt.setString(16, element.getChannel());
......
...@@ -37,8 +37,8 @@ public class TiktokSink extends RichSinkFunction<AppsFlyerEvent> { ...@@ -37,8 +37,8 @@ public class TiktokSink extends RichSinkFunction<AppsFlyerEvent> {
.body(body) .body(body)
.timeout(20000) //超时,毫秒 .timeout(20000) //超时,毫秒
.execute().body(); .execute().body();
// System.out.println(body); System.out.println(body);
// System.out.println(result); System.out.println(result);
} }
@Override @Override
...@@ -61,7 +61,7 @@ public class TiktokSink extends RichSinkFunction<AppsFlyerEvent> { ...@@ -61,7 +61,7 @@ public class TiktokSink extends RichSinkFunction<AppsFlyerEvent> {
contextDTO.setUserAgent(value.getUser_agent()); contextDTO.setUserAgent(value.getUser_agent());
contextDTO.setOriginUrl(value.getOriginal_url()); contextDTO.setOriginUrl(value.getOriginal_url());
TiktokModle.ContextDTO.AppDTO appDTO = new TiktokModle.ContextDTO.AppDTO(); TiktokModle.ContextDTO.AppDTO appDTO = new TiktokModle.ContextDTO.AppDTO();
appDTO.setId(value.getApp_id()); appDTO.setId(value.getApp_id().replace("id",""));
contextDTO.setApp(appDTO); contextDTO.setApp(appDTO);
TiktokModle.ContextDTO.DeviceDTO deviceDTO = new TiktokModle.ContextDTO.DeviceDTO(); TiktokModle.ContextDTO.DeviceDTO deviceDTO = new TiktokModle.ContextDTO.DeviceDTO();
deviceDTO.setPlatform(value.getPlatform()); deviceDTO.setPlatform(value.getPlatform());
......
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