lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature(cli,forest): 增加b5sync集群调度

Browse Source 
为同步程序上服务准备，不在服务级别区分同步和压缩的调度
This commit is contained in:
兰缘小妖
2023-07-11 10:37:33 +08:00
parent 1b591dff6d
commit f0881d65d7
8 changed files with 65 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
service-cli/service-cli-runner/src/main/resources/application.yml
View File

@@ -19,8 +19,8 @@ deploy:
- name: service-scheduler - name: service-scheduler
source-jar: service-scheduler-1.0.0-SNAPSHOT.jar source-jar: service-scheduler-1.0.0-SNAPSHOT.jar
replicas: 1 replicas: 1
- name: service-launcher-compaction-b1 - name: service-launcher-runner-b1
source-jar: service-launcher-compaction-b2b1-1.0.0-SNAPSHOT.jar source-jar: service-launcher-runner-b2b1-1.0.0-SNAPSHOT.jar
replicas: 6 replicas: 6
environments: environments:
connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM
@@ -29,12 +29,13 @@ deploy:
connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path} connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path}
connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url} connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url}
arguments: arguments:
spring_application_name: service-launcher-compaction-b1 spring_application_name: service-launcher-runner-b1
connector_cluster_cluster: b1 connector_cluster_cluster: b1
connector_cluster_queue-name: compaction-queue-b1 connector_cluster_sync-queue-name: sync-queue-b1
connector_cluster_compaction-queue-name: compaction-queue-b1
connector_zookeeper_connect-url: ${deploy.runtime.zk-url} connector_zookeeper_connect-url: ${deploy.runtime.zk-url}
- name: service-launcher-compaction-b5 - name: service-launcher-runner-b5
source-jar: service-launcher-compaction-b2b5-1.0.0-SNAPSHOT.jar source-jar: service-launcher-runner-b2b5-1.0.0-SNAPSHOT.jar
replicas: 6 replicas: 6
environments: environments:
connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM
@@ -43,12 +44,13 @@ deploy:
connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path} connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path}
connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url} connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url}
arguments: arguments:
spring_application_name: service-launcher-compaction-b5 spring_application_name: service-launcher-runner-b5
connector_cluster_cluster: b5 connector_cluster_cluster: b5
connector_cluster_queue-name: compaction-queue-b5 connector_cluster_sync-queue-name: sync-queue-b5
connector_cluster_compaction-queue-name: compaction-queue-b5
connector_zookeeper_connect-url: ${deploy.runtime.zk-url} connector_zookeeper_connect-url: ${deploy.runtime.zk-url}
- name: service-launcher-compaction-a4 - name: service-launcher-runner-b5-sync
source-jar: service-launcher-compaction-b2a4-1.0.0-SNAPSHOT.jar source-jar: service-launcher-runner-b2b5sync-1.0.0-SNAPSHOT.jar
replicas: 6 replicas: 6
environments: environments:
connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM
@@ -57,9 +59,25 @@ deploy:
connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path} connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path}
connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url} connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url}
arguments: arguments:
spring_application_name: service-launcher-compaction-a4 spring_application_name: service-launcher-runner-b5-sync
connector_cluster_cluster: b5-sync
connector_cluster_sync-queue-name: sync-queue-b5-sync
connector_cluster_compaction-queue-name: compaction-queue-b5-sync
connector_zookeeper_connect-url: ${deploy.runtime.zk-url}
- name: service-launcher-runner-a4
source-jar: service-launcher-runner-b2a4-1.0.0-SNAPSHOT.jar
replicas: 6
environments:
connector_hadoop_kerberos-principal: ${deploy.runtime.user}/$\{hostname}.hdp.dc@ECLD.COM
connector_hadoop_kerberos-keytab-path: ${deploy.runtime.kerberos-keytab-path}
connector_hudi_app-hdfs-path: ${deploy.runtime.hudi.app-hdfs-path}
connector_hudi_archive-hdfs-path: ${deploy.runtime.hudi.archive-hdfs-path}
connector_hudi_victoria-push-url: ${deploy.runtime.hudi.victoria-push-url}
arguments:
spring_application_name: service-launcher-runner-a4
connector_cluster_cluster: a4 connector_cluster_cluster: a4
connector_cluster_queue-name: compaction-queue-a4 connector_cluster_sync-queue-name: sync-queue-a4
connector_cluster_compaction-queue-name: compaction-queue-a4
connector_zookeeper_connect-url: ${deploy.runtime.zk-url} connector_zookeeper_connect-url: ${deploy.runtime.zk-url}
- name: service-info-query - name: service-info-query
source-jar: service-info-query-1.0.0-SNAPSHOT.jar source-jar: service-info-query-1.0.0-SNAPSHOT.jar

2
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/A4LauncherService.java
View File

@@ -6,6 +6,6 @@ import com.dtflys.forest.annotation.BaseRequest;
* @author lanyuanxiaoyao * @author lanyuanxiaoyao
* @date 2023-06-06 * @date 2023-06-06
*/ */
@BaseRequest(baseURL = "http://service-launcher-compaction-a4") @BaseRequest(baseURL = "http://service-launcher-runner-a4")
public interface A4LauncherService extends LauncherService{ public interface A4LauncherService extends LauncherService{
} }

