Commit 98f00dec authored by sheteng's avatar sheteng

ouid 报表

parent f4660a13
...@@ -15,9 +15,6 @@ plugins { ...@@ -15,9 +15,6 @@ plugins {
group 'com.fshark.overseas.advert' group 'com.fshark.overseas.advert'
version '1.0-SNAPSHOT' version '1.0-SNAPSHOT'
repositories {
mavenCentral()
}
group = 'com.fshark.oversea.advert' group = 'com.fshark.oversea.advert'
version = '0.3' version = '0.3'
//applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"] //applicationDefaultJvmArgs = ["-Dlog4j.configuration=log4j.properties"]
...@@ -36,10 +33,10 @@ tasks.withType(JavaCompile) { ...@@ -36,10 +33,10 @@ tasks.withType(JavaCompile) {
} }
repositories { repositories {
mavenCentral()
maven { maven {
url "https://nexus.sky-shark.com/repository/ccnet-group/" url "https://nexus.sky-shark.com/repository/ccnet-group/"
} }
mavenCentral()
} }
dependencies { dependencies {
......
...@@ -3,21 +3,30 @@ package com.fshark.overseas.advert.app; ...@@ -3,21 +3,30 @@ 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.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.EventMetrics; import com.fshark.overseas.advert.modle.metrics.EventMetrics;
import com.fshark.overseas.advert.processor.*; import com.fshark.overseas.advert.processor.*;
import com.fshark.overseas.advert.util.AppConfig; 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.FilterFunction; import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
...@@ -62,14 +71,14 @@ public class AppsflyerAnalyze { ...@@ -62,14 +71,14 @@ public class AppsflyerAnalyze {
AppConfig appConfig = loadConfigFromArgs(args); AppConfig appConfig = loadConfigFromArgs(args);
final StreamExecutionEnvironment sse = StreamExecutionEnvironment.getExecutionEnvironment(); final StreamExecutionEnvironment sse = StreamExecutionEnvironment.getExecutionEnvironment();
sse.setParallelism(1); sse.setParallelism(1);
sse.enableCheckpointing(1000 * 60 * 10); sse.enableCheckpointing(1000 * 60);
sse.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS))); sse.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS)));
//source设置 //source设置
Properties properties = new Properties(); Properties properties = new Properties();
properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers()); properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers());
properties.setProperty("group.id", appConfig.getSourceKafkaGroupId()); properties.setProperty("group.id", appConfig.getSourceKafkaGroupId());
// properties.setProperty("auto.offset.reset", "earliest"); properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties); FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
Properties props2 = new Properties(); Properties props2 = new Properties();
...@@ -81,68 +90,129 @@ public class AppsflyerAnalyze { ...@@ -81,68 +90,129 @@ public class AppsflyerAnalyze {
.filter(new FilterFunction<AppsFlyerEvent>() { .filter(new FilterFunction<AppsFlyerEvent>() {
@Override @Override
public boolean filter(AppsFlyerEvent value) throws Exception { 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 false;
} }
return true; return true;
} }
}) })
.uid("parse-event") .uid("parse-event")
.name("parse-event") .name("parse-event");
.shuffle();
adEventDataStream.print(); //测流提取出来激活,登录,创角角色登录,订单事件
OutputTag<AppsFlyerEvent> installTag = new OutputTag<AppsFlyerEvent>("install") {
};
OutputTag<AppsFlyerEvent> loginTag = new OutputTag<AppsFlyerEvent>("login") {
};
OutputTag<AppsFlyerEvent> roleLoginTag = new OutputTag<AppsFlyerEvent>("role_login") {
};
OutputTag<AppsFlyerEvent> orderTag = new OutputTag<AppsFlyerEvent>("order") {
};
OutputTag<AppsFlyerEvent> sdkRegisterTag = new OutputTag<AppsFlyerEvent>("sdk_register") {
};
SingleOutputStreamOperator<AppsFlyerEvent> mainStream = adEventDataStream.process(new ProcessFunction<AppsFlyerEvent, AppsFlyerEvent>() {
@Override
public void processElement(AppsFlyerEvent value, Context ctx, Collector<AppsFlyerEvent> out) throws Exception {
if (EventTypeEnum.installEvent(value)) {
ctx.output(installTag, value);
} else if (EventTypeEnum.loginEvent(value)) {
ctx.output(loginTag, value);
} else if (EventTypeEnum.roleEvent(value)) {
ctx.output(roleLoginTag, value);
} else if (EventTypeEnum.orderEvent(value)) {
ctx.output(orderTag, value);
} else if (EventTypeEnum.sdkEvent(value)) {
ctx.output(sdkRegisterTag, value);
} else {
out.collect(value);
}
}
}).uid("shuffle-event").name("shuffle-event");
DataStream<AppsFlyerEvent> installStream = mainStream.getSideOutput(installTag);
DataStream<AppsFlyerEvent> loginStream = mainStream.getSideOutput(loginTag);
DataStream<AppsFlyerEvent> roleloginStream = mainStream.getSideOutput(roleLoginTag);
DataStream<AppsFlyerEvent> orderStream = mainStream.getSideOutput(orderTag);
DataStream<AppsFlyerEvent> sdkRegisterStream = mainStream.getSideOutput(sdkRegisterTag);
// loginStream.print();
//广告指标(激活) //广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream DataStream<ActivationMetrics> activitionMetricStream = installStream
.filter(EventTypeEnum::installEvent)
.flatMap(new ActivitionProcess(appConfig)) .flatMap(new ActivitionProcess(appConfig))
.uid("activition-event") .uid("activition-event")
.name("activition-event"); .name("activition-event");
//注册\留存 由于可能出现创角和角色登录是一瞬间的,所以他两合并计算 //注册\留存 由于可能出现创角和角色登录是一瞬间的,所以他两合并计算
DataStream<ActivationMetrics> loginRoleMetricStream = adEventDataStream DataStream<ActivationMetrics> loginMetricStream = loginStream.union(roleloginStream)
.filter(EventTypeEnum::roleLoginEvent)
.flatMap(new RoleLoginProcess(appConfig)) .flatMap(new RoleLoginProcess(appConfig))
.uid("role-login-event") .uid("role-login-event")
.name("role-login-event"); .name("role-login-event");
//sdk激活注册 当天 //sdk激活注册 当天
DataStream<ActivationMetrics> sdkMetrics = adEventDataStream DataStream<ActivationMetrics> sdkMetrics = sdkRegisterStream
.filter(EventTypeEnum::sdkEvent)
.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(loginRoleMetricStream).union(sdkMetrics)
.flatMap(new ActivitionMetricToStringProcess(appConfig)) .flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("sink-metric-event") .uid("sink-metric-event")
.name("sink-metric-event"); .name("sink-metric-event");
// 订单信息 /**
adEventDataStream.filter(EventTypeEnum::orderEvent) * ouid相关指标计算 注册留存日活
*/
DataStream<ActivationMetrics> ouidMetrics = loginStream
.flatMap(new OuidLoginProcess(appConfig))
.uid("ouid-event")
.name("ouid-event");
// ouidMetrics.print("ouid");
ouidMetrics.flatMap(new RichFlatMapFunction<ActivationMetrics, String>() {
@Override
public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception {
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_OUID_METRIC);
out.collect(new Gson().toJson(sinkMetricEvent));
}
})
.addSink(kafkaSink)
.uid("sink_ouid-event")
.name("sink_ouid-event");
// af订单信息
orderStream
.flatMap(new PurchaseToOrderDataProcessor(appConfig)) .flatMap(new PurchaseToOrderDataProcessor(appConfig))
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("order-event") .uid("order-event")
.name("order-event"); .name("order-event");
//区服 角色注册,激活 /**
adEventDataStream * ouid 的订单信息
.filter(EventTypeEnum::roleEvent) */
.flatMap(new ServerActivationProcessor(appConfig)) orderStream.flatMap(new OuidOrderDataProcessor(appConfig))
.flatMap(new ServerMtricToStringProcessor())
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("activation-event") .uid("ouid_order-metrics")
.name("activation-event"); .name("ouid_order-metrics");
//区服 角色注册,激活 暂时删除
// roleloginStream
// .flatMap(new ServerActivationProcessor(appConfig))
// .flatMap(new ServerMtricToStringProcessor())
// .addSink(kafkaSink)
// .uid("activation-event")
// .name("activation-event");
//事件统计激活 //事件统计激活
adEventDataStream mainStream
.filter(EventTypeEnum::notHeartBeatEvent) .filter(EventTypeEnum::notHeartBeatEvent)
.keyBy(AppsFlyerEvent::getUid) .keyBy(AppsFlyerEvent::getUid)
.flatMap(new EventMectrisProcessor(appConfig)) .flatMap(new EventMectrisProcessor(appConfig))
.keyBy(EventMetrics::getKeys)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("times")
.flatMap(new EventMetricsToStringProcessor()) .flatMap(new EventMetricsToStringProcessor())
.addSink(kafkaSink) .addSink(kafkaSink)
.uid("aggregate-event") .uid("aggregate-event")
......
...@@ -15,7 +15,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; ...@@ -15,7 +15,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows; import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
...@@ -34,6 +33,8 @@ public class AppsflyerInfoDmi { ...@@ -34,6 +33,8 @@ public class AppsflyerInfoDmi {
appConfig.setDbUser(parameters.get("jdbc-user")); appConfig.setDbUser(parameters.get("jdbc-user"));
appConfig.setDbPassword(parameters.get("jdbc-password")); appConfig.setDbPassword(parameters.get("jdbc-password"));
appConfig.setSinkType(parameters.get("sink-type"));
} catch (Exception e) { } catch (Exception e) {
System.err.println("Place set args: " + System.err.println("Place set args: " +
" --source-kafka-bootstrap-servers <source-kafka-bootstrap-servers>" + " --source-kafka-bootstrap-servers <source-kafka-bootstrap-servers>" +
...@@ -79,7 +80,8 @@ public class AppsflyerInfoDmi { ...@@ -79,7 +80,8 @@ public class AppsflyerInfoDmi {
.name("parse-event") .name("parse-event")
.shuffle(); .shuffle();
adEventDataStream.filter(EventTypeEnum::notHeartBeatEvent).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).process(new EventInfoSinkProcess(appConfig)) adEventDataStream.filter(EventTypeEnum::notHeartBeatEvent).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new EventInfoSinkProcess(appConfig))
.uid("event") .uid("event")
.name("event"); .name("event");
......
...@@ -2,19 +2,11 @@ package com.fshark.overseas.advert.app; ...@@ -2,19 +2,11 @@ package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.SinkMetricEvent; import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
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.*;
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.EventTimesSinkProcess; import com.fshark.overseas.advert.sink.EventTimesSinkProcess;
import com.fshark.overseas.advert.sink.PurchaseSink; 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.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson; 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.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.restartstrategy.RestartStrategies;
...@@ -70,7 +62,7 @@ public class AppsflyerMetricStore { ...@@ -70,7 +62,7 @@ public class AppsflyerMetricStore {
Properties properties = new Properties(); Properties properties = new Properties();
properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers()); properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers());
properties.setProperty("group.id", appConfig.getSourceKafkaGroupId()); properties.setProperty("group.id", appConfig.getSourceKafkaGroupId());
// properties.setProperty("auto.offset.reset", "latest"); properties.setProperty("auto.offset.reset", "latest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties); FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
DataStream<SinkMetricEvent> activationMetricsDataStream = sse.addSource(dataConsumer) DataStream<SinkMetricEvent> activationMetricsDataStream = sse.addSource(dataConsumer)
...@@ -82,24 +74,47 @@ public class AppsflyerMetricStore { ...@@ -82,24 +74,47 @@ public class AppsflyerMetricStore {
}) })
.uid("parse-event") .uid("parse-event")
.name("parse-event"); .name("parse-event");
//角色注册
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
.uid("activation-agg-event")
.name("activation-agg-event");
activitionResult.print();
//指标详细记录 /**
activitionResult.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).process(new StoreMetricProcess(appConfig)) * afid 指标写入
*/
activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new StoreMetricProcess(appConfig))
.uid("metric-detail") .uid("metric-detail")
.name("metric-detail"); .name("metric-detail");
// 角色注册激活写入 /**
* ouid 指标写入
*/
activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ActivationMetrics>() {
@Override
public void flatMap(SinkMetricEvent value, Collector<ActivationMetrics> out) throws Exception {
if (Constants.TYPE_SINK_OUID_METRIC.equals(value.getEventType()) && value.getEventData() != null) {
out.collect(value.getEventData());
}
}
}).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new OuidStoreMetricProcess(appConfig))
.uid("ouid-sink-event")
.name("ouid-sink-event");
//订单写入 包含ouid订单信息
activationMetricsDataStream
.filter(SinkMetricEvent::isOrderEvent)
.addSink(new PurchaseSink(appConfig))
.uid("order-sink-event")
.name("order-sink-event");
// 角色注册激活写入 todo 废弃
// activitionResult.addSink(new ActicitionSink(appConfig)) // activitionResult.addSink(new ActicitionSink(appConfig))
// .uid("login-sink-event") // .uid("login-sink-event")
// .name("login-sink-event"); // .name("login-sink-event");
// 留存写入 // 留存写入 todo 废弃
// activationMetricsDataStream // activationMetricsDataStream
// .flatMap(new RemainProcess()) // .flatMap(new RemainProcess())
// .timeWindowAll(Time.seconds(30)).process() // .timeWindowAll(Time.seconds(30)).process()
...@@ -107,53 +122,48 @@ public class AppsflyerMetricStore { ...@@ -107,53 +122,48 @@ public class AppsflyerMetricStore {
// .uid("remain-sink-event") // .uid("remain-sink-event")
// .name("remain-sink-event"); // .name("remain-sink-event");
//订单写入
activationMetricsDataStream
.filter(SinkMetricEvent::isOrderEvent)
.addSink(new PurchaseSink(appConfig))
.uid("order-sink-event")
.name("order-sink-event");
// //
// //
//角色注册激活预聚合 //角色注册激活预聚合 todo 废弃
DataStream<ServerActivationMetrics> activationMetricsList = activationMetricsDataStream // DataStream<ServerActivationMetrics> activationMetricsList = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() { // .flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override // @Override
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) { // public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) {
if (value.isServer()) { // if (value.isServer()) {
out.collect(value.getServerActivationMetrics()); // out.collect(value.getServerActivationMetrics());
} // }
} // }
}) // })
.keyBy(ServerActivationMetrics::getKey) // .keyBy(ServerActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.aggregate(new ServerActivationAggregateFunction(), new ServerActivationResultWindows()) // .aggregate(new ServerActivationAggregateFunction(), new ServerActivationResultWindows())
.uid("server-activation-agg-event") // .uid("server-activation-agg-event")
.name("server-activation-agg-event"); // .name("server-activation-agg-event");
//
activationMetricsList.addSink(new ServerActivationSink(appConfig)) // activationMetricsList.addSink(new ServerActivationSink(appConfig))
.uid("server-activationsink-event") // .uid("server-activationsink-event")
.name("server-activationsink-event"); // .name("server-activationsink-event");
//留存预聚合 //留存预聚合 todo 废弃
DataStream<ServerRemainMetrics> remainMetrics = activationMetricsDataStream // DataStream<ServerRemainMetrics> remainMetrics = activationMetricsDataStream
.flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() { // .flatMap(new RichFlatMapFunction<SinkMetricEvent, ServerActivationMetrics>() {
@Override // @Override
public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) { // public void flatMap(SinkMetricEvent value, Collector<ServerActivationMetrics> out) {
if (value.isServer()) { // if (value.isServer()) {
out.collect(value.getServerActivationMetrics()); // out.collect(value.getServerActivationMetrics());
} // }
} // }
}) // })
.flatMap(new ServerRemainProcessor(appConfig)) // .flatMap(new ServerRemainProcessor(appConfig))
.keyBy(ServerRemainMetrics::getRemind_key) // .keyBy(ServerRemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum("user_num") // .sum("user_num")
.uid("server-remain-event") // .uid("server-remain-event")
.name("server-remain-event"); // .name("server-remain-event");
remainMetrics.addSink(new ServerRemindSink(appConfig)) // remainMetrics.addSink(new ServerRemindSink(appConfig))
.uid("server-remindsink-event") // .uid("server-remindsink-event")
.name("server-remindsink-event"); // .name("server-remindsink-event");
/** /**
* 事件写入 * 事件写入
......
...@@ -21,6 +21,9 @@ import org.apache.flink.util.Collector; ...@@ -21,6 +21,9 @@ import org.apache.flink.util.Collector;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/**
* 已废弃,合并到AppsflyerAnalyze
*/
public class AppsflyerOuidAnalyze { public class AppsflyerOuidAnalyze {
static AppConfig loadConfigFromArgs(String[] args) { static AppConfig loadConfigFromArgs(String[] args) {
...@@ -81,7 +84,7 @@ public class AppsflyerOuidAnalyze { ...@@ -81,7 +84,7 @@ public class AppsflyerOuidAnalyze {
.uid("parse-event") .uid("parse-event")
.name("parse-event") .name("parse-event")
.shuffle(); .shuffle();
adEventDataStream.print(); // adEventDataStream.print();
//ouid注册 //ouid注册
DataStream<ActivationMetrics> ouidMetrics = adEventDataStream DataStream<ActivationMetrics> ouidMetrics = adEventDataStream
.filter(EventTypeEnum::loginEvent) .filter(EventTypeEnum::loginEvent)
...@@ -89,7 +92,7 @@ public class AppsflyerOuidAnalyze { ...@@ -89,7 +92,7 @@ public class AppsflyerOuidAnalyze {
.uid("sdk-event") .uid("sdk-event")
.name("sdk-event"); .name("sdk-event");
//写入指标 //写入指标
ouidMetrics.print(); // ouidMetrics.print();
ouidMetrics.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))) ouidMetrics.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new StoreMetricProcess(appConfig)) .process(new StoreMetricProcess(appConfig))
.uid("store-metrics") .uid("store-metrics")
...@@ -97,10 +100,10 @@ public class AppsflyerOuidAnalyze { ...@@ -97,10 +100,10 @@ public class AppsflyerOuidAnalyze {
// ouidMetrics.timeWindowAll(Time.minutes(1)); // ouidMetrics.timeWindowAll(Time.minutes(1));
//订单信息 //订单信息
adEventDataStream.filter(EventTypeEnum::orderEvent) // adEventDataStream.filter(EventTypeEnum::orderEvent)
.addSink(new OuidOrderDataProcessor(appConfig)) // .addSink(new OuidOrderDataProcessor(appConfig))
.uid("order-metrics") // .uid("order-metrics")
.name("order-metrics"); // .name("order-metrics");
sse.execute("appsflyer-analyze"); sse.execute("appsflyer-analyze");
......
...@@ -15,6 +15,8 @@ public class GameAccount { ...@@ -15,6 +15,8 @@ public class GameAccount {
private String gameId; private String gameId;
private String region;
private String platform; private String platform;
private String mediaSource; private String mediaSource;
...@@ -29,4 +31,6 @@ public class GameAccount { ...@@ -29,4 +31,6 @@ public class GameAccount {
private String event_time; private String event_time;
private String channel;
} }
...@@ -3,6 +3,7 @@ package com.fshark.overseas.advert.entity.dao; ...@@ -3,6 +3,7 @@ package com.fshark.overseas.advert.entity.dao;
import com.fshark.overseas.advert.entity.GameAccount; import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.mapper.GameAccountMapper; import com.fshark.overseas.advert.entity.mapper.GameAccountMapper;
import com.fshark.overseas.advert.entity.mapper.OuidGameAccountMapper;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import org.jdbi.v3.sqlobject.config.RegisterRowMapper; import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
import org.jdbi.v3.sqlobject.customizer.Bind; import org.jdbi.v3.sqlobject.customizer.Bind;
...@@ -24,13 +25,19 @@ public interface GameAccountDao { ...@@ -24,13 +25,19 @@ public interface GameAccountDao {
@RegisterRowMapper(GameAccountMapper.class) @RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findByPrimary(@BindBean AppsFlyerEvent event, @Bind("event_type") String event_type); Optional<GameAccount> findByPrimary(@BindBean AppsFlyerEvent event, @Bind("event_type") String event_type);
@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") @SqlQuery("SELECT * FROM appsflyer_ouid 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) @RegisterRowMapper(OuidGameAccountMapper.class)
Optional<GameAccount> findByOuid(@BindBean AppsFlyerEvent event,@Bind("ouid") String ouid); Optional<GameAccount> findByOuid(@BindBean AppsFlyerEvent event, @Bind("ouid") String ouid, @Bind("event_type") String event_type);
@SqlUpdate("insert into appsflyer_ouid (`platform`,`did`, `uid`,`ouid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`,`channel`) " +
"values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'ouidRegister',:reg_time,:channel) ")
void insertOuidRegister(@BindBean AppsFlyerEvent event, @Bind("reg_time") String reg_time, @Bind("ouid") String ouid);
@SqlUpdate("insert into appsflyer_ouid (`platform`,`did`, `uid`,`ouid`, `region`, `game_id`,`media_source`,`campaign_id`,`ad_set_id`,`ad_id`, `time_zone`,`event_type`,`event_time`,`channel`) " +
" values ( :platform,:did ,:uid,:ouid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'ouidLogin',:event_time,:channel) " +
" on duplicate key update event_time=values(event_time)")
void insertOuidEvent(@BindBean AppsFlyerEvent event, @Bind("ouid") String ouid, @Bind("event_type") String event_type);
@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);
@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,'activation',:reg_time) ") "values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'activation',:reg_time) ")
...@@ -43,16 +50,17 @@ public interface GameAccountDao { ...@@ -43,16 +50,17 @@ public interface GameAccountDao {
@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_activation_devices (`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,'register',:reg_time ) ") " values ( :platform,:did ,:uid,:ouid, :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,@Bind("ouid") String ouid); void insertRegister(@BindBean AppsFlyerEvent value, @Bind("reg_time") String reg_time, @Bind("ouid") String ouid);
@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,'rolelogin',:event_time ) ") " values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'rolelogin',:event_time ) " +
" on duplicate key update event_time=values(event_time)")
void insertRoleLogin(@BindBean AppsFlyerEvent value); 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`, `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,'purchase',: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); // void insertPauchase(@BindBean AppsFlyerEvent value);
@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,'createRole',:event_time ) ") " values ( :platform,:did ,:uid, :region, :game_id, :media_source, :campaign_id,:ad_set_id, :ad_id, :time_zone,'createRole',:event_time ) ")
......
...@@ -24,10 +24,10 @@ public interface MetricsDao { ...@@ -24,10 +24,10 @@ public interface MetricsDao {
void insertActivationMetrics(@BindBean ActivationMetrics event); void insertActivationMetrics(@BindBean ActivationMetrics event);
@SqlUpdate("insert into appsflyer_retention " + // @SqlUpdate("insert into appsflyer_retention " +
" (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,`register_num`)" + // " (`data_date`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`currency`,`register_num`)" +
" values (:data_date,:region,:platform,:game_id,:media_source,:campaign_id,:ad_set_id,:ad_id,'KRW',:data_register)") // " values (:data_date,:region,:platform,:game_id,:media_source,:campaign_id,:ad_set_id,:ad_id,'KRW',:data_register)")
void insertRegisterNum(@BindBean ActivationMetrics event); // void insertRegisterNum(@BindBean ActivationMetrics event);
/** /**
......
...@@ -13,11 +13,11 @@ public interface OrderDao { ...@@ -13,11 +13,11 @@ public interface OrderDao {
" :pay_time, :revenue,:currency,:orderno,:reg_time,:activation_time,:time_zone,:channel,:data_type,:retargeting_time)") " :pay_time, :revenue,:currency,:orderno,:reg_time,:activation_time,:time_zone,:channel,:data_type,:retargeting_time)")
void insertOrder(@BindBean OrderData orderData); 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, " + // @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)" + // " 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," + // " 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)") // " :pay_time, :revenue,:currency,:orderno,:time_zone,:channel,:server_reg_date,:server_reg_hour,:reg_server_id)")
void insertServerOrder(@BindBean OrderData eventData); // void insertServerOrder(@BindBean OrderData eventData);
@SqlUpdate("insert into appsflyer_pay_record_ouid (region, did, uid,ouid,platform, game_id, media_source, campaign_id, ad_set_id,ad_id, " + @SqlUpdate("insert into appsflyer_pay_record_ouid (region, did, uid,ouid,platform, game_id, media_source, campaign_id, ad_set_id,ad_id, " +
" pay_time, money, currency, order_no, ouid_reg_time, time_zone ,channel,data_type)" + " pay_time, money, currency, order_no, ouid_reg_time, time_zone ,channel,data_type)" +
......
package com.fshark.overseas.advert.entity.mapper;
import com.fshark.overseas.advert.entity.GameAccount;
import org.jdbi.v3.core.mapper.RowMapper;
import org.jdbi.v3.core.statement.StatementContext;
import java.sql.ResultSet;
import java.sql.SQLException;
public class OuidGameAccountMapper implements RowMapper<GameAccount> {
@Override
public GameAccount map(ResultSet rs, StatementContext ctx) throws SQLException {
return GameAccount.builder()
.gameId(rs.getString("game_id"))
.did(rs.getString("did"))
.platform(rs.getString("platform"))
.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"))
.region(rs.getString("region"))
.channel(rs.getString("channel"))
.event_type(rs.getString("event_type"))
.event_time(rs.getString("event_time"))
.build();
}
}
...@@ -16,12 +16,12 @@ public enum EventTypeEnum { ...@@ -16,12 +16,12 @@ public enum EventTypeEnum {
第一次创角作为注册 第一次创角作为注册
*/ */
// REGISTER("createrole"), // REGISTER("createrole"),
CREATE_ROLE("rolelogin"), CREATE_ROLE("createrole"),
SDK_ACTIVATION("activation"), SDK_ACTIVATION("activation"),
SDK_REGISTER("register"), SDK_REGISTER("register"),
FAST_LOGIN("fast login"), FAST_LOGIN("fast login"),
GAME_LOGIN("gamelogin"), GAME_LOGIN("gamelogin"),
ROLE_LOGIN("login"), ROLE_LOGIN("rolelogin"),
LOGIN("login"), LOGIN("login"),
ACTIVATION("install"), ACTIVATION("install"),
PURCHASE("af_purchase"), PURCHASE("af_purchase"),
...@@ -72,7 +72,7 @@ public enum EventTypeEnum { ...@@ -72,7 +72,7 @@ public enum EventTypeEnum {
} }
public static boolean loginEvent(AppsFlyerEvent event) { public static boolean loginEvent(AppsFlyerEvent event) {
return LOGIN.getName().equals(event.getEvent()) && !event.getUid().equals(Constants.DEFAULT_FIELD_VALUE); return LOGIN.getName().equals(event.getEvent());
} }
public static boolean notHeartBeatEvent(AppsFlyerEvent event) { public static boolean notHeartBeatEvent(AppsFlyerEvent event) {
......
...@@ -145,16 +145,26 @@ public class OrderData implements Serializable { ...@@ -145,16 +145,26 @@ public class OrderData implements Serializable {
private String retargeting_time; private String retargeting_time;
private String reg_server_id; // private String reg_server_id;
private String reg_server_time; // private String reg_server_time;
private String server_reg_date; // private String server_reg_date;
private String server_reg_hour; // private String server_reg_hour;
private String ouid_reg_time; private String ouid_reg_time;
private String ouid_platform;
private String ouid_media_source;
private String ouid_campaign_id;
private String ouid_ad_set_id;
private String ouid_ad_id;
private int data_type; private int data_type;
......
...@@ -21,17 +21,21 @@ public class SinkMetricEvent { ...@@ -21,17 +21,21 @@ public class SinkMetricEvent {
public boolean isOrderEvent() { public boolean isOrderEvent() {
return Constants.TYPE_SINK_ORDER.equals(eventType); return Constants.TYPE_SINK_ORDER.equals(eventType) || Constants.TYPE_SINK_OUID_ORDER.equals(eventType);
} }
public boolean isEvent() { public boolean isEvent() {
return Constants.TYPE_SINK_EVENT.equals(eventType); return Constants.TYPE_SINK_EVENT.equals(eventType);
} }
public boolean isServer() {
return Constants.TYPE_SINK_SERVER.equals(eventType); public boolean isOuidEvent() {
return Constants.TYPE_SINK_OUID_METRIC.equals(eventType);
} }
public boolean isOuidOrderEvent() {
return Constants.TYPE_SINK_OUID_ORDER.equals(eventType);
}
} }
...@@ -6,15 +6,22 @@ import com.fshark.overseas.advert.util.AppConfig; ...@@ -6,15 +6,22 @@ import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants; import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson; 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.util.Collector; import org.apache.flink.util.Collector;
public class ActivitionMetricToStringProcess extends RichFlatMapFunction<ActivationMetrics, String> { public class ActivitionMetricToStringProcess extends RichFlatMapFunction<ActivationMetrics, String> {
public ActivitionMetricToStringProcess(AppConfig appConfig) { public ActivitionMetricToStringProcess(AppConfig appConfig) {
} }
private Gson gson;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
gson = new Gson();
}
@Override @Override
public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception { public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception {
Gson gson = new Gson();
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent(); SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value); sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_METRIC); sinkMetricEvent.setEventType(Constants.TYPE_SINK_METRIC);
......
...@@ -2,7 +2,12 @@ package com.fshark.overseas.advert.processor; ...@@ -2,7 +2,12 @@ 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.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.*;
...@@ -22,7 +27,7 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -22,7 +27,7 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
private JdbiContext jdbiContext; private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao; private GameAccountDao gameAccountDao;
private GameMappingDao gameMappingDao;
public ActivitionProcess(AppConfig appConfig) { public ActivitionProcess(AppConfig appConfig) {
this.appConfig = appConfig; this.appConfig = appConfig;
} }
...@@ -30,8 +35,9 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -30,8 +35,9 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
gameMappingDao = jdbiContext.getJdbi().onDemand(GameMappingDao.class);
} }
@Override @Override
...@@ -61,7 +67,6 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ ...@@ -61,7 +67,6 @@ public class ActivitionProcess extends RichFlatMapFunction<AppsFlyerEvent, Activ
} }
} }
} }
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) { private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
......
...@@ -40,7 +40,7 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E ...@@ -40,7 +40,7 @@ public class EventMectrisProcessor extends RichFlatMapFunction<AppsFlyerEvent, E
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
ValueStateDescriptor<Long> descriptor = ValueStateDescriptor<Long> descriptor =
new ValueStateDescriptor<>( new ValueStateDescriptor<>(
......
...@@ -12,7 +12,9 @@ import org.apache.flink.api.common.functions.RichFlatMapFunction; ...@@ -12,7 +12,9 @@ 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.util.Locale; import java.util.Locale;
import java.util.Optional; import java.util.Optional;
...@@ -33,7 +35,7 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -33,7 +35,7 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
} }
...@@ -45,19 +47,61 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -45,19 +47,61 @@ public class OuidLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
return; return;
} }
ActivationMetrics activationMetrics = new ActivationMetrics(); ActivationMetrics activationMetrics = new ActivationMetrics();
ActivationMetrics retentMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics); copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid()); copyBaseMetrics(value, retentMetrics);
retentMetrics.setUid(ouid);
activationMetrics.setUid(ouid);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time()); String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
Optional<GameAccount> account = gameAccountDao.findByOuid(value,ouid); activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
Optional<GameAccount> account = gameAccountDao.findByOuid(value, ouid, Constants.TYPE_OUID_REGISTER);
if (!account.isPresent()) { if (!account.isPresent()) {
gameAccountDao.insertOuidRegister(value, event_time,ouid); gameAccountDao.insertOuidRegister(value, event_time, ouid);
activationMetrics.setData_ouid_register(1L); activationMetrics.setData_ouid_register(1L);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time))); } else {
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time))); GameAccount gameAccount = account.get();
out.collect(activationMetrics); copyOuidMetrics(gameAccount, activationMetrics);
copyOuidMetrics(gameAccount, retentMetrics);
String activationTime = account.get().getEvent_time();
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
retentMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
retentMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByOuid(value, ouid, Constants.TYPE_OUID_LOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_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) {
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
retentMetrics.setDateRemind(daysBetween); //留存天数
activationMetrics.setData_active_day_dau(1L);
}
}
} }
out.collect(retentMetrics);
out.collect(activationMetrics);
gameAccountDao.insertOuidEvent(value, ouid, Constants.TYPE_OUID_LOGIN);
} }
private void copyOuidMetrics(GameAccount gameAccount, ActivationMetrics activationMetrics) {
activationMetrics.setPlatform(gameAccount.getPlatform());
activationMetrics.setMedia_source(gameAccount.getMediaSource());
activationMetrics.setCampaign_id(gameAccount.getCampaign_id());
activationMetrics.setAd_set_id(gameAccount.getAd_set_id());
activationMetrics.setAd_id(gameAccount.getAd_id());
activationMetrics.setChannel(gameAccount.getChannel());
activationMetrics.setRegion(gameAccount.getRegion());
}
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) { private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
metrics.setTime_zone(value.getTime_zone()); metrics.setTime_zone(value.getTime_zone());
metrics.setPlatform(value.getPlatform()); metrics.setPlatform(value.getPlatform());
......
...@@ -23,7 +23,7 @@ import org.apache.http.util.TextUtils; ...@@ -23,7 +23,7 @@ import org.apache.http.util.TextUtils;
import java.util.Optional; import java.util.Optional;
public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> { public class OuidOrderDataProcessor extends RichFlatMapFunction<AppsFlyerEvent, String> {
private AppConfig appConfig; private AppConfig appConfig;
...@@ -38,7 +38,7 @@ public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> { ...@@ -38,7 +38,7 @@ public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> {
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
orderDao = jdbiContext.getJdbi().onDemand(OrderDao.class); orderDao = jdbiContext.getJdbi().onDemand(OrderDao.class);
} }
...@@ -50,34 +50,37 @@ public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> { ...@@ -50,34 +50,37 @@ public class OuidOrderDataProcessor extends RichSinkFunction<AppsFlyerEvent> {
if (jdbiContext != null) { if (jdbiContext != null) {
jdbiContext.getDataSource().close(); jdbiContext.getDataSource().close();
} }
} }
@Override @Override
public void invoke(AppsFlyerEvent value, Context context) throws Exception { public void flatMap(AppsFlyerEvent value, Collector<String> out) throws Exception {
super.invoke(value, context); JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value());
JSONObject jsonObject = JSONObject.parseObject(value.getEvent_value()); String ouid = jsonObject.getString("ouid");
String ouid = jsonObject.getString("ouid"); AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class); if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null || TextUtils.isEmpty(ouid)) { //不是我们上报的格式
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null || TextUtils.isEmpty(ouid)) { //不是我们上报的格式 return;
return; }
} if ("ios".equalsIgnoreCase(value.getPlatform())) {
if ("ios".equalsIgnoreCase(value.getPlatform())) { String bundleId = appsflyerPurchase.getBundle_id();
String bundleId = appsflyerPurchase.getBundle_id(); value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId); }
} OrderData orderData = new OrderData(value);
OrderData orderData = new OrderData(value); orderData.setOuid(ouid);
orderData.setOuid(ouid); Optional<GameAccount> ouidRegister = gameAccountDao.findByOuid(value, ouid, Constants.TYPE_OUID_REGISTER);
Optional<GameAccount> ouidRegister = gameAccountDao.findByOuid(value, ouid); if (ouidRegister.isPresent()) {
if (ouidRegister.isPresent()) { GameAccount gameAccount = ouidRegister.get();
GameAccount gameAccount = ouidRegister.get(); orderData.setMedia_source(gameAccount.getMediaSource());
orderData.setMedia_source(gameAccount.getMediaSource()); orderData.setCampaign_id(gameAccount.getCampaign_id());
orderData.setCampaign_id(gameAccount.getCampaign_id()); orderData.setAd_set_id(gameAccount.getAd_set_id());
orderData.setAd_set_id(gameAccount.getAd_set_id()); orderData.setAd_id(gameAccount.getAd_id());
orderData.setAd_id(gameAccount.getAd_id()); orderData.setOuid_reg_time(gameAccount.getEvent_time());
orderData.setOuid_reg_time(gameAccount.getEvent_time()); orderData.setChannel(gameAccount.getChannel());
orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())), orderData.getMedia_source(), orderData.getData_type())); orderData.setRegion(gameAccount.getRegion());
} orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())), orderData.getMedia_source(), orderData.getData_type()));
orderDao.insertOuidOrder(orderData); }
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_OUID_ORDER);
sinkMetricEvent.setOrderData(orderData);
out.collect(new Gson().toJson(sinkMetricEvent));
} }
} }
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.http.util.TextUtils;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationMetrics, Object, TimeWindow> {
private AppConfig appConfig;
private Connection conn;
public OuidStoreMetricProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
Class.forName("com.mysql.cj.jdbc.Driver");
} else {
Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
}
conn = DriverManager.getConnection(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
@Override
public void process(Context context, Iterable<ActivationMetrics> elements, Collector<Object> out) throws Exception {
//详细指标写入
insertMetricsDetail(elements);
}
private void insertMetricsDetail(Iterable<ActivationMetrics> elements) {
try {
PreparedStatement insertStmt = conn.prepareStatement("insert into appsflyer_metrics_ouid_detail" +
" (`uid`,`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_type`,`metric_name`,`metric_value`)" +
" values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
for (ActivationMetrics element : elements) {
String name = dealMetrics(element);
if (!TextUtils.isEmpty(name)) {
insertStmt.setString(1, element.getUid());
insertStmt.setLong(2, element.getData_date());
insertStmt.setLong(3, element.getData_hour());
insertStmt.setString(4, element.getTime_zone());
insertStmt.setString(5, element.getRegion());
insertStmt.setString(6, element.getPlatform());
insertStmt.setString(7, element.getGame_id());
insertStmt.setString(8, element.getMedia_source());
insertStmt.setString(9, element.getCampaign_id());
insertStmt.setString(10, element.getAd_set_id());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getChannel());
insertStmt.setLong(13, element.getData_type());
insertStmt.setString(14, name);
insertStmt.setLong(15, 1l);
insertStmt.addBatch();
}
}
insertStmt.executeBatch(); //批量执行
conn.commit();
insertStmt.close();
} catch (Exception e) {
e.printStackTrace();
}
}
private String dealMetrics(ActivationMetrics element) {
if (element.getData_activation() > 0) {
return "data_activation";
}
if (element.getData_register() > 0) {
return "data_register";
}
if (element.getData_device_register() > 0) {
return "data_device_register";
}
if (element.getData_create_role() > 0) {
return "data_create_role";
}
if (element.getData_day_act_register() > 0) {
return "data_day_act_register";
}
if (element.getData_day_register_create() > 0) {
return "data_day_register_create";
}
if (element.getData_active_day_dau() > 0) {
return "data_active_day_dau";
}
if (element.getData_sdk_activation() > 0) {
return "data_sdk_activition";
}
if (element.getData_sdk_register() > 0) {
return "data_sdk_register";
}
if (element.getDateRemind() > 0) {
return "retention_day" + element.getDateRemind();
}
if (element.getData_ouid_register() > 0) {
return "ouid_role";
}
return null;
}
@Override
public void close() throws Exception {
super.close();
if (conn != null) {
conn.close();
}
}
}
...@@ -31,9 +31,9 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE ...@@ -31,9 +31,9 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
serverRoleDao = jdbiContext.getJdbi().onDemand(ServerRoleDao.class); // serverRoleDao = jdbiContext.getJdbi().onDemand(ServerRoleDao.class);
} }
@Override @Override
...@@ -71,14 +71,15 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE ...@@ -71,14 +71,15 @@ public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerE
orderData.setRetargeting_time(gameAccount.getEvent_time()); orderData.setRetargeting_time(gameAccount.getEvent_time());
} }
Optional<ServerRoleAccount> serverAccount = serverRoleDao.findByPrimary(value, appsflyerPurchase.getServerid(), Constants.TYPE_ACTIVIYION); //todo 替换为ouid的逻辑
if (serverAccount.isPresent()) { // Optional<ServerRoleAccount> serverAccount = serverRoleDao.findByPrimary(value, appsflyerPurchase.getServerid(), Constants.TYPE_ACTIVIYION);
String event_time = serverAccount.get().getEvent_time(); // if (serverAccount.isPresent()) {
orderData.setReg_server_id(appsflyerPurchase.getServerid()); // String event_time = serverAccount.get().getEvent_time();
orderData.setReg_server_time(event_time); // orderData.setReg_server_id(appsflyerPurchase.getServerid());
} // orderData.setReg_server_time(event_time);
// }
gameAccountDao.insertPauchase(value); // gameAccountDao.insertPauchase(value);
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent(); SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_ORDER); sinkMetricEvent.setEventType(Constants.TYPE_SINK_ORDER);
......
...@@ -32,7 +32,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa ...@@ -32,7 +32,7 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
} }
......
...@@ -31,7 +31,7 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe ...@@ -31,7 +31,7 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class); gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
} }
...@@ -41,18 +41,14 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe ...@@ -41,18 +41,14 @@ public class SdkProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMe
copyBaseMetrics(value, activationMetrics); copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid()); activationMetrics.setUid(value.getUid());
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time()); String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_REGISTER);
if (EventTypeEnum.SDK_REGISTER.getName().equals(value.getEvent())){ if (!account.isPresent()) {
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_SDK_REGISTER); gameAccountDao.insertSdkRegister(value, event_time);
if (!account.isPresent()){ activationMetrics.setData_sdk_register(1L);
gameAccountDao.insertSdkRegister(value, event_time); activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_sdk_register(1L); activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time))); out.collect(activationMetrics);
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
out.collect(activationMetrics);
}
} }
} }
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) { private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
......
...@@ -3,6 +3,8 @@ package com.fshark.overseas.advert.sink; ...@@ -3,6 +3,8 @@ package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.MetricsDao; import com.fshark.overseas.advert.entity.dao.MetricsDao;
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.RemainMetrics;
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.DatabaseUtils; import com.fshark.overseas.advert.util.DatabaseUtils;
...@@ -29,15 +31,43 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> { ...@@ -29,15 +31,43 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> {
+ value.getData_day_register_create() + value.getData_active_day_dau() + value.getData_sdk_activation() + value.getData_sdk_register() > 0) { + value.getData_day_register_create() + value.getData_active_day_dau() + value.getData_sdk_activation() + value.getData_sdk_register() > 0) {
metricsDao.insertActivationMetrics(value); metricsDao.insertActivationMetrics(value);
} }
if (value.getData_register() > 0){ // if (value.getData_register() > 0){
metricsDao.insertRegisterNum(value); // metricsDao.insertRegisterNum(value);
// }
//短留存写入
if (value.getDateRemind() == 0L) {
return;
}
RemainMetrics remainMetrics = new RemainMetrics();
copyBaseMetrics(value, remainMetrics);
remainMetrics.setDays(value.getDateRemind());
remainMetrics.setUser_num(1L);
//短线留存
insertRemain(remainMetrics);
}
private void insertRemain(RemainMetrics value) {
Long days = value.getDays();
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;
});
} }
} }
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){ if (appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else { } else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
...@@ -52,4 +82,19 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> { ...@@ -52,4 +82,19 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> {
jdbiContext.getDataSource().close(); jdbiContext.getDataSource().close();
} }
} }
private void copyBaseMetrics(ActivationMetrics value, BaseMetrics metrics) {
metrics.setData_date(value.getData_date());
metrics.setData_hour(value.getData_hour());
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.setData_type(value.getData_type());
metrics.setChannel(value.getChannel());
}
} }
...@@ -2,6 +2,7 @@ package com.fshark.overseas.advert.sink; ...@@ -2,6 +2,7 @@ package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.modle.AppsFlyerEvent; 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.DateUtils; import com.fshark.overseas.advert.util.DateUtils;
import lombok.val; import lombok.val;
import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.Configuration;
...@@ -27,6 +28,11 @@ public class EventInfoSinkProcess extends ProcessAllWindowFunction<AppsFlyerEven ...@@ -27,6 +28,11 @@ public class EventInfoSinkProcess extends ProcessAllWindowFunction<AppsFlyerEven
@Override @Override
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)) {
Class.forName("com.mysql.cj.jdbc.Driver");
} else {
Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
}
conn = DriverManager.getConnection(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword()); conn = DriverManager.getConnection(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} }
......
...@@ -36,14 +36,9 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> { ...@@ -36,14 +36,9 @@ public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())) { if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())) {
OrderData eventData = value.getOrderData(); OrderData eventData = value.getOrderData();
orderDao.insertOrder(eventData); orderDao.insertOrder(eventData);
if (!StringUtils.isBlank(eventData.getReg_server_time())) { } else if (Constants.TYPE_SINK_OUID_ORDER.equals(value.getEventType())){
eventData.setServer_reg_date(DateUtils.formatAsYMD(eventData.getReg_server_time())); OrderData eventData = value.getOrderData();
eventData.setServer_reg_hour(DateUtils.formatAsHH(eventData.getReg_server_time())); orderDao.insertOuidOrder(eventData);
} else {
eventData.setServer_reg_date("-1");
eventData.setServer_reg_hour("-1");
}
orderDao.insertServerOrder(eventData);
} }
} }
......
...@@ -72,8 +72,10 @@ public class Constants { ...@@ -72,8 +72,10 @@ public class Constants {
public static final String TYPE_ACTIVIYION = "activation"; public static final String TYPE_ACTIVIYION = "activation";
public static final String TYPE_RETARGETING = "retargeting"; public static final String TYPE_RETARGETING = "retargeting";
public static final String TYPE_REGISTER = "register"; public static final String TYPE_REGISTER = "register";
public static final String TYPE_OUID_REGISTER = "ouidRegister";
public static final String TYPE_CREATEROLE = "createRole"; public static final String TYPE_CREATEROLE = "createRole";
public static final String TYPE_ROLELOGIN = "rolelogin"; public static final String TYPE_ROLELOGIN = "rolelogin";
public static final String TYPE_OUID_LOGIN = "ouidLogin";
public static final String TYPE_PAUCHASE = "purchase"; public static final String TYPE_PAUCHASE = "purchase";
public static final String TYPE_SDK_ACTIVATION = "sdkActivation"; public static final String TYPE_SDK_ACTIVATION = "sdkActivation";
public static final String TYPE_SDK_REGISTER = "sdkRegister"; public static final String TYPE_SDK_REGISTER = "sdkRegister";
...@@ -82,7 +84,9 @@ public class Constants { ...@@ -82,7 +84,9 @@ public class Constants {
public static final String TYPE_SINK_METRIC="metric"; public static final String TYPE_SINK_METRIC="metric";
public static final String TYPE_SINK_OUID_METRIC="ouidMetric";
public static final String TYPE_SINK_ORDER="order"; public static final String TYPE_SINK_ORDER="order";
public static final String TYPE_SINK_OUID_ORDER="ouidOrder";
public static final String TYPE_SINK_EVENT="event"; public static final String TYPE_SINK_EVENT="event";
public static final String TYPE_SINK_SERVER="server"; public static final String TYPE_SINK_SERVER="server";
......
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