Commit 60d6faa4 authored by sheteng's avatar sheteng

详细数据写入

parent b4277acc
Manifest-Version: 1.0
Main-Class: com.fshark.overseas.advert.AppsflyerAnalyze
Main-Class: com.fshark.overseas.advert.app.AppsflyerAnalyze
package com.fshark.overseas.advert;
package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
......@@ -148,6 +148,7 @@ public class AppsflyerAnalyze {
.uid("aggregate-event")
.name("aggregate-event");
//
sse.execute("appsflyer-analyze");
}
......
package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.processor.ParseEventProcessor;
import com.fshark.overseas.advert.sink.EventInfoSink;
import com.fshark.overseas.advert.sink.EventInfoSinkProcess;
import com.fshark.overseas.advert.util.AppConfig;
import org.apache.flink.api.common.functions.FilterFunction;
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.environment.StreamExecutionEnvironment;
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 java.util.Properties;
import java.util.concurrent.TimeUnit;
public class AppsflyerInfoDmi {
static AppConfig loadConfigFromArgs(String[] args) {
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"));
} 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>" +
" --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> ");
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", "earliest");
FlinkKafkaConsumer<String> dataConsumer = new FlinkKafkaConsumer<>(appConfig.getSourceKafkaTopic(), new SimpleStringSchema(), properties);
DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer)
.flatMap(new ParseEventProcessor())
.filter(new FilterFunction<AppsFlyerEvent>() {
@Override
public boolean filter(AppsFlyerEvent value) throws Exception {
if (value.getMedia_source().contains("rank") || value.getCampaign_id().contains("rank")) {
return false;
}
return true;
}
})
.uid("parse-event")
.name("parse-event")
.shuffle();
adEventDataStream.timeWindowAll(Time.seconds(30)).process(new EventInfoSinkProcess(appConfig))
.uid("event")
.name("event");
sse.execute("appsflyer-analyze");
}
}
package com.fshark.overseas.advert;
package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
......
......@@ -18,7 +18,7 @@ public enum EventTypeEnum {
// REGISTER("createrole"),
CREATE_ROLE("createrole"),
SDK_ACTIVATION("activation"),
SDK_REGISTER("register"),
SDK_REGISTER("login"),
FAST_LOGIN("fast login"),
GAME_LOGIN("gamelogin"),
ROLE_LOGIN("rolelogin"),
......
package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.entity.dao.EventInfoDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.SinkMetricEvent;
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;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class EventInfoSink extends RichSinkFunction<AppsFlyerEvent> {
private AppConfig appConfig;
public EventInfoSink(AppConfig appConfig) {
this.appConfig = appConfig;
}
private JdbiContext jdbiContext;
private EventInfoDao eventInfoDao;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
eventInfoDao = jdbiContext.getJdbi().onDemand(EventInfoDao.class);
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
@Override
public void invoke(AppsFlyerEvent value, Context context) throws Exception {
super.invoke(value, context);
eventInfoDao.insertEventInfo(value);
}
}
package com.fshark.overseas.advert.sink;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.DateUtils;
import lombok.val;
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 java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
public class EventInfoSinkProcess extends ProcessAllWindowFunction<AppsFlyerEvent, Object, TimeWindow> {
private AppConfig appConfig;
public EventInfoSinkProcess(AppConfig appConfig) {
this.appConfig = appConfig;
}
// 定义连接、预编译语句
private Connection conn;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
conn = DriverManager.getConnection(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
}
@Override
public void close() throws Exception {
super.close();
if (conn != null) {
conn.close();
}
}
@Override
public void process(Context context, Iterable<AppsFlyerEvent> elements, Collector<Object> out) throws Exception {
try {
PreparedStatement insertStmt = conn.prepareStatement(" insert into appsflyer_event_info (`region`, `uid`, `did`, `platform`, `game_id`, `media_source`, `campaign_id`,`campaign_name`, `ad_set_id`,`ad_set_name`, `ad_id`,`ad_name`, `event_name`, `event_time`, `event_value`,`channel`,`time_zone`) values" +
"(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
for (AppsFlyerEvent element : elements) {
insertStmt.setString(1, element.getRegion());
insertStmt.setString(2, element.getUid());
insertStmt.setString(3, element.getDid());
insertStmt.setString(4, element.getPlatform());
insertStmt.setString(5, element.getGame_id());
insertStmt.setString(6, element.getMedia_source());
insertStmt.setString(7, element.getCampaign_id());
insertStmt.setString(8, element.getCampaign());
insertStmt.setString(9, element.getAd_set_id());
insertStmt.setString(10, element.getAd_set());
insertStmt.setString(11, element.getAd_id());
insertStmt.setString(12, element.getAd());
insertStmt.setString(13, element.getEvent());
insertStmt.setString(14, DateUtils.formatAsYMDHMS(element.getEvent_time()));
insertStmt.setString(15, element.getEvent_value());
insertStmt.setString(16, element.getChannel());
insertStmt.setString(17, element.getTime_zone());
insertStmt.addBatch();
}
insertStmt.executeBatch(); //批量执行
conn.commit();
insertStmt.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
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