Commit 32343379 authored by sheteng's avatar sheteng

配置打包,修改bug

parent e592aab7
......@@ -448,7 +448,6 @@ order by(`data_date`,`game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`,
SETTINGS index_granularity = 8192
//留存迁移数据
//留存迁移数据
INSERT into appsflyer_retention
select
......
import com.github.jengelman.gradle.plugins.shadow.tasks.ConfigureShadowRelocation
buildscript {
repositories {
mavenCentral()
}
}
apply plugin: "java"
apply plugin: "application"
plugins {
//gradle的插件,shadow这个插件可以构建的时候把引用的jar包合进去
id 'java'
id 'application'
id 'com.github.johnrengelman.shadow' version '6.0.0'
}
group 'com.fshark.overseas.advert'
version '1.0-SNAPSHOT'
......@@ -13,7 +18,10 @@ version '1.0-SNAPSHOT'
repositories {
mavenCentral()
}
group = 'com.fshark.oversea.advert'
version = '0.3'
//applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"]
mainClassName = 'com.fshark.overseas.advert.AppsflyerAnalyze'
ext {
javaVersion = '1.8'
flinkVersion = '1.11.1'
......@@ -27,28 +35,14 @@ tasks.withType(JavaCompile) {
options.encoding = 'UTF-8'
}
//applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"]
mainClassName = 'com.fshark.overseas.advert.AppsflyerAnalyze'
repositories {
mavenCentral()
}
// NOTE: We cannot use "compileOnly" or "shadow" configurations since then we could not run code
// in the IDE or with "gradle run". We also cannot exclude transitive dependencies from the
// shadowJar yet (see https://github.com/johnrengelman/shadow/issues/159).
// -> Explicitly define the // libraries we want to be included in the "flinkShadowJar" configuration!
configurations {
flinkShadowJar // dependencies which go into the shadowJar
// always exclude these (also from transitive dependencies) since they are provided by Flink
flinkShadowJar.exclude group: 'org.apache.flink', module: 'force-shading'
flinkShadowJar.exclude group: 'com.google.code.findbugs', module: 'jsr305'
flinkShadowJar.exclude group: 'org.slf4j'
flinkShadowJar.exclude group: 'log4j'
}
dependencies {
shadow localGroovy()
shadow gradleApi()
testImplementation 'org.junit.jupiter:junit-jupiter-api:5.7.0'
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.7.0'
compile 'com.github.ben-manes.caffeine:caffeine:2.8.5'
......@@ -61,12 +55,12 @@ dependencies {
compile 'com.alibaba:fastjson:1.2.62'
compile "org.apache.flink:flink-connector-kafka_${scalaBinaryVersion}:${flinkVersion}"
// 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'
compile "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-java:${flinkVersion}"
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-java:${flinkVersion}"
compileOnly "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52'
// --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g.
......@@ -88,27 +82,10 @@ dependencies {
testCompile "org.apache.flink:flink-test-utils_${scalaBinaryVersion}:${flinkVersion}"
}
test {
useJUnitPlatform()
}
// make compileOnly dependencies available for tests:
sourceSets {
main.compileClasspath += configurations.flinkShadowJar
main.runtimeClasspath += configurations.flinkShadowJar
test.compileClasspath += configurations.flinkShadowJar
test.runtimeClasspath += configurations.flinkShadowJar
javadoc.classpath += configurations.flinkShadowJar
task relocateShadowJar(type: ConfigureShadowRelocation) {
target = tasks.shadowJar
}
run.classpath = sourceSets.main.runtimeClasspath
tasks.shadowJar.dependsOn tasks.relocateShadowJar
jar {
manifest {
attributes 'Built-By': System.getProperty('user.name'),
'Build-Jdk': System.getProperty('java.version')
}
}
......@@ -2,8 +2,8 @@ com/fshark/overseas/advert/modle/AppsFlyerWithAccountEvent.java
com.fshark.overseas.advert.modle.AppsFlyerWithAccountEvent
com/fshark/overseas/advert/modle/PlatformEnum.java
com.fshark.overseas.advert.modle.PlatformEnum
com/fshark/overseas/advert/util/Common.java
com.fshark.overseas.advert.util.Common
com/fshark/overseas/advert/sink/RemainSink.java
com.fshark.overseas.advert.sink.RemainSink
com/fshark/overseas/advert/util/AppConfig.java
com.fshark.overseas.advert.util.AppConfig
com.fshark.overseas.advert.util.AppConfig$AppConfigBuilder
......@@ -28,12 +28,12 @@ com/fshark/overseas/advert/util/DatabaseUtils.java
com.fshark.overseas.advert.util.DatabaseUtils
com/fshark/overseas/advert/entity/dao/GameMappingDao.java
com.fshark.overseas.advert.entity.dao.GameMappingDao
com/fshark/overseas/advert/modle/metrics/ActivationMetrics.java
com.fshark.overseas.advert.modle.metrics.ActivationMetrics
com/fshark/overseas/advert/AppsflyerAnalyze.java
com.fshark.overseas.advert.AppsflyerAnalyze
com/fshark/overseas/advert/modle/metrics/BaseMetrics.java
com.fshark.overseas.advert.modle.metrics.BaseMetrics
com/fshark/overseas/advert/modle/mapping/AdMapping.java
com.fshark.overseas.advert.modle.mapping.AdMapping
com/fshark/overseas/advert/sink/ActicitionSink.java
com.fshark.overseas.advert.sink.ActicitionSink
com/fshark/overseas/advert/modle/AppsflyerRoleLogin.java
com.fshark.overseas.advert.modle.AppsflyerRoleLogin
com/fshark/overseas/advert/util/JdbiContext.java
......@@ -50,12 +50,14 @@ com/fshark/overseas/advert/sink/PurchaseSink.java
com.fshark.overseas.advert.sink.PurchaseSink
com/fshark/overseas/advert/modle/mapping/MediaCampaign.java
com.fshark.overseas.advert.modle.mapping.MediaCampaign
com/fshark/overseas/advert/util/CommonUtils.java
com.fshark.overseas.advert.util.CommonUtils
com/fshark/overseas/advert/modle/OrderData.java
com.fshark.overseas.advert.modle.OrderData
com/fshark/overseas/advert/modle/AppsFlyerEvent.java
com.fshark.overseas.advert.modle.AppsFlyerEvent
com/fshark/overseas/advert/processor/ParseEventProcessor.java
com.fshark.overseas.advert.processor.ParseEventProcessor
com/fshark/overseas/advert/entity/dao/MetricsDao.java
com.fshark.overseas.advert.entity.dao.MetricsDao
com/fshark/overseas/advert/processor/CreateRoleProcess.java
com.fshark.overseas.advert.processor.CreateRoleProcess
com/fshark/overseas/advert/processor/ActivationResultWindows.java
......@@ -87,11 +89,11 @@ com/fshark/overseas/advert/util/ClickFeedbackRecord.java
com.fshark.overseas.advert.util.ClickFeedbackRecord$ClickFeedbackRecordBuilder
com/fshark/overseas/advert/entity/mapper/MediaMapper.java
com.fshark.overseas.advert.entity.mapper.MediaMapper
com/fshark/overseas/advert/sink/RemainSink.java
com.fshark.overseas.advert.sink.RemainSink
com/fshark/overseas/advert/AppsflyerAnalyze.java
com.fshark.overseas.advert.AppsflyerAnalyze
com/fshark/overseas/advert/entity/dao/MetricsDao.java
com.fshark.overseas.advert.entity.dao.MetricsDao
com/fshark/overseas/advert/sink/ActicitionSink.java
com.fshark.overseas.advert.sink.ActicitionSink
com/fshark/overseas/advert/modle/mapping/AdMapping.java
com.fshark.overseas.advert.modle.mapping.AdMapping
com/fshark/overseas/advert/processor/ParseEventProcessor.java
com.fshark.overseas.advert.processor.ParseEventProcessor
com/fshark/overseas/advert/modle/metrics/ActivationMetrics.java
com.fshark.overseas.advert.modle.metrics.ActivationMetrics
com/fshark/overseas/advert/util/Common.java
com.fshark.overseas.advert.util.Common
......@@ -65,13 +65,15 @@ public class AppsflyerAnalyze {
properties.setProperty("group.id", appConfig.getSourceKafkaGroupId());
properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
// TODO: 2021/8/10 设置消费开始的时间戳1628524944000
dataConsumer.setStartFromTimestamp(1628524944000L);
DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer)
.flatMap(new ParseEventProcessor())
.uid("parse-event")
.name("parse-event")
.shuffle();
adEventDataStream.print();
// adEventDataStream.print();
//广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream
......
......@@ -23,26 +23,26 @@ public interface GameAccountDao {
* @param
* @return
*/
@SqlQuery("SELECT * FROM appsflyer_activation_devices WHERE game_id=:game_id AND uid=:uid And platform =:platform AND event_type =:event_type order by event_time limit 1")
@SqlQuery("SELECT * FROM appsflyer_activation_devices WHERE game_id=:game_id AND uid=:uid And platform =:platform AND event_type =:event_type order by event_time desc limit 1")
@RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findByPrimary(@BindBean AppsFlyerEvent event, @Bind("event_type") String event_type);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`,`ip`, `time_zone`,`event_type`,`event_time`) " +
"values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :ip, :time_zone,'activition',:reg_time) ")
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
"values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'activation',:reg_time) ")
void insertActivition(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`,`ip`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :ip, :time_zone,'register',:reg_time ) ")
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'register',:reg_time ) ")
void insertRegister(@BindBean AppsFlyerEvent value, @Bind("reg_time") String reg_time);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`,`ip`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :ip, :time_zone,'rolelogin',: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,'rolelogin',:event_time ) ")
void insertRoleLogin(@BindBean AppsFlyerEvent value);
@SqlUpdate("insert into appsflyer_activation_devices (`platform`,`did`, `uid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`,`ip`, `time_zone`,`event_type`,`event_time`) " +
" values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :ip, :time_zone,'pauchase',: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,'purchase',:event_time ) ")
void insertPauchase(@BindBean AppsFlyerEvent value);
}
......@@ -42,7 +42,7 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
if (!account.isPresent() || value.getIs_retargeting()) { //
if (!account.isPresent()) { //
gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
......
......@@ -41,23 +41,19 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
activationMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
Date activition;
if (!account.isPresent()) { // 设备未激活
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);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_create_role(1L);
activationMetrics.setData_day_register_create(1L);
activationMetrics.setData_day_act_register(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)));
Date activition = DateUtils.formatAsDate(activationTime);
activition = DateUtils.formatAsDate(activationTime);
}
activationMetrics.setData_create_role(1L);
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) {
......@@ -72,7 +68,6 @@ public class CreateRoleProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
activationMetrics.setData_day_register_create(1L);
}
}
}
gameAccountDao.insertRegister(value, event_time);
out.collect(activationMetrics);
}
......
......@@ -45,18 +45,14 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
if (!account.isPresent()) { // 设备未激活
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);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
activationMetrics.setData_day_act_register(1L);
gameAccountDao.insertRegister(value, event_time);
// activationMetrics.setData_activation(1L);
} 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)));
CommonUtils.setType(activationMetrics);
Date activition = DateUtils.formatAsDate(activationTime);
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { //设备未注册
......@@ -72,7 +68,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
String last_login_time = accountRoleLogin.get().getEvent_time();
if (!StringUtils.isBlank(last_login_time)) {
Date date = DateUtils.formatAsDate(last_login_time);
if (DateUtils.daysBetween(date, value.getEvent_time()) != 0) { //排除当日登录用户
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween); //留存天数
......
......@@ -46,19 +46,17 @@ public class PurchaseSink extends RichSinkFunction<AppsFlyerEvent> {
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
OrderData orderData = new OrderData(value);
String reg_time = null;
Optional<GameAccount> accountActivition = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (accountActivition.isPresent()) {
GameAccount gameAccount = accountActivition.get();
orderData.setActivation_time(gameAccount.getEvent_time());
reg_time = gameAccount.getEvent_time();
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());
reg_time = gameAccount.getEvent_time();
}
orderDao.insertOrder(orderData);
gameAccountDao.insertPauchase(value);
......
......@@ -26,7 +26,7 @@ public class RemainSink extends RichSinkFunction<RemainMetrics> {
if ((days >= 1 && days <= 30) || days == 35 || days == 40 || days == 45 || days == 50 || days == 55
|| days == 60 || days == 65 || days == 70 || days == 75 || days == 80 || days == 85 || days == 90 || days == 120 || days == 150 || days == 180) {
String column = "retention_day" + days;
// metricsDao.insertLongRetentionNum(value, column);
jdbiContext.getJdbi().withHandle(handle -> {
String sql = String.format("insert into appsflyer_retention" +
" (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,%s)" +
......
package com.fshark.overseas.advert.util;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
public class CommonUtils {
public static void setType(BaseMetrics date){
if (null != date.getData_date() && (date.getData_date() >= 20210519 && date.getData_date() <= 20210521 && "organic".equals(date.getMedia_source()))){
date.setData_type(3);
}
}
public static int setType(Long actTime,String mediasource,int dataType){
if (actTime>= 20210519 && actTime <= 20210521 && "organic".equals(mediasource)){
return 3;
}
return dataType;
}
}
......@@ -69,9 +69,9 @@ public class Constants {
*/
public static final String REMATCHED = "rematched";
public static final String TYPE_ACTIVIYION = "activition";
public static final String TYPE_ACTIVIYION = "activation";
public static final String TYPE_REGISTER = "register";
public static final String TYPE_ROLELOGIN = "rolelogin";
public static final String TYPE_PAUCHASE = "pauchase";
public static final String TYPE_PAUCHASE = "purchase";
}
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