Commit 949152e9 authored by sheteng's avatar sheteng

rank数据处理

parent efdb9968
/build/ /build/
/.gradle /.gradle
/.gradle/
appsflyer_advert_new_flink
\ No newline at end of file
<component name="ArtifactManager">
<artifact type="jar" build-on-make="true" name="appsflyer_advert_new_flink:jar">
<output-path>$PROJECT_DIR$/out/artifacts/appsflyer_advert_new_flink_jar</output-path>
<root id="archive" name="appsflyer_advert_new_flink.jar">
<element id="directory" name="META-INF">
<element id="file-copy" path="$PROJECT_DIR$/META-INF/MANIFEST.MF" />
</element>
</root>
</artifact>
</component>
\ No newline at end of file
...@@ -16,5 +16,10 @@ ...@@ -16,5 +16,10 @@
<option name="name" value="MavenRepo" /> <option name="name" value="MavenRepo" />
<option name="url" value="https://repo.maven.apache.org/maven2/" /> <option name="url" value="https://repo.maven.apache.org/maven2/" />
</remote-repository> </remote-repository>
<remote-repository>
<option name="id" value="maven" />
<option name="name" value="maven" />
<option name="url" value="https://nexus.sky-shark.com/repository/ccnet-group/" />
</remote-repository>
</component> </component>
</project> </project>
\ No newline at end of file
...@@ -37,6 +37,9 @@ tasks.withType(JavaCompile) { ...@@ -37,6 +37,9 @@ tasks.withType(JavaCompile) {
repositories { repositories {
mavenCentral() mavenCentral()
maven {
url "https://nexus.sky-shark.com/repository/ccnet-group/"
}
} }
dependencies { dependencies {
...@@ -57,10 +60,10 @@ dependencies { ...@@ -57,10 +60,10 @@ dependencies {
// https://mvnrepository.com/artifact/mysql/mysql-connector-java // 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'
compile "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}" compileOnly "org.apache.flink:flink-statebackend-rocksdb_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-java:${flinkVersion}" compileOnly "org.apache.flink:flink-java:${flinkVersion}"
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}" compileOnly "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
compile "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}" compileOnly "org.apache.flink:flink-clients_${scalaBinaryVersion}:${flinkVersion}"
implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52' implementation 'ru.yandex.clickhouse:clickhouse-jdbc:0.1.52'
// -------------------------------------------------------------- // --------------------------------------------------------------
// Dependencies that should be part of the shadow jar, e.g. // Dependencies that should be part of the shadow jar, e.g.
......
...@@ -7,6 +7,7 @@ import com.fshark.overseas.advert.modle.metrics.ActivationMetrics; ...@@ -7,6 +7,7 @@ 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 org.apache.flink.api.common.functions.FilterFunction;
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.java.utils.ParameterTool; import org.apache.flink.api.java.utils.ParameterTool;
...@@ -69,10 +70,6 @@ public class AppsflyerAnalyze { ...@@ -69,10 +70,6 @@ public class AppsflyerAnalyze {
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);
// TODO: 2021/8/10 设置消费开始的时间戳1632326555000
// if (appConfig.getStartOffsetTime() > 0) {
// dataConsumer.setStartFromTimestamp(appConfig.getStartOffsetTime());
// }
Properties props2 = new Properties(); Properties props2 = new Properties();
props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers()); props2.setProperty("bootstrap.servers", appConfig.getSinkKafkaBootstrapServers());
...@@ -80,6 +77,15 @@ public class AppsflyerAnalyze { ...@@ -80,6 +77,15 @@ public class AppsflyerAnalyze {
DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer) DataStream<AppsFlyerEvent> adEventDataStream = sse.addSource(dataConsumer)
.flatMap(new ParseEventProcessor()) .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") .uid("parse-event")
.name("parse-event") .name("parse-event")
.shuffle(); .shuffle();
......
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