Commit 7abb7c42 authored by sheteng's avatar sheteng

将指标写入kafka

然后从kafka读取,分别写入mysql和clickhouse
parent 76d18631
#Thu Jul 22 10:20:45 CST 2021
#Wed Aug 18 16:25:38 CST 2021
gradle.version=6.8
......@@ -55,7 +55,7 @@ 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'
compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-java:${flinkVersion}"
......
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/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
com/fshark/overseas/advert/entity/dao/MetricsDetailDao.java
com.fshark.overseas.advert.entity.dao.MetricsDetailDao
com/fshark/overseas/advert/entity/RegisterAccount.java
com.fshark.overseas.advert.entity.RegisterAccount
com.fshark.overseas.advert.entity.RegisterAccount$RegisterAccountBuilder
com/fshark/overseas/advert/processor/ActivationAggregateFunction.java
com.fshark.overseas.advert.processor.ActivationAggregateFunction
com/fshark/overseas/advert/modle/mapping/Media.java
com.fshark.overseas.advert.modle.mapping.Media
com.fshark.overseas.advert.modle.mapping.Media$MediaBuilder
com/fshark/overseas/advert/entity/dao/OrderDao.java
com.fshark.overseas.advert.entity.dao.OrderDao
com/fshark/overseas/advert/entity/mapper/RegisterAccountMapper.java
com.fshark.overseas.advert.entity.mapper.RegisterAccountMapper
com/fshark/overseas/advert/entity/GameAccount.java
com.fshark.overseas.advert.entity.GameAccount
com.fshark.overseas.advert.entity.GameAccount$GameAccountBuilder
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/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/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
com.fshark.overseas.advert.util.JdbiContext
com/fshark/overseas/advert/processor/GameAdMappingProcessor.java
com.fshark.overseas.advert.processor.GameAdMappingProcessor
com/fshark/overseas/advert/util/RedisClientUtils.java
com.fshark.overseas.advert.util.RedisClientUtils
com/fshark/overseas/advert/processor/ActivitionProcess.java
com.fshark.overseas.advert.processor.ActivitionProcess
com/fshark/overseas/advert/processor/StoreMetricProcess.java
com.fshark.overseas.advert.processor.StoreMetricProcess
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/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
com.fshark.overseas.advert.processor.ActivationResultWindows
com/fshark/overseas/advert/util/DateUtils.java
com.fshark.overseas.advert.util.DateUtils
com/fshark/overseas/advert/util/Constants.java
com.fshark.overseas.advert.util.Constants
com/fshark/overseas/advert/processor/RemainProcess.java
com.fshark.overseas.advert.processor.RemainProcess
com/fshark/overseas/advert/modle/AppsflyerPurchase.java
com.fshark.overseas.advert.modle.AppsflyerPurchase
com/fshark/overseas/advert/modle/EventTypeEnum.java
com.fshark.overseas.advert.modle.EventTypeEnum
com/fshark/overseas/advert/util/DateTimeUtils.java
com.fshark.overseas.advert.util.DateTimeUtils
com/fshark/overseas/advert/entity/mapper/GameAccountMapper.java
com.fshark.overseas.advert.entity.mapper.GameAccountMapper
com/fshark/overseas/advert/processor/RoleLoginProcess.java
com.fshark.overseas.advert.processor.RoleLoginProcess
com/fshark/overseas/advert/modle/metrics/RemainMetrics.java
com.fshark.overseas.advert.modle.metrics.RemainMetrics
com/fshark/overseas/advert/entity/dao/GameAccountDao.java
com.fshark.overseas.advert.entity.dao.GameAccountDao
com/fshark/overseas/advert/modle/mapping/Game.java
com.fshark.overseas.advert.modle.mapping.Game
com/fshark/overseas/advert/util/ClickFeedbackRecord.java
com.fshark.overseas.advert.util.ClickFeedbackRecord
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/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
......@@ -10,6 +10,9 @@ import com.fshark.overseas.advert.sink.ActicitionSink;
import com.fshark.overseas.advert.sink.PurchaseSink;
import com.fshark.overseas.advert.sink.RemainSink;
import com.fshark.overseas.advert.util.AppConfig;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
......@@ -19,6 +22,8 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
......@@ -35,6 +40,9 @@ public class AppsflyerAnalyze {
appConfig.setSourceKafkaGroupId(parameters.get("source-kafka-group-id"));
appConfig.setSourceKafkaTopic(parameters.get("source-kafka-topic"));
appConfig.setSinkKafkaBootstrapServers(parameters.get("sink-kafka-bootstrap-servers"));
appConfig.setSinkKafkaTopic(parameters.get("sink-kafka-topic"));
appConfig.setDbHost(parameters.get("jdbc-host"));
appConfig.setDbUser(parameters.get("jdbc-user"));
appConfig.setDbPassword(parameters.get("jdbc-password"));
......@@ -44,6 +52,9 @@ public class AppsflyerAnalyze {
" --source-kafka-bootstrap-servers <source-kafka-bootstrap-servers>" +
" --source-kafka-group-id <source-kafka-group-id>" +
" --source-kafka-topic <source-kafka-topic>" +
" --sink-kafka-bootstrap-servers <sink-kafka-bootstrap-servers>" +
" --sink-kafka-group-id <sink-kafka-group-id>" +
" --sink-kafka-topic <sink-kafka-topic>" +
" --jdbc-host <jdbc-host> " +
" --jdbc-user <jdbc-user> " +
" --jdbc-password <jdbc-password> ");
......@@ -66,15 +77,18 @@ public class AppsflyerAnalyze {
properties.setProperty("auto.offset.reset", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
// TODO: 2021/8/10 设置消费开始的时间戳1628524944000
dataConsumer.setStartFromTimestamp(1628524944000L);
dataConsumer.setStartFromTimestamp(1629129773000L);
Properties props2 = new Properties();
props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers());
FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(appConfig.getSinkKafkaTopic(), new SimpleStringSchema(), props2);
DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer)
.flatMap(new ParseEventProcessor())
.uid("parse-event")
.name("parse-event")
.shuffle();
// adEventDataStream.print();
//广告指标(激活)
DataStream<ActivationMetrics> activitionMetricStream = adEventDataStream
.filter(EventTypeEnum::installEvent)
......@@ -96,41 +110,23 @@ public class AppsflyerAnalyze {
.uid("role-login-event")
.name("role-login-event");
//角色注册激活预聚合
DataStream<ActivationMetrics> activitionResult = activitionMetricStream
.union(loginRoleMetricStream, createRoleMetricStream)
.keyBy(ActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.aggregate(new ActivationAggregateFunction(), new ActivationResultWindows())
.uid("activation-agg-event")
.name("activation-agg-event");
// 角色注册激活写入
activitionResult.addSink(new ActicitionSink(appConfig))
.uid("login-sink-event")
.name("login-sink-event");
//
// //指标详细记录
activitionMetricStream.union(loginRoleMetricStream, createRoleMetricStream)
.timeWindowAll(Time.minutes(1)).process(new StoreMetricProcess(appConfig))
.uid("metric-detail")
.name("metric-detail");
//// 留存预聚合
DataStream<RemainMetrics> remainResult = loginRoleMetricStream.flatMap(new RemainProcess())
.keyBy(RemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("user_num")
.uid("remain-event")
.name("remain-event");
//留存写入
remainResult.addSink(new RemainSink(appConfig))
.uid("remain-sink-event")
.name("remain-sink-event");
// 订单信息
//日活
DataStream<ActivationMetrics> dauMetricStream = adEventDataStream
.filter(EventTypeEnum::roleLoginEvent)
.flatMap(new DauProcess(appConfig))
.uid("dau-event")
.name("dau-event");
activitionMetricStream.union(loginRoleMetricStream, createRoleMetricStream,dauMetricStream)
.flatMap(new ActivitionMetricToStringProcess(appConfig))
.addSink(kafkaSink)
.uid("sink-metric-event")
.name("sink-metric-event");
// 订单信息
adEventDataStream.filter(EventTypeEnum::orderEvent)
.addSink(new PurchaseSink(appConfig))
.flatMap(new PurchaseToOrderDataProcessor(appConfig))
.addSink(kafkaSink)
.uid("order-event")
.name("order-event");
......
package com.fshark.overseas.advert;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.processor.*;
import com.fshark.overseas.advert.sink.ActicitionSink;
import com.fshark.overseas.advert.sink.PurchaseSink;
import com.fshark.overseas.advert.sink.RemainSink;
import com.fshark.overseas.advert.util.AppConfig;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class AppsflyerMetricStore {
static AppConfig loadConfigFromArgs(String[] args) throws Exception {
ParameterTool parameters = ParameterTool.fromArgs(args);
AppConfig appConfig = new AppConfig();
try {
appConfig.setSourceKafkaBootstrapServers(parameters.get("source-kafka-bootstrap-servers"));
appConfig.setSourceKafkaGroupId(parameters.get("source-kafka-group-id"));
appConfig.setSourceKafkaTopic(parameters.get("source-kafka-topic"));
appConfig.setDbHost(parameters.get("jdbc-host"));
appConfig.setDbUser(parameters.get("jdbc-user"));
appConfig.setDbPassword(parameters.get("jdbc-password"));
appConfig.setSinkType(parameters.get("sink-type"));
} catch (Exception e) {
System.err.println("Place set args: " +
" --source-kafka-bootstrap-servers <source-kafka-bootstrap-servers>" +
" --source-kafka-group-id <source-kafka-group-id>" +
" --source-kafka-topic <source-kafka-topic>" +
" --jdbc-host <jdbc-host> " +
" --jdbc-user <jdbc-user> " +
" --jdbc-password <jdbc-password> " +
" --sink-type <sink-type>");
throw e;
}
return appConfig;
}
public static void main(String[] args) throws Exception {
AppConfig appConfig = loadConfigFromArgs(args);
final StreamExecutionEnvironment sse = StreamExecutionEnvironment.getExecutionEnvironment();
sse.setParallelism(1);
sse.enableCheckpointing(1000 * 60 * 10);
sse.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS)));
//source设置
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", appConfig.getSourceKafkaBootstrapServers());
properties.setProperty("group.id", appConfig.getSourceKafkaGroupId());
properties.setProperty("auto.offset.reset", "latest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
DataStream<SinkMetricEvent> activationMetricsDataStream = sse.addSource(dataConsumer).flatMap(new FlatMapFunction<String, SinkMetricEvent>() {
@Override
public void flatMap(String value, Collector<SinkMetricEvent> out) throws Exception {
out.collect(new Gson().fromJson(value, SinkMetricEvent.class));
}
}).uid("parse-event").name("parse-event");
// activationMetricsDataStream.print();
//角色注册激活预聚合
DataStream<ActivationMetrics> activitionResult = activationMetricsDataStream
.flatMap(new MetricJsonToObProcessor())
.keyBy(ActivationMetrics::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.aggregate(new ActivationAggregateFunction(), new ActivationResultWindows())
.uid("activation-agg-event")
.name("activation-agg-event");
// 角色注册激活写入
activitionResult.addSink(new ActicitionSink(appConfig))
.uid("login-sink-event")
.name("login-sink-event");
//
// //指标详细记录
activationMetricsDataStream.timeWindowAll(Time.minutes(1)).process(new StoreMetricProcess(appConfig))
.uid("metric-detail")
.name("metric-detail");
// 留存预聚合
DataStream<RemainMetrics> remainResult = activationMetricsDataStream.flatMap(new RemainProcess())
.keyBy(RemainMetrics::getRemind_key)
.window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
.sum("user_num")
.uid("remain-event")
.name("remain-event");
//留存写入
remainResult
.addSink(new RemainSink(appConfig))
.uid("remain-sink-event")
.name("remain-sink-event");
//订单写入
activationMetricsDataStream
.filter(SinkMetricEvent::isOrderEvent)
.addSink(new PurchaseSink(appConfig))
.uid("order-sink-event")
.name("order-sink-event");
sse.execute("AppsflyerMetricStore-analyze");
}
}
......@@ -12,6 +12,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Objects;
......@@ -51,7 +53,7 @@ public class AppsFlyerEvent implements Serializable {
/**
* 日志时间(ms)
*/
private Date event_time;
private LocalDateTime event_time;
/**
* json event_value
......@@ -414,7 +416,7 @@ public class AppsFlyerEvent implements Serializable {
appsFlyerEvent.setEvent_type(eventType);
String[] timeZoneParts = eventTimeWithZone.split("\\+");
Date eventTime = DateUtils.formatHMSSAsDate(timeZoneParts[0]);
LocalDateTime eventTime = DateUtils.formatHMSSAsDate(timeZoneParts[0]);
// 如果有错,直接视为错误; +0900
String timeZone = eventTimeWithZone.substring(23);
......
......@@ -41,7 +41,10 @@ public enum EventTypeEnum {
}
public static boolean roleLoginEvent(AppsFlyerEvent event) {
return ROLE_LOGIN.getName().equals(event.getEvent());
return ROLE_LOGIN.getName().equals(event.getEvent())
|| FAST_LOGIN.getName().equals(event.getEvent())
|| GAME_LOGIN.getName().equals(event.getEvent())
|| SDK_LOGIN.getName().equals(event.getEvent());
}
public static boolean createRoleEvent(AppsFlyerEvent event) {
......
......@@ -7,6 +7,8 @@ import com.fshark.overseas.advert.util.DateUtils;
import lombok.Data;
import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
......@@ -25,7 +27,7 @@ public class OrderData implements Serializable {
/**
* 日志时间(ms)
*/
private Date event_time;
private LocalDateTime event_time;
/**
* json event_value
......@@ -138,15 +140,11 @@ public class OrderData implements Serializable {
*/
private String activation_time;
/**
* 再营销时间
*/
private String retargeting_time;
private String reg_server_id;
private int data_hour;
private String reg_server_time;
private Long reg_date;
private int data_type;
......
package com.fshark.overseas.advert.modle;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.Constants;
import lombok.Data;
@Data
public class SinkMetricEvent {
private String eventType;
private ActivationMetrics eventData;
private OrderData orderData;
public boolean isMetricEvent() {
return Constants.TYPE_SINK_EVENT.equals(eventType);
}
public boolean isOrderEvent() {
return Constants.TYPE_SINK_ORDER.equals(eventType);
}
}
package com.fshark.overseas.advert.modle.metrics;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.io.Serializable;
@EqualsAndHashCode(callSuper=false)
@Data
public class ActivationMetrics extends BaseMetrics {
public class ActivationMetrics extends BaseMetrics implements Serializable {
private String uid;
/**
......
......@@ -21,7 +21,7 @@ public class RemainMetrics extends BaseMetrics {
private Long user_num;
public String getRemind_key() {
return String.format("%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s",getData_date(),getTime_zone(),getRegion(),getMedia_source(),getPlatform(),getGame_id(),getCampaign_id(),getAd_set_id(),getAd_id(),getChannel(),getDays());
return String.format("%s|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s",getData_date(),getData_hour(),getTime_zone(),getPlatform(),getMedia_source(),getGame_id(),getCampaign_id(),getAd_set_id(),getAd_id(),getRegion(),getChannel(),getData_type(),getDays());
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
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.RichFlatMapFunction;
import org.apache.flink.util.Collector;
public class ActivitionMetricToStringProcess extends RichFlatMapFunction<ActivationMetrics, String> {
public ActivitionMetricToStringProcess(AppConfig appConfig) {
}
@Override
public void flatMap(ActivationMetrics value, Collector<String> out) throws Exception {
Gson gson = new Gson();
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventData(value);
sinkMetricEvent.setEventType(Constants.TYPE_SINK_EVENT);
out.collect(gson.toJson(sinkMetricEvent));
}
}
......@@ -13,6 +13,8 @@ import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
import java.util.Optional;
......@@ -41,7 +43,7 @@ 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;
LocalDateTime activition;
if (!account.isPresent()) { // 设备未激活
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
......
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.util.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
import java.util.Optional;
public class DauProcess extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
public DauProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
ActivationMetrics activationMetrics = new ActivationMetrics();
copyBaseMetrics(value, activationMetrics);
activationMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(event_time)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(event_time)));
String activationTime = event_time;
if (!account.isPresent()) { // 设备未激活
gameAccountDao.insertActivition(value, event_time);
activationMetrics.setData_activation(1L);
} else {
GameAccount gameAccount = account.get();
activationTime = gameAccount.getEvent_time();
}
CommonUtils.setType(activationMetrics);
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_time();
if (!StringUtils.isBlank(last_login_time)) {
LocalDateTime date = DateUtils.formatAsDate(last_login_time);
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setData_active_day_dau(1L);
}
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setData_active_day_dau(1L);
}
}
out.collect(activationMetrics);
}
private void copyBaseMetrics(AppsFlyerEvent value, BaseMetrics metrics) {
metrics.setTime_zone(value.getTime_zone());
metrics.setPlatform(value.getPlatform());
metrics.setMedia_source(value.getMedia_source());
metrics.setGame_id(value.getGame_id());
metrics.setCampaign_id(value.getCampaign_id());
metrics.setAd_set_id(value.getAd_set_id());
metrics.setAd_id(value.getAd_id());
metrics.setRegion(value.getRegion());
metrics.setChannel(value.getChannel());
metrics.setData_type(1);
if (value.getIs_retargeting()) { //再营销数据
metrics.setData_type(2);
}
if (value.getCampaign().toLowerCase(Locale.ROOT).startsWith("acp")) { //预注册数据
metrics.setData_type(3);
}
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.util.Constants;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;
public class MetricJsonToObProcessor extends RichFlatMapFunction<SinkMetricEvent, ActivationMetrics> {
@Override
public void flatMap(SinkMetricEvent value, Collector<ActivationMetrics> out) throws Exception {
if (Constants.TYPE_SINK_EVENT.equals(value.getEventType()) && value.getEventData() != null) {
out.collect(value.getEventData());
}
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.GameAccount;
import com.fshark.overseas.advert.entity.dao.GameAccountDao;
import com.fshark.overseas.advert.entity.dao.OrderDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsflyerPurchase;
import com.fshark.overseas.advert.modle.OrderData;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.util.Optional;
public class PurchaseToOrderDataProcessor extends RichFlatMapFunction<AppsFlyerEvent, String> {
private AppConfig appConfig;
private JdbiContext jdbiContext;
private GameAccountDao gameAccountDao;
public PurchaseToOrderDataProcessor(AppConfig appConfig) {
this.appConfig = appConfig;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<String> out) throws Exception {
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null) { //不是我们上报的格式
return;
}
if ("ios".equalsIgnoreCase(value.getPlatform())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
OrderData orderData = new OrderData(value);
Optional<GameAccount> accountActivition = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (accountActivition.isPresent()) {
GameAccount gameAccount = accountActivition.get();
orderData.setActivation_time(gameAccount.getEvent_time());
orderData.setData_type(CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())),orderData.getMedia_source(),orderData.getData_type()));
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (accountRegister.isPresent()) {
GameAccount gameAccount = accountRegister.get();
orderData.setReg_time(gameAccount.getEvent_time());
}
gameAccountDao.insertPauchase(value);
SinkMetricEvent sinkMetricEvent = new SinkMetricEvent();
sinkMetricEvent.setEventType(Constants.TYPE_SINK_ORDER) ;
sinkMetricEvent.setOrderData(orderData);
out.collect(new Gson().toJson(sinkMetricEvent));
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
......@@ -2,27 +2,33 @@ package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.modle.metrics.BaseMetrics;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.util.Constants;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.util.Collector;
/**
* 留存指标
*/
public class RemainProcess extends RichFlatMapFunction<ActivationMetrics, RemainMetrics> {
public class RemainProcess extends RichFlatMapFunction<SinkMetricEvent, RemainMetrics> {
@Override
public void flatMap(ActivationMetrics value, Collector<RemainMetrics> out) throws Exception {
if (value.getDateRemind() == 0L) {
return;
public void flatMap(SinkMetricEvent value, Collector<RemainMetrics> out) throws Exception {
if (Constants.TYPE_SINK_EVENT.equals(value.getEventType())) {
ActivationMetrics eventData = value.getEventData();
if (eventData.getDateRemind() == 0L) {
return;
}
RemainMetrics remainMetrics = new RemainMetrics();
copyBaseMetrics(eventData, remainMetrics);
remainMetrics.setDays(eventData.getDateRemind());
remainMetrics.setUser_num(1L);
out.collect(remainMetrics);
}
RemainMetrics remainMetrics = new RemainMetrics();
copyBaseMetrics(value, remainMetrics);
remainMetrics.setDays(value.getDateRemind());
remainMetrics.setUser_num(1L);
out.collect(remainMetrics);
}
private void copyBaseMetrics(ActivationMetrics value, BaseMetrics metrics) {
......
......@@ -13,6 +13,8 @@ import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Locale;
import java.util.Optional;
......@@ -41,60 +43,43 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
activationMetrics.setUid(value.getUid());
Optional<GameAccount> account = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
String event_time = DateUtils.formatAsYMDHMS(value.getEvent_time());
String activationTime = event_time;
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);
} 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()) { //设备未注册
gameAccountDao.insertRegister(value, event_time);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
if (DateUtils.daysBetween(activition, value.getEvent_time()) == 0) { //当日激活并且注册
activationMetrics.setData_day_act_register(1L);
}
}
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
if (accountRoleLogin.isPresent()) {
String last_login_time = accountRoleLogin.get().getEvent_time();
if (!StringUtils.isBlank(last_login_time)) {
Date date = DateUtils.formatAsDate(last_login_time);
if (DateUtils.daysBetween(date, value.getEvent_time()) > 0) { //排除当日登录用户
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween); //留存天数
ActivationMetrics dauMetric = new ActivationMetrics();
copyBaseMetrics(value, dauMetric);
dauMetric.setUid(value.getUid());
dauMetric.setData_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
dauMetric.setData_hour(Integer.valueOf(DateUtils.formatAsHH(value.getEvent_time())));
dauMetric.setData_active_day_dau(1L);
out.collect(dauMetric);
}
}
}
} else {
activationTime = gameAccount.getEvent_time();
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { //设备未注册
gameAccountDao.insertRegister(value, event_time);
activationMetrics.setData_register(1L);
activationMetrics.setData_device_register(1L);
} else {
activationTime = accountRegister.get().getEvent_time();
}
LocalDateTime activition = DateUtils.formatAsDate(activationTime);
activationMetrics.setData_date(Long.parseLong(DateUtils.formatAsYMD(activationTime)));
activationMetrics.setData_hour(Integer.parseInt(DateUtils.formatAsHH(activationTime)));
// CommonUtils.setType(activationMetrics);
Optional<GameAccount> accountRoleLogin = gameAccountDao.findByPrimary(value, Constants.TYPE_ROLELOGIN);
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) {
activationMetrics.setDateRemind(daysBetween); //留存天数
ActivationMetrics dauMetric = new ActivationMetrics();
copyBaseMetrics(value, dauMetric);
dauMetric.setUid(value.getUid());
dauMetric.setData_date(Long.valueOf(DateUtils.formatAsYMD(value.getEvent_time())));
dauMetric.setData_hour(Integer.valueOf(DateUtils.formatAsHH(value.getEvent_time())));
dauMetric.setData_active_day_dau(1L);
out.collect(dauMetric);
}
}
} else {
long daysBetween = DateUtils.daysBetween(activition, value.getEvent_time());
if (daysBetween > 0) {
activationMetrics.setDateRemind(daysBetween); //留存天数
}
}
gameAccountDao.insertRoleLogin(value);
out.collect(activationMetrics);
......
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.dao.MetricsDetailDao;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DatabaseUtils;
import com.fshark.overseas.advert.util.JdbiContext;
import org.apache.flink.configuration.Configuration;
......@@ -11,7 +13,7 @@ import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.checkerframework.checker.units.qual.C;
public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetrics, Object, TimeWindow> {
public class StoreMetricProcess extends ProcessAllWindowFunction<SinkMetricEvent, Object, TimeWindow> {
private AppConfig appConfig;
......@@ -26,45 +28,52 @@ public class StoreMetricProcess extends ProcessAllWindowFunction<ActivationMetri
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
metricsDetailDao = jdbiContext.getJdbi().onDemand(MetricsDetailDao.class);
}
@Override
public void process(Context context, Iterable<ActivationMetrics> elements, Collector<Object> out) throws Exception {
for (ActivationMetrics element : elements) {
if (element.getData_activation() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_activation");
}
if (element.getData_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_register");
}
if (element.getData_device_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_device_register");
}
if (element.getData_create_role() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_create_role");
}
if (element.getData_day_act_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_day_act_register");
}
if (element.getData_day_register_create() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_day_register_create");
}
if (element.getData_active_day_dau() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_active_day_dau");
}
if (element.getDateRemind() > 0) {
String column = "retention_day" + element.getDateRemind();
metricsDetailDao.insertMetricDetail(element, column);
public void process(Context context, Iterable<SinkMetricEvent> elements, Collector<Object> out) throws Exception {
for (SinkMetricEvent sinkMetricEvent : elements) {
if (Constants.TYPE_SINK_EVENT.equals(sinkMetricEvent.getEventType())){
ActivationMetrics element = sinkMetricEvent.getEventData();
if (element.getData_activation() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_activation");
}
if (element.getData_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_register");
}
if (element.getData_device_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_device_register");
}
if (element.getData_create_role() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_create_role");
}
if (element.getData_day_act_register() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_day_act_register");
}
if (element.getData_day_register_create() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_day_register_create");
}
if (element.getData_active_day_dau() > 0) {
metricsDetailDao.insertMetricDetail(element, "data_active_day_dau");
}
if (element.getDateRemind() > 0) {
String column = "retention_day" + element.getDateRemind();
metricsDetailDao.insertMetricDetail(element, column);
}
}
}
......
......@@ -4,6 +4,7 @@ package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.MetricsDao;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DatabaseUtils;
import com.fshark.overseas.advert.util.JdbiContext;
import org.apache.flink.configuration.Configuration;
......@@ -35,7 +36,11 @@ public class ActicitionSink extends RichSinkFunction<ActivationMetrics> {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
metricsDao = jdbiContext.getJdbi().onDemand(MetricsDao.class);
}
......
......@@ -6,6 +6,7 @@ import com.fshark.overseas.advert.entity.dao.OrderDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.AppsflyerPurchase;
import com.fshark.overseas.advert.modle.OrderData;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.util.*;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
......@@ -13,7 +14,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.util.Optional;
public class PurchaseSink extends RichSinkFunction<AppsFlyerEvent> {
public class PurchaseSink extends RichSinkFunction<SinkMetricEvent> {
private AppConfig appConfig;
......@@ -28,40 +29,25 @@ public class PurchaseSink extends RichSinkFunction<AppsFlyerEvent> {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
orderDao = jdbiContext.getJdbi().onDemand(OrderDao.class);
gameAccountDao = jdbiContext.getJdbi().onDemand(GameAccountDao.class);
}
@Override
public void invoke(AppsFlyerEvent value, Context context) throws Exception {
public void invoke(SinkMetricEvent value, Context context) throws Exception {
super.invoke(value, context);
AppsflyerPurchase appsflyerPurchase = new Gson().fromJson(value.getEvent_value(), AppsflyerPurchase.class);
if (appsflyerPurchase == null || appsflyerPurchase.getBundle_id() == null) { //不是我们上报的格式
return;
}
if ("ios".equalsIgnoreCase(value.getPlatform())) {
String bundleId = appsflyerPurchase.getBundle_id();
value.setGame_id(bundleId == null ? value.getGame_id() : bundleId);
}
OrderData orderData = new OrderData(value);
Optional<GameAccount> accountActivition = gameAccountDao.findByPrimary(value, Constants.TYPE_ACTIVIYION);
if (accountActivition.isPresent()) {
GameAccount gameAccount = accountActivition.get();
orderData.setActivation_time(gameAccount.getEvent_time());
CommonUtils.setType(Long.parseLong(DateUtils.formatAsYMD(gameAccount.getEvent_time())),orderData.getMedia_source(),orderData.getData_type());
if (Constants.TYPE_SINK_ORDER.equals(value.getEventType())){
OrderData eventData = value.getOrderData();
orderDao.insertOrder(eventData);
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (accountRegister.isPresent()) {
GameAccount gameAccount = accountRegister.get();
orderData.setReg_time(gameAccount.getEvent_time());
}
orderDao.insertOrder(orderData);
gameAccountDao.insertPauchase(value);
}
@Override
public void close() throws Exception {
super.close();
......
......@@ -3,6 +3,7 @@ package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.MetricsDao;
import com.fshark.overseas.advert.modle.metrics.RemainMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DatabaseUtils;
import com.fshark.overseas.advert.util.JdbiContext;
import org.apache.flink.configuration.Configuration;
......@@ -59,7 +60,11 @@ public class RemainSink extends RichSinkFunction<RemainMetrics> {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
if (appConfig.getSinkType().equals(Constants.TYPE_SINK)){
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
} else {
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
metricsDao = jdbiContext.getJdbi().onDemand(MetricsDao.class);
}
......
......@@ -38,4 +38,12 @@ public class AppConfig implements Serializable {
private String sourceKafkaTopic;
private String sinkKafkaBootstrapServers;
private String sinkKafkaGroupId;
private String sinkKafkaTopic;
private String sinkType;
}
......@@ -74,4 +74,12 @@ public class Constants {
public static final String TYPE_ROLELOGIN = "rolelogin";
public static final String TYPE_PAUCHASE = "purchase";
public static final String TYPE_SINK="mysql";
public static final String TYPE_SINK_EVENT="metric";
public static final String TYPE_SINK_ORDER="order";
}
......@@ -17,6 +17,19 @@ import java.sql.SQLException;
*/
public class DatabaseUtils {
public static JdbiContext getGenJdbiClient(String url, String username, String password) {
HikariConfig hc = new HikariConfig();
hc.setDriverClassName("com.mysql.cj.jdbc.Driver");
hc.setJdbcUrl(url);
hc.setUsername(username);
hc.setPassword(password);
hc.setMaximumPoolSize(4);
HikariDataSource hikariDataSource = new HikariDataSource(hc);
Jdbi jdbi = Jdbi.create(hikariDataSource);
jdbi.installPlugin(new SqlObjectPlugin());
return new JdbiContext(hikariDataSource, jdbi);
}
public static JdbiContext getCHGenJdbiClient(String url, String username, String password) throws SQLException {
HikariConfig hc = new HikariConfig();
......
package com.fshark.overseas.advert.util;
import org.gradle.internal.impldep.org.joda.time.DateTime;
import org.gradle.internal.impldep.org.joda.time.Days;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
......@@ -23,21 +28,9 @@ public class DateUtils {
static DateTimeFormatter YMD_FORMATTER = DateTimeFormatter.ofPattern(Constants.YMD);
public static String formatAsYMD(Date ts) {
if (ts == null){
return "-1";
}
LocalDateTime localDateTime = ts.toInstant()
.atZone(ZoneId.systemDefault())
.toLocalDateTime();
return localDateTime.format(YMD_FORMATTER);
}
public static String formatAsYMDHMS(Date ts) {
LocalDateTime localDateTime = ts.toInstant()
.atZone(ZoneId.systemDefault())
.toLocalDateTime();
return localDateTime.format(YMDHMS_FORMATTER);
public static String formatAsYMDHMS(LocalDateTime ts) {
return YMDHMS_FORMATTER.format(ts);
}
......@@ -53,72 +46,30 @@ public class DateUtils {
}
public static String formatAsHH(Date ts) {
LocalDateTime localDateTime = ts.toInstant()
.atZone(ZoneId.systemDefault())
.toLocalDateTime();
return localDateTime.format(HH_FORMATTER);
}
public static String formatAsHH(String ts) {
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_FORMATTER);
return parse.format(HH_FORMATTER);
}
public static Date formatAsDate(String ts) {
if (ts.length() > 19) {
return formatHMSSAsDate(ts);
}
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_FORMATTER);
return Date.from(parse.atZone(ZoneId.systemDefault()).toInstant());
}
public static Date formatHMSSAsDate(String ts) {
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_MILLIS_FORMATTER);
return Date.from(parse.atZone(ZoneId.systemDefault()).toInstant());
public static LocalDateTime formatAsDate(String ts) {
return LocalDateTime.parse(ts, YMDHMS_FORMATTER);
}
public static String formatHMSSAsYMD(String ts) {
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_MILLIS_FORMATTER);
return parse.format(YMD_FORMATTER);
}
public static String formatHMSSAsHH(String ts) {
LocalDateTime parse = LocalDateTime.parse(ts, YMDHMS_MILLIS_FORMATTER);
return parse.format(HH_FORMATTER);
public static LocalDateTime formatHMSSAsDate(String ts) {
return LocalDateTime.parse(ts, YMDHMS_MILLIS_FORMATTER);
}
/**
* @return 相差天数
*/
public static long daysBetween(Date smdate, Date bdate) throws ParseException {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
smdate = sdf.parse(sdf.format(smdate));
bdate = sdf.parse(sdf.format(bdate));
Calendar cal = Calendar.getInstance();
cal.setTime(smdate);
long time1 = cal.getTimeInMillis();
cal.setTime(bdate);
long time2 = cal.getTimeInMillis();
long between_days = (time2 - time1) / (1000 * 3600 * 24);
return Integer.parseInt(String.valueOf(between_days));
}
public static long daysBetween(LocalDateTime smdate, Date bdate) throws ParseException {
Duration between = Duration.between(smdate, dateToLocalDateTime(bdate));
public static long daysBetween(LocalDateTime smdate, LocalDateTime bdate) {
Duration between = Duration.between(smdate.toLocalDate().atStartOfDay(), bdate.toLocalDate().atStartOfDay());
return between.toDays();
}
public static Date localDateTimeToDate(LocalDateTime localDateTime) {
if (localDateTime == null) return null;
return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
}
public static LocalDateTime dateToLocalDateTime(Date date) {
if (date == null) return null;
return date.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
}
}
......
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