2
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/B1LauncherService.java
View File

@@ -8,6 +8,6 @@ import com.dtflys.forest.annotation.Query;
* @author lanyuanxiaoyao * @author lanyuanxiaoyao
* @date 2023-06-06 * @date 2023-06-06
*/ */
@BaseRequest(baseURL = "http://service-launcher-compaction-b1") @BaseRequest(baseURL = "http://service-launcher-runner-b1")
public interface B1LauncherService extends LauncherService{ public interface B1LauncherService extends LauncherService{
} }

2
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/B5LauncherService.java
View File

@@ -6,6 +6,6 @@ import com.dtflys.forest.annotation.BaseRequest;
* @author lanyuanxiaoyao * @author lanyuanxiaoyao
* @date 2023-06-06 * @date 2023-06-06
*/ */
@BaseRequest(baseURL = "http://service-launcher-compaction-b5") @BaseRequest(baseURL = "http://service-launcher-runner-b5")
public interface B5LauncherService extends LauncherService{ public interface B5LauncherService extends LauncherService{
} }

11
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/B5SyncLauncherService.java Normal file
View File

@@ -0,0 +1,11 @@
package com.lanyuanxiaoyao.service.forest.service.launcher;
import com.dtflys.forest.annotation.BaseRequest;
/**
* @author lanyuanxiaoyao
* @date 2023-06-06
*/
@BaseRequest(baseURL = "http://service-launcher-runner-b5-sync")
public interface B5SyncLauncherService extends LauncherService{
}

8
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/LauncherService.java
View File

@@ -8,9 +8,9 @@ import com.dtflys.forest.annotation.Query;
* @date 2023-06-06 * @date 2023-06-06
*/ */
public interface LauncherService { public interface LauncherService {
@Get("/compaction/stop") @Get("/launcher/compaction/stop")
void stop(@Query("flink_job_id") Long flinkJobId, @Query("alias") String alias); void compactionStop(@Query("flink_job_id") Long flinkJobId, @Query("alias") String alias);
@Get("/compaction/stop_app") @Get("/launcher/compaction/stop_app")
void stopApp(@Query("application_id") String applicationId); void compactionStopApp(@Query("application_id") String applicationId);
} }

25
service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/launcher/LaunchersService.java
View File

@@ -22,15 +22,18 @@ public class LaunchersService {
public LaunchersService( public LaunchersService(
A4LauncherService a4LauncherService, A4LauncherService a4LauncherService,
B1LauncherService b1LauncherService, B1LauncherService b1LauncherService,
B5LauncherService b5LauncherService B5LauncherService b5LauncherService,
) { B5SyncLauncherService b5SyncLauncherService
) {
serviceMap = Maps.immutable.of( serviceMap = Maps.immutable.of(
Constants.CLUSTER_A4, Constants.CLUSTER_A4,
a4LauncherService, a4LauncherService,
Constants.CLUSTER_B1, Constants.CLUSTER_B1,
b1LauncherService, b1LauncherService,
Constants.CLUSTER_B5, Constants.CLUSTER_B5,
b5LauncherService b5LauncherService,
Constants.CLUSTER_B5_SYNC,
b5SyncLauncherService
); );
} }
@@ -45,25 +48,25 @@ public class LaunchersService {
return serviceMap.valuesView().toList().toImmutable(); return serviceMap.valuesView().toList().toImmutable();
} }
public void stop(Long flinkJobId, String alias) { public void compactionStop(Long flinkJobId, String alias) {
for (LauncherService service : getServices()) { for (LauncherService service : getServices()) {
service.stop(flinkJobId, alias); service.compactionStop(flinkJobId, alias);
} }
} }
public void stop(String cluster, Long flinkJobId, String alias) { public void compactionStop(String cluster, Long flinkJobId, String alias) {
LauncherService service = getService(cluster); LauncherService service = getService(cluster);
service.stop(flinkJobId, alias); service.compactionStop(flinkJobId, alias);
} }
public void stopApp(String applicationId) { public void compactionStopApp(String applicationId) {
for (LauncherService service : getServices()) { for (LauncherService service : getServices()) {
service.stopApp(applicationId); service.compactionStopApp(applicationId);
} }
} }
public void stopApp(String cluster, String applicationId) { public void compactionStopApp(String cluster, String applicationId) {
LauncherService service = getService(cluster); LauncherService service = getService(cluster);
service.stopApp(applicationId); service.compactionStopApp(applicationId);
} }
} }

3
test/test.http
View File

@@ -47,5 +47,8 @@ GET http://{{username}}:{{password}}@132.122.116.150:27510/table/list_compaction
### Query Scheduler ### Query Scheduler
GET {{scheduler-url}}/schedule/all_focus GET {{scheduler-url}}/schedule/all_focus
### schedule table force
GET {{scheduler-url}}/schedule/table?flink_job_id=1542097993662164992&alias=acct_oa_attach_dg&force=b5-sync
### Yarn query ### Yarn query
GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.122.116.143:34903/job/list?cluster=a4 GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.122.116.143:34903/job/list?cluster=a4