Skip to content

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

update codes

parent 410048e3
Hide whitespace changes
Inline Side-by-side
Showing with 104 additions and 101 deletions
bin/start.sh
View file @ 73fbf112
...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \ ...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \
--maidianInGroupId 'prod-ctr-estimate' \ --maidianInGroupId 'prod-ctr-estimate' \
--windowSize 5 \ --windowSize 5 \
--slideSize 5 \ --slideSize 5 \
--jdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \ --outJdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \
--maxRetry 3 \ --maxRetry 3 \
--retryInteral 3000 \ --retryInteral 3000 \
--checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate/checkpoint' \ --checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate/checkpoint' \
......
bin/start_clk.sh
View file @ 73fbf112
...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \ ...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \
--maidianInGroupId 'test-ctr-estimate-clk' \ --maidianInGroupId 'test-ctr-estimate-clk' \
--windowSize 5 \ --windowSize 5 \
--slideSize 5 \ --slideSize 5 \
--jdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \ --outJdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \
--maxRetry 3 \ --maxRetry 3 \
--retryInteral 3000 \ --retryInteral 3000 \
--checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-clk/checkpoint' \ --checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-clk/checkpoint' \
......
bin/start_tag.sh
View file @ 73fbf112
...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \ ...@@ -22,7 +22,7 @@ $JAR_DIR/ctr-estimate-1.0-SNAPSHOT.jar \
--maidianInGroupId 'test-ctr-estimate-tag' \ --maidianInGroupId 'test-ctr-estimate-tag' \
--windowSize 5 \ --windowSize 5 \
--slideSize 5 \ --slideSize 5 \
--jdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \ --outJdbcUrl 'jdbc:mysql://172.16.40.170:4000/jerry_test?user=data_user&password=YPEzp78HQBuhByWPpefQu6X3D6hEPfD6&autoReconnect=true&useSSL=false' \
--maxRetry 3 \ --maxRetry 3 \
--retryInteral 3000 \ --retryInteral 3000 \
--checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint' \ --checkpointPath 'hdfs://bj-gmei-hdfs/user/data/flink/ctr-estimate-tag/checkpoint' \
......
src/main/java/com/gmei/data/ctr/DevCtrEstimateMainClk.java
View file @ 73fbf112
...@@ -27,7 +27,7 @@ public class DevCtrEstimateMainClk { ...@@ -27,7 +27,7 @@ public class DevCtrEstimateMainClk {
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk"); String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk");
Integer windowSize = parameterTool.getInt("windowSize",60); Integer windowSize = parameterTool.getInt("windowSize",60);
Integer slideSize = parameterTool.getInt("slideSize",60); Integer slideSize = parameterTool.getInt("slideSize",60);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -41,7 +41,7 @@ public class DevCtrEstimateMainClk { ...@@ -41,7 +41,7 @@ public class DevCtrEstimateMainClk {
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
...@@ -69,7 +69,7 @@ public class DevCtrEstimateMainClk { ...@@ -69,7 +69,7 @@ public class DevCtrEstimateMainClk {
).getInstance(); ).getInstance();
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateClkOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run(); new CtrEstimateClkOperator(MaidianDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
// 常驻执行 // 常驻执行
env.execute("ctr-estimate-clk"); env.execute("ctr-estimate-clk");
......
src/main/java/com/gmei/data/ctr/DevCtrEstimateMainTag.java
View file @ 73fbf112
...@@ -27,7 +27,7 @@ public class DevCtrEstimateMainTag { ...@@ -27,7 +27,7 @@ public class DevCtrEstimateMainTag {
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-tag"); String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-tag");
Integer windowSize = parameterTool.getInt("windowSize",5); Integer windowSize = parameterTool.getInt("windowSize",5);
Integer slideSize = parameterTool.getInt("slideSize",5); Integer slideSize = parameterTool.getInt("slideSize",5);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -36,28 +36,29 @@ public class DevCtrEstimateMainTag { ...@@ -36,28 +36,29 @@ public class DevCtrEstimateMainTag {
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false); Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime"); String startTime = parameterTool.get("startTime");
Integer parallelism = parameterTool.getInt("parallelism",2); 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 inZxJdbcUrl = parameterTool.get("inZxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work"); String inZxUsername = parameterTool.get("inZxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd"); String inZxPassword = parameterTool.get("inZxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false"); String inJerryJdbcUrl = parameterTool.get("inJerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user"); String inJerryUsername = parameterTool.get("inJerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6"); String inJerryPassword = parameterTool.get("inJerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
// 核心参数打印
System.out.println("**********************************************************"); System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize); System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl); System.out.println("*** inZxJdbcUrl: " + inZxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername); System.out.println("*** inZxUsername: " + inZxUsername);
System.out.println("*** zxPassword: " + zxPassword); System.out.println("*** inZxPassword: " + inZxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl); System.out.println("*** inJerryJdbcUrl: " + inJerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername); System.out.println("*** inJerryUsername: " + inJerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword); System.out.println("*** inJerryPassword: " + inJerryPassword);
System.out.println("**********************************************************"); System.out.println("**********************************************************");
// 获得流处理环境对象 // 获得流处理环境对象
...@@ -83,18 +84,18 @@ public class DevCtrEstimateMainTag { ...@@ -83,18 +84,18 @@ public class DevCtrEstimateMainTag {
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateTagOperator( new CtrEstimateTagOperator(
MaidianDataStream, MaidianDataStream,
jdbcUrl, outJdbcUrl,
maxRetry, maxRetry,
retryInteral, retryInteral,
parallelism, parallelism,
windowSize, windowSize,
slideSize, slideSize,
zxJdbcUrl, inZxJdbcUrl,
zxUsername, inZxUsername,
zxPassword, inZxPassword,
jerryJdbcUrl, inJerryJdbcUrl,
jerryUsername, inJerryUsername,
jerryPassword inJerryPassword
).run(); ).run();
// 常驻执行 // 常驻执行
......
src/main/java/com/gmei/data/ctr/ProdCtrEstimateMain.java
View file @ 73fbf112
...@@ -28,7 +28,7 @@ public class ProdCtrEstimateMain { ...@@ -28,7 +28,7 @@ public class ProdCtrEstimateMain {
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate"); String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate");
Integer windowSize = parameterTool.getInt("windowSize",60); Integer windowSize = parameterTool.getInt("windowSize",60);
Integer slideSize = parameterTool.getInt("slideSize",60); Integer slideSize = parameterTool.getInt("slideSize",60);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -37,29 +37,29 @@ public class ProdCtrEstimateMain { ...@@ -37,29 +37,29 @@ public class ProdCtrEstimateMain {
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false); Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime"); String startTime = parameterTool.get("startTime");
Integer parallelism = parameterTool.getInt("parallelism",2); 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 inZxJdbcUrl = parameterTool.get("inZxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work"); String inZxUsername = parameterTool.get("inZxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd"); String inZxPassword = parameterTool.get("inZxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false"); String inJerryJdbcUrl = parameterTool.get("inJerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user"); String inJerryUsername = parameterTool.get("inJerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6"); String inJerryPassword = parameterTool.get("inJerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
// 核心参数打印 // 核心参数打印
System.out.println("**********************************************************"); System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize); System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl); System.out.println("*** inZxJdbcUrl: " + inZxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername); System.out.println("*** inZxUsername: " + inZxUsername);
System.out.println("*** zxPassword: " + zxPassword); System.out.println("*** inZxPassword: " + inZxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl); System.out.println("*** inJerryJdbcUrl: " + inJerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername); System.out.println("*** inJerryUsername: " + inJerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword); System.out.println("*** inJerryPassword: " + inJerryPassword);
System.out.println("**********************************************************"); System.out.println("**********************************************************");
// 获得流处理环境对象 // 获得流处理环境对象
...@@ -84,27 +84,28 @@ public class ProdCtrEstimateMain { ...@@ -84,27 +84,28 @@ public class ProdCtrEstimateMain {
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateClkOperator(MaidianDataStream, new CtrEstimateClkOperator(MaidianDataStream,
jdbcUrl, outJdbcUrl,
maxRetry, maxRetry,
retryInteral, retryInteral,
parallelism, parallelism,
windowSize, windowSize,
slideSize slideSize
).run(); ).run();
// 执行处理核心逻辑
new CtrEstimateTagOperator( new CtrEstimateTagOperator(
MaidianDataStream, MaidianDataStream,
jdbcUrl, outJdbcUrl,
maxRetry, maxRetry,
retryInteral, retryInteral,
parallelism, parallelism,
windowSize, windowSize,
slideSize, slideSize,
zxJdbcUrl, inZxJdbcUrl,
zxUsername, inZxUsername,
zxPassword, inZxPassword,
jerryJdbcUrl, inJerryJdbcUrl,
jerryUsername, inJerryUsername,
jerryPassword inJerryPassword
).run(); ).run();
// 常驻执行 // 常驻执行
......
src/main/java/com/gmei/data/ctr/ProdCtrEstimateMainClk.java
View file @ 73fbf112
...@@ -27,7 +27,7 @@ public class ProdCtrEstimateMainClk { ...@@ -27,7 +27,7 @@ public class ProdCtrEstimateMainClk {
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk"); String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk");
Integer windowSize = parameterTool.getInt("windowSize",60); Integer windowSize = parameterTool.getInt("windowSize",60);
Integer slideSize = parameterTool.getInt("slideSize",60); Integer slideSize = parameterTool.getInt("slideSize",60);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -41,7 +41,7 @@ public class ProdCtrEstimateMainClk { ...@@ -41,7 +41,7 @@ public class ProdCtrEstimateMainClk {
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
...@@ -69,7 +69,7 @@ public class ProdCtrEstimateMainClk { ...@@ -69,7 +69,7 @@ public class ProdCtrEstimateMainClk {
).getInstance(); ).getInstance();
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateClkOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run(); new CtrEstimateClkOperator(MaidianDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
// 常驻执行 // 常驻执行
env.execute("ctr-estimate-clk"); env.execute("ctr-estimate-clk");
......
src/main/java/com/gmei/data/ctr/ProdCtrEstimateMainTag.java
View file @ 73fbf112
...@@ -27,7 +27,7 @@ public class ProdCtrEstimateMainTag { ...@@ -27,7 +27,7 @@ public class ProdCtrEstimateMainTag {
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-tag"); String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-tag");
Integer windowSize = parameterTool.getInt("windowSize",5); Integer windowSize = parameterTool.getInt("windowSize",5);
Integer slideSize = parameterTool.getInt("slideSize",5); Integer slideSize = parameterTool.getInt("slideSize",5);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -36,28 +36,28 @@ public class ProdCtrEstimateMainTag { ...@@ -36,28 +36,28 @@ public class ProdCtrEstimateMainTag {
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false); Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime"); String startTime = parameterTool.get("startTime");
Integer parallelism = parameterTool.getInt("parallelism",2); 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 inZxJdbcUrl = parameterTool.get("inZxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work"); String inZxUsername = parameterTool.get("inZxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd"); String inZxPassword = parameterTool.get("inZxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false"); String inJerryJdbcUrl = parameterTool.get("inJerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user"); String inJerryUsername = parameterTool.get("inJerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6"); String inJerryPassword = parameterTool.get("inJerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
System.out.println("**********************************************************"); System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize); System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl); System.out.println("*** inZxJdbcUrl: " + inZxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername); System.out.println("*** inZxUsername: " + inZxUsername);
System.out.println("*** zxPassword: " + zxPassword); System.out.println("*** inZxPassword: " + inZxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl); System.out.println("*** inJerryJdbcUrl: " + inJerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername); System.out.println("*** inJerryUsername: " + inJerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword); System.out.println("*** inJerryPassword: " + inJerryPassword);
System.out.println("**********************************************************"); System.out.println("**********************************************************");
// 获得流处理环境对象 // 获得流处理环境对象
...@@ -80,21 +80,22 @@ public class ProdCtrEstimateMainTag { ...@@ -80,21 +80,22 @@ public class ProdCtrEstimateMainTag {
startTime startTime
).getInstance(); ).getInstance();
// 执行处理核心逻辑
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateTagOperator( new CtrEstimateTagOperator(
MaidianDataStream, MaidianDataStream,
jdbcUrl, outJdbcUrl,
maxRetry, maxRetry,
retryInteral, retryInteral,
parallelism, parallelism,
windowSize, windowSize,
slideSize, slideSize,
zxJdbcUrl, inZxJdbcUrl,
zxUsername, inZxUsername,
zxPassword, inZxPassword,
jerryJdbcUrl, inJerryJdbcUrl,
jerryUsername, inJerryUsername,
jerryPassword inJerryPassword
).run(); ).run();
// 常驻执行 // 常驻执行
......
src/main/java/com/gmei/data/ctr/TestCtrEstimateMainClk.java
View file @ 73fbf112
...@@ -21,13 +21,13 @@ public class TestCtrEstimateMainClk { ...@@ -21,13 +21,13 @@ public class TestCtrEstimateMainClk {
public static void main(String[] args) throws Exception{ public static void main(String[] args) throws Exception{
// 获取运行参数 // 获取运行参数
ParameterTool parameterTool = ParameterTool.fromArgs(args); ParameterTool parameterTool = ParameterTool.fromArgs(args);
String inBrokers = parameterTool.get("inBrokers","test003:9092"); 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 batchSize = parameterTool.get("batchSize","1000");
String maidianInTopic = parameterTool.get("maidianInTopic", "test11"); String maidianInTopic = parameterTool.get("maidianInTopic", "gm-maidian-data");
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-clk"); String maidianInGroupId = parameterTool.get("maidianInGroupId","test-ctr-estimate-clk");
Integer windowSize = parameterTool.getInt("windowSize",60); Integer windowSize = parameterTool.getInt("windowSize",60);
Integer slideSize = parameterTool.getInt("slideSize",60); Integer slideSize = parameterTool.getInt("slideSize",60);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -41,7 +41,7 @@ public class TestCtrEstimateMainClk { ...@@ -41,7 +41,7 @@ public class TestCtrEstimateMainClk {
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
...@@ -64,7 +64,7 @@ public class TestCtrEstimateMainClk { ...@@ -64,7 +64,7 @@ public class TestCtrEstimateMainClk {
).getInstance(); ).getInstance();
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateClkOperator(MaidianDataStream,jdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run(); new CtrEstimateClkOperator(MaidianDataStream,outJdbcUrl,maxRetry,retryInteral,parallelism,windowSize,slideSize).run();
// 常驻执行 // 常驻执行
env.execute("ctr-estimate-clk"); env.execute("ctr-estimate-clk");
......
src/main/java/com/gmei/data/ctr/TestCtrEstimateMainTag.java
View file @ 73fbf112
...@@ -21,10 +21,10 @@ public class TestCtrEstimateMainTag { ...@@ -21,10 +21,10 @@ public class TestCtrEstimateMainTag {
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","172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092");
String batchSize = parameterTool.get("batchSize","1000"); String batchSize = parameterTool.get("batchSize","1000");
String maidianInTopic = parameterTool.get("maidianInTopic", "gm-maidian-data"); String maidianInTopic = parameterTool.get("maidianInTopic", "gm-maidian-data");
String maidianInGroupId = parameterTool.get("maidianInGroupId","ctr-estimate-flink-tag"); String maidianInGroupId = parameterTool.get("maidianInGroupId","test-ctr-estimate-tag");
Integer windowSize = parameterTool.getInt("windowSize",5); Integer windowSize = parameterTool.getInt("windowSize",5);
Integer slideSize = parameterTool.getInt("slideSize",5); Integer slideSize = parameterTool.getInt("slideSize",5);
String jdbcUrl = parameterTool.get("jdbcUrl", String outJdbcUrl = parameterTool.get("outJdbcUrl",
"jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false"); "jdbc:mysql://172.18.44.3:3306/jerry_test?user=root&password=5OqYM^zLwotJ3oSo&autoReconnect=true&useSSL=false");
Integer maxRetry = parameterTool.getInt("maxRetry",3); Integer maxRetry = parameterTool.getInt("maxRetry",3);
Long retryInteral = parameterTool.getLong("retryInteral",3000); Long retryInteral = parameterTool.getLong("retryInteral",3000);
...@@ -33,29 +33,29 @@ public class TestCtrEstimateMainTag { ...@@ -33,29 +33,29 @@ public class TestCtrEstimateMainTag {
Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false); Boolean isStartFromLatest = parameterTool.getBoolean("isStartFromLatest",false);
String startTime = parameterTool.get("startTime","2020-04-04 20:42:00"); String startTime = parameterTool.get("startTime","2020-04-04 20:42:00");
Integer parallelism = parameterTool.getInt("parallelism",2); 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 inZxJdbcUrl = parameterTool.get("inZxJdbcUrl","jdbc:mysql://172.16.30.141:3306/zhengxing?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String zxUsername = parameterTool.get("zxUsername","work"); String inZxUsername = parameterTool.get("inZxUsername","work");
String zxPassword = parameterTool.get("zxPassword","BJQaT9VzDcuPBqkd"); String inZxPassword = parameterTool.get("inZxPassword","BJQaT9VzDcuPBqkd");
String jerryJdbcUrl = parameterTool.get("jerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false"); String inJerryJdbcUrl = parameterTool.get("inJerryJdbcUrl","jdbc:mysql://172.16.40.170:4000/jerry_test?characterEncoding=UTF-8&autoReconnect=true&useSSL=false");
String jerryUsername = parameterTool.get("jerryUsername","data_user"); String inJerryUsername = parameterTool.get("inJerryUsername","data_user");
String jerryPassword = parameterTool.get("jerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6"); String inJerryPassword = parameterTool.get("inJerryPassword","YPEzp78HQBuhByWPpefQu6X3D6hEPfD6");
// 核心参数打印 // 核心参数打印
System.out.println("**********************************************************"); System.out.println("**********************************************************");
System.out.println("*** inBrokers: " + inBrokers); System.out.println("*** inBrokers: " + inBrokers);
System.out.println("*** maidianInTopic: "+ maidianInTopic); System.out.println("*** maidianInTopic: "+ maidianInTopic);
System.out.println("*** maidianInGroupId: " + maidianInGroupId); System.out.println("*** maidianInGroupId: " + maidianInGroupId);
System.out.println("*** jdbcUrl: " + jdbcUrl); System.out.println("*** outJdbcUrl: " + outJdbcUrl);
System.out.println("*** checkpointPath: " + checkpointPath); System.out.println("*** checkpointPath: " + checkpointPath);
System.out.println("*** startTime: " + startTime); System.out.println("*** startTime: " + startTime);
System.out.println("*** windowSize: " + windowSize); System.out.println("*** windowSize: " + windowSize);
System.out.println("*** slideSize: " + slideSize); System.out.println("*** slideSize: " + slideSize);
System.out.println("*** zxJdbcUrl: " + zxJdbcUrl); System.out.println("*** inZxJdbcUrl: " + inZxJdbcUrl);
System.out.println("*** zxUsername: " + zxUsername); System.out.println("*** inZxUsername: " + inZxUsername);
System.out.println("*** zxPassword: " + zxPassword); System.out.println("*** inZxPassword: " + inZxPassword);
System.out.println("*** jerryJdbcUrl: " + jerryJdbcUrl); System.out.println("*** inJerryJdbcUrl: " + inJerryJdbcUrl);
System.out.println("*** jerryUsername: " + jerryUsername); System.out.println("*** inJerryUsername: " + inJerryUsername);
System.out.println("*** jerryPassword: " + jerryPassword); System.out.println("*** inJerryPassword: " + inJerryPassword);
System.out.println("**********************************************************"); System.out.println("**********************************************************");
// 获得流处理环境对象 // 获得流处理环境对象
...@@ -76,18 +76,18 @@ public class TestCtrEstimateMainTag { ...@@ -76,18 +76,18 @@ public class TestCtrEstimateMainTag {
// 执行处理核心逻辑 // 执行处理核心逻辑
new CtrEstimateTagOperator( new CtrEstimateTagOperator(
MaidianDataStream, MaidianDataStream,
jdbcUrl, outJdbcUrl,
maxRetry, maxRetry,
retryInteral, retryInteral,
parallelism, parallelism,
windowSize, windowSize,
slideSize, slideSize,
zxJdbcUrl, inZxJdbcUrl,
zxUsername, inZxUsername,
zxPassword, inZxPassword,
jerryJdbcUrl, inJerryJdbcUrl,
jerryUsername, inJerryUsername,
jerryPassword inJerryPassword
).run(); ).run();
// 常驻执行 // 常驻执行
......
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