Commit 921b6f74 authored by she's avatar she

sdk注册数 取自 ouid注册数

parent bbbdc8ac
......@@ -55,10 +55,10 @@ dependencies {
// https://mvnrepository.com/artifact/mysql/mysql-connector-java
implementation group: 'mysql', name: 'mysql-connector-java', version: '8.0.25'
compile "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-java:${flinkVersion}"
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-java:${flinkVersion}"
compileOnly "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compileOnly "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52'
// --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g.
......
......@@ -72,5 +72,5 @@ public interface GameAccountDao {
@SqlQuery("SELECT * FROM appsflyer_activation_devices WHERE game_id=:game_id AND ouid=:ouid And platform =:platform AND event_type =:event_type order by event_time desc limit 1")
@RegisterRowMapper(GameAccountMapper.class)
Optional<GameAccount> findRegisterByouid(AppsFlyerEvent value, String ouid, String typeRegister);
Optional<GameAccount> findRegisterByouid(@BindBean AppsFlyerEvent value, @Bind("ouid")String ouid, @Bind("event_type")String typeRegister);
}
......@@ -19,6 +19,7 @@ public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationM
private AppConfig appConfig;
private Connection conn;
public OuidStoreMetricProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
......@@ -39,7 +40,11 @@ public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationM
public void process(Context context, Iterable<ActivationMetrics> elements, Collector<Object> out) throws Exception {
//详细指标写入
insertMetricsDetail(elements);
for (ActivationMetrics element : elements) {
if (element.getData_ouid_register() > 0) {
insertActivtionMetrics(element);
}
}
}
private void insertMetricsDetail(Iterable<ActivationMetrics> elements) {
......@@ -58,6 +63,42 @@ public class OuidStoreMetricProcess extends ProcessAllWindowFunction<ActivationM
}
}
private void insertActivtionMetrics(ActivationMetrics value) {
try {
PreparedStatement insertStmt = conn.prepareStatement("insert into appsflyer_metrics" +
" (`data_date`,`data_hour`,`time_zone`,`region`, `platform`, `game_id`, `media_source`, `campaign_id` ,`ad_set_id`,`ad_id`,`channel`,`data_activation`,`data_sdk_activation`,`data_sdk_register`,`data_register`,`data_device_register`,`data_create_role`,`data_day_act_register`,`data_day_register_create`,`data_active_day_dau`,`data_type`)" +
" values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ");
insertStmt.setString(1, value.getData_date().toString());
insertStmt.setString(2, value.getData_hour().toString());
insertStmt.setString(3, value.getTime_zone());
insertStmt.setString(4, value.getRegion());
insertStmt.setString(5, value.getPlatform());
insertStmt.setString(6, value.getGame_id());
insertStmt.setString(7, value.getMedia_source());
insertStmt.setString(8, value.getCampaign_id());
insertStmt.setString(9, value.getAd_set_id());
insertStmt.setString(10, value.getAd_id());
insertStmt.setString(11, value.getChannel());
insertStmt.setLong(12, 0);
insertStmt.setLong(13, 0);
insertStmt.setLong(14, 1);
insertStmt.setLong(15, 0);
insertStmt.setLong(16, 0);
insertStmt.setLong(17, 0);
insertStmt.setLong(18, 0);
insertStmt.setLong(19, 0);
insertStmt.setLong(20, 0);
insertStmt.setLong(21, value.getData_type());
insertStmt.addBatch();
insertStmt.executeBatch(); //批量执行
conn.commit();
insertStmt.close();
} catch (Exception e) {
e.printStackTrace();
}
}
private void createBatch(String name, PreparedStatement insertStmt, ActivationMetrics element) throws SQLException {
if (!TextUtils.isEmpty(name)) {
......
......@@ -89,11 +89,6 @@ public class RoleLoginProcess extends RichFlatMapFunction<AppsFlyerEvent, Activa
} else {
Optional<GameAccount> sdkaAccount = gameAccountDao.findRegisterByouid(value,ouid, Constants.TYPE_REGISTER);
if (!sdkaAccount.isPresent()) {
activationMetrics.setData_sdk_register(1L);
}
Optional<GameAccount> accountRegister = gameAccountDao.findByPrimary(value, Constants.TYPE_REGISTER);
if (!accountRegister.isPresent()) { //设备未注册
gameAccountDao.insertRegister(value, event_time,ouid);
......
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