Commit 3d588d75 authored by she's avatar she

ios 模糊归因

parent 761ab623
......@@ -78,6 +78,8 @@ dependencies {
testCompile group: 'it.ozimov', name: 'embedded-redis', version: '0.7.2'
testCompile "org.mockito:mockito-core:2.+"
testCompile "org.apache.flink:flink-test-utils_${scalaBinaryVersion}:${flinkVersion}"
compile "cn.hutool:hutool-all:5.8.10"
}
task relocateShadowJar(type: ConfigureShadowRelocation) {
......
package com.fshark.overseas.advert.app;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.processor.GameAdMappingProcessor;
import com.fshark.overseas.advert.processor.MutiplatDataProcess;
import com.fshark.overseas.advert.processor.ParseEventProcessor;
import com.fshark.overseas.advert.sink.TiktokSink;
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.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/**
* 广告计算
*/
public class AppsflyerAdmapping {
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.setStartOffsetTime(Long.valueOf(parameters.get("start-offset-time")));
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"));
} 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);
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");
//ios模糊匹配
adEventDataStream
.addSink(new TiktokSink())
.uid("tiktok")
.name("tiktok");
//测流提取出来激活,登录,创角角色登录,订单事件
adEventDataStream.process(new MutiplatDataProcess()).filter(EventTypeEnum::loginEvent).process(new GameAdMappingProcessor(appConfig)).name("admapping").uid("admapping");
//
sse.execute("appsflyer-admapping");
}
}
......@@ -5,7 +5,6 @@ import com.fshark.overseas.advert.modle.AppsFlyerEvent;
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.EventMetrics;
import com.fshark.overseas.advert.processor.*;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.Constants;
......@@ -14,15 +13,11 @@ 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.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
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.SingleOutputStreamOperator;
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.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
......@@ -99,6 +94,7 @@ public class AppsflyerAnalyze {
.uid("parse-event")
.name("parse-event");
//测流提取出来激活,登录,创角角色登录,订单事件
OutputTag<AppsFlyerEvent> installTag = new OutputTag<AppsFlyerEvent>("install") {
};
......
......@@ -2,7 +2,6 @@ package com.fshark.overseas.advert.modle;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.util.Constants;
import com.fshark.overseas.advert.util.DateUtils;
import lombok.Data;
......@@ -12,9 +11,7 @@ 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;
......@@ -178,6 +175,20 @@ public class AppsFlyerEvent implements Serializable {
*/
private Boolean is_retargeting = false;
/**
* 原始数据
*/
private String original_url;
private String idfa;
private String idfv;
private String user_agent;
private String attributed_touch_type;
private String match_type;
private static List<String> validation_fields =
......@@ -246,8 +257,11 @@ public class AppsFlyerEvent implements Serializable {
private static void setDidByType(AppsFlyerEvent appsFlyerEvent, JSONObject jsonObject) {
String advertisingId = jsonObject.getString("advertising_id");
String idfa = jsonObject.getString("idfa");
appsFlyerEvent.setIdfa(idfa);
String idfv = jsonObject.getString("idfv");
appsFlyerEvent.setIdfv(idfv);
if (StringUtils.isBlank(idfa)) {
idfa = jsonObject.getString("idfv");
idfa = idfv;
}
String platform = jsonObject.getString("platform");
......@@ -256,7 +270,7 @@ public class AppsFlyerEvent implements Serializable {
did = nvl(idfa, Constants.DEFAULT_FIELD_VALUE);
} else if (PlatformEnum.Android.getName().equals(platform)) {
did = nvl(advertisingId, Constants.DEFAULT_FIELD_VALUE);
} else if(PlatformEnum.PC.getName().equals(platform)) {
} else if (PlatformEnum.PC.getName().equals(platform)) {
did = nvl(advertisingId, Constants.DEFAULT_FIELD_VALUE);
} else {
did = Constants.DEFAULT_FIELD_VALUE;
......@@ -324,7 +338,7 @@ public class AppsFlyerEvent implements Serializable {
}
//有的渠道Campaign有值但是campaign_id没有值,做这个兼容设置
String modifiedCampaign = "None".equals(campaign) ? Constants.DEFAULT_FIELD_VALUE : campaign;
if (!Constants.DEFAULT_FIELD_VALUE.equals(modifiedCampaign) && Constants.DEFAULT_FIELD_VALUE.equals(campaign_id)){
if (!Constants.DEFAULT_FIELD_VALUE.equals(modifiedCampaign) && Constants.DEFAULT_FIELD_VALUE.equals(campaign_id)) {
campaign_id = modifiedCampaign;
}
appsFlyerEvent.setCampaign_id(campaign_id);
......@@ -381,6 +395,10 @@ public class AppsFlyerEvent implements Serializable {
Boolean is_retargeting = Boolean.parseBoolean(jsonObject.getString("is_retargeting"));
// String af_channel = String.valueOf(jsonObject.getOrDefault("af_channel", Constants.DEFAULT_FIELD_VALUE));
String original_url = jsonObject.getString("original_url");
String user_agent = jsonObject.getString("user_agent");
String attributed_touch_type = jsonObject.getString("attributed_touch_type");
String match_type = jsonObject.getString("match_type");
// TODO 这个地方如果对于同一账户,不同设备,会有问题,因为这个是appflyer 针对每个安装自动生成的唯一的ID
String uid = String.valueOf(jsonObject.getOrDefault("appsflyer_device_id", Constants.DEFAULT_FIELD_VALUE));
......@@ -413,12 +431,15 @@ public class AppsFlyerEvent implements Serializable {
}
AppsFlyerEvent appsFlyerEvent = new AppsFlyerEvent();
appsFlyerEvent.setUser_agent(user_agent);
appsFlyerEvent.setOriginal_url(original_url);
appsFlyerEvent.setAttributed_touch_type(attributed_touch_type);
appsFlyerEvent.setMatch_type(match_type);
String event = getEvent(jsonObject, eventType, eventName);
appsFlyerEvent.setEvent_type(eventType);
// String[] timeZoneParts = ;
LocalDateTime eventTime = DateUtils.formatHMSSAsDate(eventTimeWithZone.substring(0,23));
LocalDateTime eventTime = DateUtils.formatHMSSAsDate(eventTimeWithZone.substring(0, 23));
// 如果有错,直接视为错误; +0900
String timeZone = eventTimeWithZone.substring(23);
......@@ -450,7 +471,7 @@ public class AppsFlyerEvent implements Serializable {
if (StringUtils.isBlank(af_channel)) {
af_channel = Constants.DEFAULT_FIELD_VALUE;
}
if(StringUtils.equals(af_channel,"Pangle")){
if (StringUtils.equals(af_channel, "Pangle")) {
appsFlyerEvent.setMedia_source("Pangle");
}
appsFlyerEvent.setChannel(af_channel);
......
......@@ -8,7 +8,11 @@ public enum PlatformEnum {
IOS("ios"),
Android("android"),
PC("pc");
PC("pc"),
MUTI("muti")
;
PlatformEnum(String name) {
this.name = name;
......
package com.fshark.overseas.advert.modle;
import com.google.gson.annotations.SerializedName;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.util.List;
@NoArgsConstructor
@Data
public class TiktokModle implements Serializable {
@SerializedName("tiktok_app_id")
private Long tiktokAppId;
@SerializedName("event")
private String event;
@SerializedName("timestamp")
private String timestamp;
@SerializedName("context")
private ContextDTO context;
@SerializedName("properties")
private PropertiesDTO properties;
@NoArgsConstructor
@Data
public static class ContextDTO {
@SerializedName("app")
private AppDTO app;
@SerializedName("device")
private DeviceDTO device;
@SerializedName("locale")
private String locale;
@SerializedName("ip")
private String ip;
@SerializedName("userAgent")
private String userAgent;
@SerializedName("origin_url")
private String originUrl;
@SerializedName("ad")
private AdDTO ad;
@NoArgsConstructor
@Data
public static class AppDTO {
@SerializedName("id")
private String id;
@SerializedName("name")
private String name;
@SerializedName("namespace")
private String namespace;
@SerializedName("version")
private String version;
@SerializedName("build")
private String build;
}
@NoArgsConstructor
@Data
public static class DeviceDTO {
@SerializedName("platform")
private String platform;
@SerializedName("idfa")
private String idfa;
@SerializedName("idfv")
private String idfv;
}
@NoArgsConstructor
@Data
public static class AdDTO {
@SerializedName("placement")
private String placement;
@SerializedName("match_type")
private String matchType;
@SerializedName("attributed")
private Boolean attributed;
@SerializedName("ad_id")
private Long adId;
@SerializedName("campaign_id")
private Long campaignId;
@SerializedName("creative_id")
private Long creativeId;
@SerializedName("isRetargeting")
private boolean isRetargeting ;
@SerializedName("attributionType")
private String attributionType;
@SerializedName("attribution_provider")
private String attributionProvider;
}
}
@NoArgsConstructor
@Data
public static class PropertiesDTO {
@SerializedName("currency")
private String currency;
@SerializedName("value")
private Integer value;
@SerializedName("contents")
private List<ContentsDTO> contents;
@NoArgsConstructor
@Data
public static class ContentsDTO {
@SerializedName("price")
private Integer price;
@SerializedName("quantity")
private Integer quantity;
@SerializedName("content_id")
private String contentId;
}
}
}
package com.fshark.overseas.advert.processor;
import com.fshark.overseas.advert.entity.dao.GameMappingDao;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.PlatformEnum;
......@@ -12,8 +11,8 @@ import com.fshark.overseas.advert.modle.metrics.ActivationMetrics;
import com.fshark.overseas.advert.util.AppConfig;
import com.fshark.overseas.advert.util.DatabaseUtils;
import com.fshark.overseas.advert.util.JdbiContext;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.util.Optional;
......@@ -21,7 +20,7 @@ import java.util.Optional;
/**
* 更新游戏广告的映射
*/
public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent, ActivationMetrics> {
public class GameAdMappingProcessor extends ProcessFunction<AppsFlyerEvent, ActivationMetrics> {
private AppConfig appConfig;
......@@ -38,23 +37,36 @@ public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent,
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
jdbiContext = DatabaseUtils.getCHGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
jdbiContext = DatabaseUtils.getGenJdbiClient(appConfig.getDbHost(), appConfig.getDbUser(), appConfig.getDbPassword());
gameMappingDao = jdbiContext.getJdbi().onDemand(GameMappingDao.class);
}
@Override
public void flatMap(AppsFlyerEvent value, Collector<ActivationMetrics> out) throws Exception {
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
@Override
public void processElement(AppsFlyerEvent value, ProcessFunction<AppsFlyerEvent, ActivationMetrics>.Context ctx, Collector<ActivationMetrics> out) throws Exception {
int platform_id = 1;
if (PlatformEnum.IOS.getName().equals(value.getPlatform())) {
platform_id = 2;
} else if (PlatformEnum.PC.getName().equals(value.getPlatform())) {
platform_id = 3;
} else if (PlatformEnum.MUTI.getName().equals(value.getPlatform())) {
platform_id = 4;
}
//更新game
// boolean gameByPrimary = gameMappingDao.findGameByPrimary(value, platform_id);
// if (!gameByPrimary) {
// gameMappingDao.saveGame(value, platform_id);
// }
boolean gameByPrimary = gameMappingDao.findGameByPrimary(value.getGame_id(), platform_id).isPresent();
if (!gameByPrimary) {
gameMappingDao.saveGame(value, platform_id);
}
//mediaSource更新
Optional<Media> mediaByPrimary = gameMappingDao.findMediaByPrimary(value);
......@@ -76,17 +88,5 @@ public class GameAdMappingProcessor extends RichFlatMapFunction<AppsFlyerEvent,
gameMappingDao.saveAdMapping(value, media);
}
}
}
@Override
public void close() throws Exception {
super.close();
if (jdbiContext != null) {
jdbiContext.getDataSource().close();
}
}
}
......@@ -24,6 +24,6 @@ public class MutiplatDataProcess extends ProcessFunction<AppsFlyerEvent, AppsFly
private void setMutiValue(AppsFlyerEvent value) {
value.setGame_id("com.chaosportal.shark.all");
value.setPlatform("android");
value.setPlatform("muti");
}
}
package com.fshark.overseas.advert.processor;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
......@@ -11,6 +11,7 @@ import org.slf4j.LoggerFactory;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
public class ParseEventProcessor implements FlatMapFunction<String, AppsFlyerEvent> {
......@@ -19,8 +20,10 @@ public class ParseEventProcessor implements FlatMapFunction<String, AppsFlyerEve
@Override
public void flatMap(String value, Collector<AppsFlyerEvent> out) throws Exception {
try {
JSONObject jsonObject = JSON.parseObject(value);
String message = jsonObject.getString("message");
// JSONObject jsonObject = JSON.parseObject(value);
// String message = jsonObject.getString("message");
HashMap<String, Object> hashMap = new Gson().fromJson(value, HashMap.class);
String message = (String) hashMap.get("message");
String parseLine = "";
try {
......
package com.fshark.overseas.advert.sink;
import cn.hutool.http.Header;
import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fshark.overseas.advert.modle.AppsFlyerEvent;
import com.fshark.overseas.advert.modle.EventTypeEnum;
import com.fshark.overseas.advert.modle.TiktokModle;
import com.google.gson.Gson;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
public class TiktokSink extends RichSinkFunction<AppsFlyerEvent> {
private Gson gson;
private static String ACCESS_TOKEN = "0e76c1138387d6ee7805eca2334bba89ad855557";
private static Long TIKTOK_APPID = 7174239434366320641L;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
gson = new Gson();
}
public void sendReport(TiktokModle tiktokModle) throws JsonProcessingException {
String body = gson.toJson(tiktokModle);
String result = HttpRequest.post("https://business-api.tiktok.com/open_api/v1.2/app/track/")
.header(Header.CONTENT_TYPE, "application/json")
.header("Access-Token", ACCESS_TOKEN)
.body(body)
.timeout(20000) //超时,毫秒
.execute().body();
// System.out.println(body);
// System.out.println(result);
}
@Override
public void invoke(AppsFlyerEvent value, Context context) throws Exception {
if (!value.getGame_id().equals("com.chaosportal.shark") || !"pangle".equals(value.getMedia_source())) {
return;
}
if (EventTypeEnum.installEvent(value) || EventTypeEnum.orderEvent(value)) {
String event_name = "";
if (EventTypeEnum.installEvent(value)) {
event_name = "InstallApp";
}
TiktokModle tiktokModle = new TiktokModle();
tiktokModle.setTiktokAppId(TIKTOK_APPID);
tiktokModle.setEvent(event_name);
tiktokModle.setTimestamp(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'hh:mm:ss'Z'").format(value.getEvent_time().atOffset(ZoneOffset.UTC)));
TiktokModle.ContextDTO contextDTO = new TiktokModle.ContextDTO();
contextDTO.setIp(value.getIp());
contextDTO.setUserAgent(value.getUser_agent());
contextDTO.setOriginUrl(value.getOriginal_url());
TiktokModle.ContextDTO.AppDTO appDTO = new TiktokModle.ContextDTO.AppDTO();
appDTO.setId(value.getApp_id());
contextDTO.setApp(appDTO);
TiktokModle.ContextDTO.DeviceDTO deviceDTO = new TiktokModle.ContextDTO.DeviceDTO();
deviceDTO.setPlatform(value.getPlatform());
deviceDTO.setIdfa(value.getIdfa());
deviceDTO.setIdfv(value.getIdfv());
contextDTO.setDevice(deviceDTO);
TiktokModle.ContextDTO.AdDTO adDTO = new TiktokModle.ContextDTO.AdDTO();
adDTO.setPlacement("Pangle");
if (!"id_matching".equals(value.getMatch_type())) {
adDTO.setMatchType("probabilistic");
} else {
adDTO.setMatchType(value.getMatch_type());
}
adDTO.setCampaignId(Long.valueOf(value.getCampaign_id()));
adDTO.setAdId(Long.valueOf(value.getAd_set_id()));
adDTO.setCreativeId(Long.valueOf(value.getAd_id()));
adDTO.setRetargeting(value.getIs_retargeting());
adDTO.setAttributed(true);
adDTO.setAttributionType(value.getAttributed_touch_type() + "_through");
adDTO.setAttributionProvider("mmp");
contextDTO.setAd(adDTO);
tiktokModle.setContext(contextDTO);
if (EventTypeEnum.orderEvent(value)) {
JSONObject json = JSON.parseObject(value.getEvent_value());
String currency = json.getString("af_currency");
Integer revenue = Integer.valueOf(json.getString("af_revenue"));
TiktokModle.PropertiesDTO propertiesDTO = new TiktokModle.PropertiesDTO();
propertiesDTO.setCurrency(currency);
propertiesDTO.setValue(revenue);
ArrayList<TiktokModle.PropertiesDTO.ContentsDTO> contentsDTOS = new ArrayList<>();
TiktokModle.PropertiesDTO.ContentsDTO contentsDTOS1 = new TiktokModle.PropertiesDTO.ContentsDTO();
contentsDTOS1.setPrice(revenue);
contentsDTOS1.setQuantity(1);
contentsDTOS.add(contentsDTOS1);
propertiesDTO.setContents(contentsDTOS);
tiktokModle.setProperties(propertiesDTO);
}
sendReport(tiktokModle);
}
}
}
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