Skip to content

C
ctr-estimate
Commit 0ae1b1f7 authored by 赵建伟's avatar 赵建伟
Browse files
Options

update codes

parent 28e9eadc
Hide whitespace changes
Inline Side-by-side
Showing with 190 additions and 36 deletions
pom.xml
View file @ 0ae1b1f7
......@@ -207,7 +207,7 @@
</filters>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.gmei.data.ctr.CtrEstimateMainTagTest</mainClass>
<mainClass>com.gmei.data.ctr.CtrEstimateMainTagDev</mainClass>
</transformer>
</transformers>
<createDependencyReducedPom>false</createDependencyReducedPom>
......
src/main/java/com/gmei/data/ctr/CtrEstimateMain.java
View file @ 0ae1b1f7
......@@ -6,7 +6,6 @@ import com.gmei.data.ctr.source.MaidianKafkaSource;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
......
src/main/java/com/gmei/data/ctr/CtrEstimateMainTagTest.java → src/main/java/com/gmei/data/ctr/CtrEstimateMainClkDev.java
View file @ 0ae1b1f7
package com.gmei.data.ctr;
import com.gmei.data.ctr.operator.CtrEstimateTagOperator;
import com.gmei.data.ctr.operator.CtrEstimateClkOperator;
import com.gmei.data.ctr.source.MaidianKafkaSource;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
......@@ -13,25 +16,25 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
* @Date 2020/3/30
* @Version V1.0
**/
public class CtrEstimateMainTagTest {
public class CtrEstimateMainClkDev {
public static void main(String[] args) throws Exception{
// 获取运行参数
ParameterTool parameterTool = ParameterTool.fromArgs(args);
String inBrokers = parameterTool.get("inBrokers","172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092");
String inBrokers = parameterTool.get("inBrokers","test003:9092");
String batchSize = parameterTool.get("batchSize","1000");
String maidianInTopic = parameterTool.get("maidianInTopic", "gm-maidian-data");
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-flink-tag");
Integer windowSize = parameterTool.getInt("windowSize",5);
Integer slideSize = parameterTool.getInt("slideSize",5);
String maidianInTopic = parameterTool.get("maidianInTopic", "test11");
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk");
Integer windowSize = parameterTool.getInt("windowSize",60);
Integer slideSize = parameterTool.getInt("slideSize",60);
String jdbcUrl = parameterTool.get("jdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000);
String checkpointPath = parameterTool.get("checkpointPath","hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint");
Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",true);
String checkpointPath = parameterTool.get("checkpointPath","hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate/checkpoint");
Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",false);
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime","2020-04-04 20:42:00");
String startTime = parameterTool.get("startTime");
Integer parallelism = parameterTool.getInt("parallelism",2);
System.out.println("**********************************************************");
......@@ -48,11 +51,6 @@ public class CtrEstimateMainTagTest {
// 获得流处理环境对象
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// env.enableCheckpointing(1000);
// env.setStateBackend(new FsStateBackend(checkpointPath));
// env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 3000));
// CheckpointConfig config = env.getCheckpointConfig();
// config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
DataStream MaidianDataStream = new MaidianKafkaSource(
env,
......@@ -66,9 +64,9 @@ public class CtrEstimateMainTagTest {
).getInstance();
// 执行处理核心逻辑
new CtrEstimateTagOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
new CtrEstimateClkOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
// 常驻执行
env.execute("ctr-estimate-tag");
env.execute("ctr-estimate-clk");
}
}
src/main/java/com/gmei/data/ctr/CtrEstimateMainTag.java
View file @ 0ae1b1f7
......@@ -36,6 +36,12 @@ public class CtrEstimateMainTag {
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime");
Integer parallelism = parameterTool.getInt("parallelism",2);
String zxJdbcUrl = parameterTool.get("zxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers);
......@@ -46,16 +52,22 @@ public class CtrEstimateMainTag {
System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername);
System.out.println("*** zxPassword: " + zxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword);
System.out.println("**********************************************************");
// 获得流处理环境对象
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// env.enableCheckpointing(1000);
// env.setStateBackend(new FsStateBackend(checkpointPath));
// env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 3000));
// CheckpointConfig config = env.getCheckpointConfig();
// config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.enableCheckpointing(1000);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 3000));
CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
DataStream MaidianDataStream = new MaidianKafkaSource(
env,
......@@ -69,7 +81,21 @@ public class CtrEstimateMainTag {
).getInstance();
// 执行处理核心逻辑
new CtrEstimateTagOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
new CtrEstimateTagOperator(
MaidianDataStream,
jdbcUrl,
maxRetry,
retryInteral,
parallelism,
windowSize,
slideSize,
zxJdbcUrl,
zxUsername,
zxPassword,
jerryJdbcUrl,
jerryUsername,
jerryPassword
).run();
// 常驻执行
env.execute("ctr-estimate-tag");
......
src/main/java/com/gmei/data/ctr/CtrEstimateMainTagDev.java 0 → 100644
View file @ 0ae1b1f7
package com.gmei.data.ctr;
import com.gmei.data.ctr.operator.CtrEstimateTagOperator;
import com.gmei.data.ctr.source.MaidianKafkaSource;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
* @ClassName CtrEstimateMain
* @Description: CTR预估特征实时处理入口
* @Author apple
* @Date 2020/3/30
* @Version V1.0
**/
public class CtrEstimateMainTagDev {
public static void main(String[] args) throws Exception{
// 获取运行参数
ParameterTool parameterTool = ParameterTool.fromArgs(args);
String inBrokers = parameterTool.get("inBrokers","172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092");
String batchSize = parameterTool.get("batchSize","1000");
String maidianInTopic = parameterTool.get("maidianInTopic", "gm-maidian-data");
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-flink-tag");
Integer windowSize = parameterTool.getInt("windowSize",5);
Integer slideSize = parameterTool.getInt("slideSize",5);
String jdbcUrl = parameterTool.get("jdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000);
String checkpointPath = parameterTool.get("checkpointPath","hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint");
Boolean isStartFromEarliest = parameterTool.getBoolean("isStartFromEarliest",true);
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime","2020-04-04 20:42:00");
Integer parallelism = parameterTool.getInt("parallelism",2);
String zxJdbcUrl = parameterTool.get("zxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
// 核心参数打印
System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername);
System.out.println("*** zxPassword: " + zxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword);
System.out.println("**********************************************************");
// 获得流处理环境对象
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream MaidianDataStream = new MaidianKafkaSource(
env,
inBrokers,
maidianInTopic,
maidianInGroupId,
batchSize,
isStartFromEarliest,
isStartFromLatest,
startTime
).getInstance();
// 执行处理核心逻辑
new CtrEstimateTagOperator(
MaidianDataStream,
jdbcUrl,
maxRetry,
retryInteral,
parallelism,
windowSize,
slideSize,
zxJdbcUrl,
zxUsername,
zxPassword,
jerryJdbcUrl,
jerryUsername,
jerryPassword
).run();
// 常驻执行
env.execute("ctr-estimate-tag");
}
}
src/main/java/com/gmei/data/ctr/operator/CtrEstimateTagOperator.java
View file @ 0ae1b1f7
......@@ -32,8 +32,16 @@ public class CtrEstimateTagOperator implements BaseOperator{
private int parallelism;
private int windowSize;
private int slideSize;
private String zxJdbcUrl;
private String zxUsername;
private String zxPassword;
private String jerryJdbcUrl;
private String jerryUsername;
private String jerryPassword;
public CtrEstimateTagOperator(DataStream dataStream, String outJdbcUrl, int maxRetry, long retryInteral, int parallelism,int windowSize,int slideSize) {
public CtrEstimateTagOperator(DataStream dataStream, String outJdbcUrl, int maxRetry, long retryInteral, int parallelism,
int windowSize, int slideSize, String zxJdbcUrl, String zxUsername, String zxPassword,
String jerryJdbcUrl, String jerryUsername, String jerryPassword) {
this.dataStream = dataStream;
this.outJdbcUrl = outJdbcUrl;
this.maxRetry = maxRetry;
......@@ -41,6 +49,12 @@ public class CtrEstimateTagOperator implements BaseOperator{
this.parallelism = parallelism;
this.windowSize = windowSize;
this.slideSize = slideSize;
this.zxJdbcUrl = zxJdbcUrl;
this.zxUsername = zxUsername;
this.zxPassword = zxPassword;
this.jerryJdbcUrl = jerryJdbcUrl;
this.jerryUsername = jerryUsername;
this.jerryPassword = jerryPassword;
}
@Override
......@@ -160,11 +174,11 @@ public class CtrEstimateTagOperator implements BaseOperator{
});
//map.print();
DataStream<DeviceCurrentEstimateTagTmp> tidbAsyncDataStream = AsyncDataStream
.unorderedWait(map, new TidbMysqlAsyncSource(), 1, TimeUnit.MINUTES, 1000)
.unorderedWait(map, new TidbMysqlAsyncSource(jerryJdbcUrl,jerryUsername,jerryPassword), 1, TimeUnit.MINUTES, 1000)
.uid("tidbAsyncDataStream")
.setParallelism(parallelism);
DataStream<DeviceCurrentEstimateTagTmp> zhengxingAsyncDataStream = AsyncDataStream
.unorderedWait(map, new ZhengxingMysqlAsyncSource(), 1, TimeUnit.MINUTES, 1000)
.unorderedWait(map, new ZhengxingMysqlAsyncSource(zxJdbcUrl,zxUsername,zxPassword), 1, TimeUnit.MINUTES, 1000)
.uid("zhengxingAsyncDataStream")
.setParallelism(parallelism);
DataStream<DeviceCurrentEstimateTagTmp> asyncDataStream = tidbAsyncDataStream.union(zhengxingAsyncDataStream);
......
src/main/java/com/gmei/data/ctr/source/TidbMysqlAsyncSource.java
View file @ 0ae1b1f7
......@@ -3,6 +3,7 @@ package com.gmei.data.ctr.source;
import com.alibaba.druid.pool.DruidDataSource;
import com.gmei.data.ctr.bean.CtrEstimateTagEtl;
import com.gmei.data.ctr.bean.DeviceCurrentEstimateTagTmp;
import com.gmei.data.ctr.common.Constants;
import com.gmei.data.ctr.utils.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
......@@ -28,18 +29,28 @@ import static java.util.concurrent.Executors.newFixedThreadPool;
* @Version V1.0
**/
public class TidbMysqlAsyncSource extends RichAsyncFunction<CtrEstimateTagEtl, DeviceCurrentEstimateTagTmp> {
private String jerryJdbcUrl;
private String jerryUsername;
private String jerryPassword;
private transient DruidDataSource dataSource;
private transient ExecutorService executorService;
public TidbMysqlAsyncSource(String jerryJdbcUrl, String jerryUsername, String jerryPassword) {
this.jerryJdbcUrl = jerryJdbcUrl;
this.jerryUsername = jerryUsername;
this.jerryPassword = jerryPassword;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
executorService = newFixedThreadPool(20);
dataSource = new DruidDataSource();
dataSource.setDriverClassName("com.mysql.jdbc.Driver");
dataSource.setUsername("data_user");
dataSource.setPassword("YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
dataSource.setUrl("jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
dataSource.setDriverClassName(Constants.MYSQL_DRIVER_CLASS);
dataSource.setUrl(jerryJdbcUrl);
dataSource.setUsername(jerryUsername);
dataSource.setPassword(jerryPassword);
dataSource.setInitialSize(5);
dataSource.setMinIdle(10);
dataSource.setMaxActive(20);
......
src/main/java/com/gmei/data/ctr/source/ZhengxingMysqlAsyncSource.java
View file @ 0ae1b1f7
......@@ -3,6 +3,7 @@ package com.gmei.data.ctr.source;
import com.alibaba.druid.pool.DruidDataSource;
import com.gmei.data.ctr.bean.CtrEstimateTagEtl;
import com.gmei.data.ctr.bean.DeviceCurrentEstimateTagTmp;
import com.gmei.data.ctr.common.Constants;
import com.gmei.data.ctr.utils.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
......@@ -28,18 +29,27 @@ import static java.util.concurrent.Executors.newFixedThreadPool;
* @Version V1.0
**/
public class ZhengxingMysqlAsyncSource extends RichAsyncFunction<CtrEstimateTagEtl, DeviceCurrentEstimateTagTmp> {
private String zxJdbcUrl;
private String zxUsername;
private String zxPassword;
private transient DruidDataSource dataSource;
private transient ExecutorService executorService;
public ZhengxingMysqlAsyncSource(String zxJdbcUrl, String zxUsername, String zxPassword) {
this.zxJdbcUrl = zxJdbcUrl;
this.zxUsername = zxUsername;
this.zxPassword = zxPassword;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
executorService = newFixedThreadPool(20);
dataSource = new DruidDataSource();
dataSource.setDriverClassName("com.mysql.jdbc.Driver");
dataSource.setUsername("work");
dataSource.setPassword("BJQaT9VzDcuPBqkd");
dataSource.setUrl("jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
dataSource.setDriverClassName(Constants.MYSQL_DRIVER_CLASS);
dataSource.setUrl(zxJdbcUrl);
dataSource.setUsername(zxUsername);
dataSource.setPassword(zxPassword);
dataSource.setInitialSize(5);
dataSource.setMinIdle(10);
dataSource.setMaxActive(20);
......
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