feat(scheduler): 禁止b5、a4在闲时调度期间使用

停止11、14点的全表压缩调度

.idea/httpRequests/http-requests-log.http

@@ -1,3 +1,168 @@
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s20.hdp.dc:19521/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+content-length: 0
+
+<> 2024-10-12T154854.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T154825.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T154754.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T154616.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/list?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T154529.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/list?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T151839.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/file_count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T151753.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/file_count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=EE0952421D19909D0A80BB5A1216DE93
+content-length: 0
+
+<> 2024-10-12T151727.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s12.hdp.dc:25961/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+content-length: 0
+
+<> 2024-10-12T151704.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s22.hdp.dc:13241/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=D4B48AD7708DF28D7AFA0A74B26CF45A
+content-length: 0
+
+<> 2024-10-12T151540.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s22.hdp.dc:13241/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=D4B48AD7708DF28D7AFA0A74B26CF45A
+content-length: 0
+
+<> 2024-10-12T151442.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s22.hdp.dc:13241/hdfs/count?root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=D4B48AD7708DF28D7AFA0A74B26CF45A
+content-length: 0
+
+<> 2024-10-12T151417.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s22.hdp.dc:13241/hdfs/count?hdfs=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+Cookie: JSESSIONID=D4B48AD7708DF28D7AFA0A74B26CF45A
+content-length: 0
+
+<> 2024-10-12T151409.500.txt
+
+###
+
+GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s22.hdp.dc:13241/hdfs?hdfs=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data
+Authorization: Basic AxhEbscwsJDbYMH2 cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4
+User-Agent: IntelliJ HTTP Client/IntelliJ IDEA 2024.1.4
+Accept-Encoding: br, deflate, gzip, x-gzip
+Accept: */*
+content-length: 0
+
+<> 2024-10-12T151340.404.json
+
+###
+
 GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:31719/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_sz/acct_item_755&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
 Connection: Keep-Alive
 User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
@@ -358,163 +523,3 @@ Accept-Encoding: br,deflate,gzip,x-gzip
 
 ###
 
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=08E78CE8926806AAB5D110D0FE9B05F7
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T164901.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=08E78CE8926806AAB5D110D0FE9B05F7
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T164758.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=08E78CE8926806AAB5D110D0FE9B05F7
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T164303.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=08E78CE8926806AAB5D110D0FE9B05F7
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T164220.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=08E78CE8926806AAB5D110D0FE9B05F7
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T164107.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s10.hdp.dc:33535/task/law_enforcement?pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&pulsar_topic=persistent://odcp/acct_dg/acct_item_760&start_time=1716858000000&end_time=1716861600000&primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=C5D2666661F27F68E53223FE5B74AF35
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-28T163410.200.txt
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.126.207.130:35690/hudi_services/queue/queue/clear?name=compaction-queue-pre
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=8516C92140B5118AF9AA61025D0F8C93
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.126.207.130:35690/hudi_services/service_scheduler/schedule/all
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=7A4C34E0240A98C1186F3A2551BC5E80
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.126.207.130:35690/hudi_services/service_web/cloud/list
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=F5F155198FAF72435339CC2E21B873CC
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-<> 2024-05-09T170723.200.json
-
-###
-
-GET http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@132.126.207.130:35690/hudi_services/hudi_api/api/message_id?flink_job_id=1542097984132706304&alias=crm_cfguse_mkt_cam_strategy_rel
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=F5F155198FAF72435339CC2E21B873CC
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-###
-
-POST http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s8.hdp.dc:15391/hdfs/write?root=hdfs://b2/apps/datalake/test/test.txt&overwrite=true
-Content-Type: text/plain
-Content-Length: 738
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=E6FF5447C8553BA4268979B8C5779363
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-\#Properties saved on 2023-12-26T09:18:39.583Z
-\#Tue Dec 26 17:18:39 CST 2023
-hoodie.table.precombine.field=update_ts
-hoodie.datasource.write.drop.partition.columns=false
-hoodie.table.partition.fields=CITY_ID
-hoodie.table.type=MERGE_ON_READ
-hoodie.archivelog.folder=archived
-hoodie.compaction.payload.class=org.apache.hudi.common.model.OverwriteWithLatestAvroPayload
-hoodie.timeline.layout.version=1
-hoodie.table.version=4
-hoodie.table.recordkey.fields=_key
-hoodie.datasource.write.partitionpath.urlencode=false
-hoodie.table.name=dws_account
-hoodie.table.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
-hoodie.table.timeline.timezone=LOCAL
-hoodie.datasource.write.hive_style_partitioning=false
-hoodie.table.checksum=989688289
-
-###
-
-POST http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s8.hdp.dc:15391/hdfs/write?root=hdfs://b2/apps/datalake/test/test.txt&overwrite=true
-Content-Length: 11
-Content-Type: */*; charset=UTF-8
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=D12E206603C453F1429C0B7DF1519A4B
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-Hello world
-
-###
-
-POST http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s8.hdp.dc:34469/hdfs/write?root=hdfs://b2/apps/datalake/test/test.txt&overwrite=true
-Content-Length: 11
-Content-Type: */*; charset=UTF-8
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=D12E206603C453F1429C0B7DF1519A4B
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-Hello world
-
-###
-
-POST http://AxhEbscwsJDbYMH2:cYxg3b4PtWoVD5SjFayWxtnSVsjzRsg4@b12s8.hdp.dc:34469/hdfs/write?root=hdfs://b2/apps/datalake/test/test.txt
-Content-Length: 11
-Content-Type: */*; charset=UTF-8
-Connection: Keep-Alive
-User-Agent: Apache-HttpClient/4.5.14 (Java/17.0.10)
-Cookie: JSESSIONID=D12E206603C453F1429C0B7DF1519A4B
-Accept-Encoding: br,deflate,gzip,x-gzip
-
-Hello world
-
-<> 2024-05-08T095641.500.txt
-
-###
-

service-command-pro/src/main/java/com/lanyuanxiaoyao/service/command/pro/commands/HudiCommand.java

@@ -205,6 +205,41 @@ public class HudiCommand {
         fileSystem.close();
     }
 
+    @ShellMethod("Max meta files")
+    public void maxMetaFiles() throws IOException {
+        MutableList<P> list = Lists.mutable.<P>empty().asSynchronized();
+        FileSystem fileSystem = FileSystem.get(new Configuration());
+        infoService
+                .tableMetaList()
+                .collect(TableMeta::getHudi)
+                .collect(TableMeta.HudiMeta::getTargetHdfsPath)
+                .asParallel(ExecutorProvider.EXECUTORS_20, 1)
+                .forEach(hdfs -> {
+                    Path root = new Path(hdfs, ".hoodie");
+                    try {
+                        FileStatus[] statuses = fileSystem.listStatus(root);
+                        long num = 0;
+                        for (FileStatus status : statuses) {
+                            if (status.isFile()) {
+                                num++;
+                            }
+                            if (StrUtil.containsIgnoreCase(status.getPath().toString(), "INVALID")) {
+                                logger.info("{}", status.getPath().toString());
+                            }
+                        }
+                        list.add(new P(num, hdfs));
+                        logger.info("Count: {} Hdfs: {}", num, hdfs);
+                    } catch (IOException e) {
+                        logger.warn("List file error", e);
+                    }
+                });
+        MutableList<P> listP = list.select(p -> p.count > 1000);
+        for (P maxP : listP) {
+            logger.info("Max: {} Hdfs: {}", maxP.count, maxP.hdfs);
+        }
+        fileSystem.close();
+    }
+
     @ShellMethod("Get timeline instants")
     public void timelineInstant(@ShellOption(help = "root hdfs path") String hdfs) {
         hudiService.timelineHdfsAllActive(hdfs).forEach(instant -> logger.info(instant.toString()));
@@ -229,4 +264,14 @@ public class HudiCommand {
     public interface Runnable {
         void run(LongAdder counter);
     }
+
+    private static final class P {
+        String hdfs;
+        long count;
+
+        public P(long count, String hdfs) {
+            this.count = count;
+            this.hdfs = hdfs;
+        }
+    }
 }

service-common/src/main/java/com/lanyuanxiaoyao/service/common/Constants.java

@@ -128,6 +128,12 @@ public interface Constants {
     String METRICS_PULSAR_PREFIX = METRICS_PREFIX + "_pulsar";
     String METRICS_PULSAR_BACKLOG = METRICS_PULSAR_PREFIX + "_backlog";
 
+    String METRICS_HUDI_TABLE = METRICS_PREFIX + "_hudi_table";
+    String METRICS_HUDI_TABLE_FILE_COUNT = METRICS_HUDI_TABLE + "_file_count";
+    String METRICS_HUDI_TABLE_FILE_COUNT_AVERAGE_PER_TABLE = METRICS_HUDI_TABLE_FILE_COUNT + "_average_per_table";
+    String METRICS_HUDI_TABLE_TIMELINE_FILE_COUNT = METRICS_HUDI_TABLE + "_timeline_file_count";
+    String METRICS_HUDI_TABLE_TIMELINE_FILE_COUNT_AVERAGE_PER_TABLE = METRICS_HUDI_TABLE_TIMELINE_FILE_COUNT + "_average_per_table";
+
     String METRICS_LABEL_FLINK_JOB_ID = "flink_job_id";
     String METRICS_LABEL_FLINK_JOB_NAME = "flink_job_name";
     String METRICS_LABEL_FLINK_NATIVE_JOB_ID = "flink_native_job_id";

service-configuration/src/main/java/com/lanyuanxiaoyao/service/configuration/entity/monitor/MetricsProgress.java

@@ -0,0 +1,55 @@
+package com.lanyuanxiaoyao.service.configuration.entity.monitor;
+
+/**
+ * 指标运行进度
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-10-14
+ */
+public class MetricsProgress {
+    private String name;
+    private Boolean running;
+    private Double progress;
+
+    public MetricsProgress() {
+    }
+
+    public MetricsProgress(String name, Boolean running, Double progress) {
+        this.name = name;
+        this.running = running;
+        this.progress = progress;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Double getProgress() {
+        return progress;
+    }
+
+    public void setProgress(Double progress) {
+        this.progress = progress;
+    }
+
+    public Boolean getRunning() {
+        return running;
+    }
+
+    public void setRunning(Boolean running) {
+        this.running = running;
+    }
+
+    @Override
+    public String toString() {
+        return "MetricsProgress{" +
+               "name='" + name + '\'' +
+               ", running=" + running +
+               ", progress=" + progress +
+               '}';
+    }
+}

service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/HudiService.java

@@ -106,5 +106,14 @@ public interface HudiService {
     InputStream download(@Query("root") String root);
 
     @Get("/hdfs/size")
-    String size(@Query("root") String root);
+    Long size(@Query("root") String root);
+
+    @Get("/hdfs/count")
+    Long count(@Query("root") String root);
+
+    @Get("/hdfs/file_count")
+    Long fileCount(@Query("root") String root);
+
+    @Get("/hdfs/directory_count")
+    Long directoryCount(@Query("root") String root);
 }

service-forest/src/main/java/com/lanyuanxiaoyao/service/forest/service/MonitorService.java

@@ -0,0 +1,18 @@
+package com.lanyuanxiaoyao.service.forest.service;
+
+import com.dtflys.forest.annotation.BaseRequest;
+import com.dtflys.forest.annotation.Get;
+import com.lanyuanxiaoyao.service.configuration.entity.monitor.MetricsProgress;
+import org.eclipse.collections.api.list.ImmutableList;
+
+/**
+ * 监控指标查询
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-10-14
+ */
+@BaseRequest(baseURL = "http://service-monitor")
+public interface MonitorService {
+    @Get("/metrics_control/progress")
+    ImmutableList<MetricsProgress> progress();
+}

service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/controller/HdfsController.java

@@ -82,4 +82,19 @@ public class HdfsController {
     public Long size(@RequestParam("root") String root) throws IOException {
         return hdfsService.size(root);
     }
+
+    @GetMapping("count")
+    public Long count(@RequestParam("root") String root) throws IOException {
+        return hdfsService.count(root);
+    }
+
+    @GetMapping("file_count")
+    public Long fileCount(@RequestParam("root") String root) throws IOException {
+        return hdfsService.countFiles(root);
+    }
+
+    @GetMapping("directory_count")
+    public Long DirectoryCount(@RequestParam("root") String root) throws IOException {
+        return hdfsService.countDirectories(root);
+    }
 }

service-hudi-query/src/main/java/com/lanyuanxiaoyao/service/hudi/service/HdfsService.java

@@ -8,6 +8,7 @@ import com.lanyuanxiaoyao.service.forest.service.InfoService;
 import java.io.IOException;
 import java.io.OutputStream;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
@@ -160,4 +161,26 @@ public class HdfsService {
             return fileSystem.getContentSummary(new Path(root)).getLength();
         }
     }
+
+    @Cacheable(value = "count-hpath", sync = true)
+    public Long count(String root) throws IOException {
+        try (FileSystem fileSystem = FileSystem.get(new Configuration())) {
+            ContentSummary summary = fileSystem.getContentSummary(new Path(root));
+            return summary.getFileCount() + summary.getDirectoryCount();
+        }
+    }
+
+    @Cacheable(value = "file-count-hpath", sync = true)
+    public Long countFiles(String root) throws IOException {
+        try (FileSystem fileSystem = FileSystem.get(new Configuration())) {
+            return fileSystem.getContentSummary(new Path(root)).getFileCount();
+        }
+    }
+
+    @Cacheable(value = "directory-count-hpath", sync = true)
+    public Long countDirectories(String root) throws IOException {
+        try (FileSystem fileSystem = FileSystem.get(new Configuration())) {
+            return fileSystem.getContentSummary(new Path(root)).getDirectoryCount();
+        }
+    }
 }

service-monitor/src/main/java/com/lanyuanxiaoyao/service/monitor/controller/MetricsController.java

@@ -0,0 +1,40 @@
+package com.lanyuanxiaoyao.service.monitor.controller;
+
+import com.lanyuanxiaoyao.service.configuration.entity.monitor.MetricsProgress;
+import com.lanyuanxiaoyao.service.monitor.metric.Metrics;
+import java.util.Map;
+import org.eclipse.collections.api.factory.Lists;
+import org.eclipse.collections.api.list.ImmutableList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * 操作进度
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-10-14
+ */
+@RestController
+@RequestMapping("metrics_control")
+public class MetricsController {
+    private static final Logger logger = LoggerFactory.getLogger(MetricsController.class);
+
+    private final ApplicationContext context;
+
+    public MetricsController(ApplicationContext context) {
+        this.context = context;
+    }
+
+    @GetMapping("progress")
+    public ImmutableList<MetricsProgress> progress() {
+        Map<String, Metrics> metricsMap = context.getBeansOfType(Metrics.class);
+        return Lists.immutable.ofAll(metricsMap.entrySet())
+                .toImmutableSortedList(Map.Entry.comparingByKey())
+                .collect(Map.Entry::getValue)
+                .collect(metrics -> new MetricsProgress(metrics.name(), metrics.running(), metrics.progress()));
+    }
+}

service-monitor/src/main/java/com/lanyuanxiaoyao/service/monitor/metric/HudiTableFilesCountMetrics.java

@@ -0,0 +1,111 @@
+package com.lanyuanxiaoyao.service.monitor.metric;
+
+import cn.hutool.core.util.StrUtil;
+import com.lanyuanxiaoyao.service.common.Constants;
+import com.lanyuanxiaoyao.service.common.entity.TableMeta;
+import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
+import com.lanyuanxiaoyao.service.forest.service.HudiService;
+import com.lanyuanxiaoyao.service.forest.service.InfoService;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import java.util.concurrent.atomic.AtomicLong;
+import org.eclipse.collections.api.factory.Lists;
+import org.eclipse.collections.api.factory.Maps;
+import org.eclipse.collections.api.list.ImmutableList;
+import org.eclipse.collections.api.map.MutableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import static com.lanyuanxiaoyao.service.common.Constants.MINUTE;
+
+/**
+ * Hudi表相关指标
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-03-05
+ */
+@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
+@Service
+public class HudiTableFilesCountMetrics extends Metrics {
+    private static final Logger logger = LoggerFactory.getLogger(HudiTableFilesCountMetrics.class);
+
+    private final MeterRegistry registry;
+    private final InfoService infoService;
+    private final HudiService hudiService;
+
+    private final MutableMap<String, AtomicLong> fileCountCacheMap;
+    private final MutableMap<String, AtomicLong> timelineFileCountCacheMap;
+
+    public HudiTableFilesCountMetrics(MeterRegistry registry, InfoService infoService, HudiService hudiService) {
+        this.registry = registry;
+        this.infoService = infoService;
+        this.hudiService = hudiService;
+
+        fileCountCacheMap = Maps.mutable.empty();
+        timelineFileCountCacheMap = Maps.mutable.empty();
+    }
+
+    @Override
+    public String name() {
+        return "Hudi表文件数量监控";
+    }
+
+    @Scheduled(fixedDelay = 30 * MINUTE, initialDelay = MINUTE)
+    @Override
+    public void update() {
+        try {
+            start();
+            ImmutableList<TableMeta> metas = infoService.tableMetaList();
+            setTotal(metas.size());
+            metas
+                    .asParallel(ExecutorProvider.EXECUTORS_2, 1)
+                    .reject(meta -> StrUtil.isBlank(meta.getPulsarAddress()))
+                    .forEach(meta -> {
+                        try {
+                            AtomicLong filecountCache = fileCountCacheMap.getIfAbsentPut(
+                                    meta.getAlias(),
+                                    registry.gauge(
+                                            Constants.METRICS_HUDI_TABLE_FILE_COUNT,
+                                            Lists.immutable.of(
+                                                    Tag.of(Constants.METRICS_LABEL_FLINK_JOB_ID, meta.getJob().getId().toString()),
+                                                    Tag.of(Constants.METRICS_LABEL_ALIAS, meta.getAlias()),
+                                                    Tag.of(Constants.METRICS_LABEL_SCHEMA, meta.getSchema()),
+                                                    Tag.of(Constants.METRICS_LABEL_TABLE, meta.getTable())
+                                            ),
+                                            new AtomicLong(0)
+                                    )
+                            );
+                            AtomicLong timelineFileCountCache = timelineFileCountCacheMap.getIfAbsentPut(
+                                    meta.getAlias(),
+                                    registry.gauge(
+                                            Constants.METRICS_HUDI_TABLE_TIMELINE_FILE_COUNT,
+                                            Lists.immutable.of(
+                                                    Tag.of(Constants.METRICS_LABEL_FLINK_JOB_ID, meta.getJob().getId().toString()),
+                                                    Tag.of(Constants.METRICS_LABEL_ALIAS, meta.getAlias()),
+                                                    Tag.of(Constants.METRICS_LABEL_SCHEMA, meta.getSchema()),
+                                                    Tag.of(Constants.METRICS_LABEL_TABLE, meta.getTable())
+                                            ),
+                                            new AtomicLong(0)
+                                    )
+                            );
+
+                            String hdfs = meta.getHudi().getTargetHdfsPath();
+                            if (hudiService.existsHudiTable(hdfs)) {
+                                Long count = hudiService.fileCount(hdfs);
+                                filecountCache.set(count);
+
+                                String timelineHdfs = hdfs + "/.hoodie";
+                                timelineFileCountCache.set(hudiService.fileCount(timelineHdfs));
+                            }
+                        } catch (Exception exception) {
+                            logger.warn("Get file count fail for {}", meta.getAlias(), exception);
+                        }
+                        finished();
+                    });
+        } finally {
+            reset();
+        }
+    }
+}

service-monitor/src/main/java/com/lanyuanxiaoyao/service/monitor/metric/Metrics.java

@@ -1,9 +1,56 @@
 package com.lanyuanxiaoyao.service.monitor.metric;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
 /**
  * @author lanyuanxiaoyao
  * @date 2024-03-05
  */
 public abstract class Metrics {
-    abstract void update();
+    private final AtomicBoolean running = new AtomicBoolean(false);
+    private final AtomicLong finished = new AtomicLong(0);
+    private final AtomicLong total = new AtomicLong(0);
+
+    public abstract String name();
+
+    public abstract void update();
+
+    public boolean running() {
+        return running.get();
+    }
+
+    public double progress() {
+        if (total.get() == 0) {
+            return 0;
+        } else {
+            return finished.get() * 1.0 / total.get();
+        }
+    }
+
+    protected void start() {
+        running.set(true);
+    }
+
+    protected void stop() {
+        running.set(false);
+    }
+
+    protected void setTotal(Long total) {
+        this.total.set(total);
+    }
+
+    protected void setTotal(Integer total) {
+        this.total.set(total);
+    }
+
+    protected void finished() {
+        finished.incrementAndGet();
+    }
+
+    protected void reset() {
+        stop();
+        setTotal(0);
+        finished.set(0);
+    }
 }

service-monitor/src/main/java/com/lanyuanxiaoyao/service/monitor/metric/PulsarBacklogMetrics.java

@@ -0,0 +1,97 @@
+package com.lanyuanxiaoyao.service.monitor.metric;
+
+import cn.hutool.core.util.StrUtil;
+import com.lanyuanxiaoyao.service.common.Constants;
+import com.lanyuanxiaoyao.service.common.entity.TableMeta;
+import com.lanyuanxiaoyao.service.common.utils.NameHelper;
+import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
+import com.lanyuanxiaoyao.service.configuration.HudiServiceProperties;
+import com.lanyuanxiaoyao.service.forest.service.InfoService;
+import com.lanyuanxiaoyao.service.forest.service.PulsarService;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import java.util.concurrent.atomic.AtomicLong;
+import org.eclipse.collections.api.factory.Lists;
+import org.eclipse.collections.api.factory.Maps;
+import org.eclipse.collections.api.list.ImmutableList;
+import org.eclipse.collections.api.map.MutableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import static com.lanyuanxiaoyao.service.common.Constants.MINUTE;
+
+/**
+ * Pulsar
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-03-05
+ */
+@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
+@Service
+public class PulsarBacklogMetrics extends Metrics {
+    private static final Logger logger = LoggerFactory.getLogger(PulsarBacklogMetrics.class);
+
+    private final MeterRegistry registry;
+    private final InfoService infoService;
+    private final PulsarService pulsarService;
+    private final HudiServiceProperties hudiServiceProperties;
+
+    private final MutableMap<String, AtomicLong> backlogMap;
+
+    public PulsarBacklogMetrics(MeterRegistry registry, InfoService infoService, PulsarService pulsarService, HudiServiceProperties hudiServiceProperties) {
+        this.registry = registry;
+        this.infoService = infoService;
+        this.pulsarService = pulsarService;
+        this.hudiServiceProperties = hudiServiceProperties;
+
+        backlogMap = Maps.mutable.empty();
+    }
+
+    @Override
+    public String name() {
+        return "Pulsar backlog监控";
+    }
+
+    @Scheduled(fixedDelay = 30 * MINUTE, initialDelay = MINUTE)
+    @Override
+    public void update() {
+        try {
+            start();
+            ImmutableList<TableMeta> metas = infoService.tableMetaList();
+            setTotal(metas.size());
+            metas
+                    .asParallel(ExecutorProvider.EXECUTORS_2, 1)
+                    .reject(meta -> StrUtil.isBlank(meta.getPulsarAddress()))
+                    .forEach(meta -> {
+                        try {
+                            AtomicLong backlogCache = backlogMap.getIfAbsentPut(
+                                    meta.getAlias(),
+                                    registry.gauge(
+                                            Constants.METRICS_PULSAR_BACKLOG,
+                                            Lists.immutable.of(
+                                                    Tag.of(Constants.METRICS_LABEL_FLINK_JOB_ID, meta.getJob().getId().toString()),
+                                                    Tag.of(Constants.METRICS_LABEL_ALIAS, meta.getAlias()),
+                                                    Tag.of(Constants.METRICS_LABEL_SCHEMA, meta.getSchema()),
+                                                    Tag.of(Constants.METRICS_LABEL_TABLE, meta.getTable())
+                                            ),
+                                            new AtomicLong(0)
+                                    )
+                            );
+                            String name = pulsarService.name(meta.getPulsarAddress());
+                            if (StrUtil.isNotBlank(name)) {
+                                Long backlog = pulsarService.backlog(name, meta.getTopic(), NameHelper.pulsarSubscriptionName(meta.getJob().getId(), meta.getAlias(), hudiServiceProperties.getSignature()));
+                                backlogCache.set(backlog);
+                                infoService.savePulsarBacklog(meta.getJob().getId(), meta.getAlias(), backlog);
+                            }
+                        } catch (Exception exception) {
+                            logger.warn("Update pulsar backlog fail for {}", meta.getAlias(), exception);
+                        }
+                        finished();
+                    });
+        } finally {
+            reset();
+        }
+    }
+}

service-monitor/src/main/java/com/lanyuanxiaoyao/service/monitor/metric/PulsarMetrics.java

@@ -1,82 +0,0 @@
-package com.lanyuanxiaoyao.service.monitor.metric;
-
-import cn.hutool.core.util.StrUtil;
-import com.lanyuanxiaoyao.service.common.Constants;
-import com.lanyuanxiaoyao.service.common.utils.NameHelper;
-import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
-import com.lanyuanxiaoyao.service.configuration.HudiServiceProperties;
-import com.lanyuanxiaoyao.service.forest.service.InfoService;
-import com.lanyuanxiaoyao.service.forest.service.PulsarService;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Tag;
-import java.util.concurrent.atomic.AtomicLong;
-import org.eclipse.collections.api.factory.Lists;
-import org.eclipse.collections.api.factory.Maps;
-import org.eclipse.collections.api.map.MutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Service;
-
-import static com.lanyuanxiaoyao.service.common.Constants.MINUTE;
-
-/**
- * Pulsar
- *
- * @author lanyuanxiaoyao
- * @date 2024-03-05
- */
-@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
-@Service
-public class PulsarMetrics extends Metrics {
-    private static final Logger logger = LoggerFactory.getLogger(PulsarMetrics.class);
-
-    private final MeterRegistry registry;
-    private final InfoService infoService;
-    private final PulsarService pulsarService;
-    private final HudiServiceProperties hudiServiceProperties;
-
-    private final MutableMap<String, AtomicLong> backlogMap;
-
-    public PulsarMetrics(MeterRegistry registry, InfoService infoService, PulsarService pulsarService, HudiServiceProperties hudiServiceProperties) {
-        this.registry = registry;
-        this.infoService = infoService;
-        this.pulsarService = pulsarService;
-        this.hudiServiceProperties = hudiServiceProperties;
-
-        backlogMap = Maps.mutable.empty();
-    }
-
-    @Scheduled(fixedDelay = 30 * MINUTE, initialDelay = MINUTE)
-    @Override
-    void update() {
-        infoService.tableMetaList()
-                .asParallel(ExecutorProvider.EXECUTORS_2, 1)
-                .reject(meta -> StrUtil.isBlank(meta.getPulsarAddress()))
-                .forEach(meta -> {
-                    try {
-                        AtomicLong backlogCache = backlogMap.getIfAbsentPut(
-                                meta.getAlias(),
-                                registry.gauge(
-                                        Constants.METRICS_PULSAR_BACKLOG,
-                                        Lists.immutable.of(
-                                                Tag.of(Constants.METRICS_LABEL_FLINK_JOB_ID, meta.getJob().getId().toString()),
-                                                Tag.of(Constants.METRICS_LABEL_ALIAS, meta.getAlias()),
-                                                Tag.of(Constants.METRICS_LABEL_SCHEMA, meta.getSchema()),
-                                                Tag.of(Constants.METRICS_LABEL_TABLE, meta.getTable())
-                                        ),
-                                        new AtomicLong(0)
-                                )
-                        );
-                        String name = pulsarService.name(meta.getPulsarAddress());
-                        if (StrUtil.isNotBlank(name)) {
-                            Long backlog = pulsarService.backlog(name, meta.getTopic(), NameHelper.pulsarSubscriptionName(meta.getJob().getId(), meta.getAlias(), hudiServiceProperties.getSignature()));
-                            backlogCache.set(backlog);
-                            infoService.savePulsarBacklog(meta.getJob().getId(), meta.getAlias(), backlog);
-                        }
-                    } catch (Exception exception) {
-                        logger.warn("Update pulsar backlog fail for " + meta.getAlias(), exception);
-                    }
-                });
-    }
-}

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/configuration/ScheduleStrategyProvider.java

@@ -26,7 +26,11 @@ public class ScheduleStrategyProvider {
         return Lists.immutable.of(
                 ScheduleStrategyImpl.simple(false, "distribute_schedule", "定时分布式调度", DistributeScheduleJob.class, "0/2 * * * * ?"),
                 // 普通调度
-                ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,10,13,16,19 * * ?"),
+                // ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,10,13,16,19 * * ?"),
+                // 普通调度（20240925不调度11点、14点）
+                ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,16,19 * * ?"),
+                // 普通调度（20240925不调度8点、11点、14点）
+                // ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,16,19 * * ?"),
                 // 重点表调度
                 ScheduleStrategyImpl.simple("focus_evening_schedule", "晚间重点表调度", FocusScheduleJob.class, "0 50 20,21,22 * * ?"),
                 // ODS重点表调度

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/cluster/A4Cluster.java

@@ -3,6 +3,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.cluster;
 import com.lanyuanxiaoyao.service.common.Constants;
 import com.lanyuanxiaoyao.service.forest.service.YarnService;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.AvailableStrategy;
+import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.DatetimeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.QueueSizeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.YarnQueueUsedLimit;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
@@ -23,7 +24,8 @@ public class A4Cluster extends Cluster {
                 Constants.COMPACTION_QUEUE_A4,
                 AvailableStrategy.and(
                         new QueueSizeLimit(client, Constants.COMPACTION_QUEUE_A4, 10),
-                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_A4, "ten_iap.datalake", 0.8)
+                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_A4, "ten_iap.datalake", 0.8),
+                        new DatetimeLimit(false, "* * 7-22 * * ?")
                 )
         );
     }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/cluster/B1Cluster.java

@@ -3,6 +3,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.cluster;
 import com.lanyuanxiaoyao.service.common.Constants;
 import com.lanyuanxiaoyao.service.forest.service.YarnService;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.AvailableStrategy;
+import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.DatetimeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.QueueSizeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.YarnQueueUsedLimit;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
@@ -23,7 +24,8 @@ public class B1Cluster extends Cluster {
                 Constants.COMPACTION_QUEUE_B1,
                 AvailableStrategy.and(
                         new QueueSizeLimit(client, Constants.COMPACTION_QUEUE_B1, 20),
-                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_B1, "datalake", 1.0)
+                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_B1, "datalake", 1.0),
+                        new DatetimeLimit(false, "* * 7-22 * * ?")
                 )
         );
     }

service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/OverviewController.java

@@ -6,10 +6,16 @@ import com.lanyuanxiaoyao.service.common.entity.FlinkJob;
 import com.lanyuanxiaoyao.service.common.utils.NameHelper;
 import com.lanyuanxiaoyao.service.configuration.ExecutorProvider;
 import com.lanyuanxiaoyao.service.configuration.entity.info.JobIdAndAlias;
+import com.lanyuanxiaoyao.service.configuration.entity.monitor.MetricsProgress;
 import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnApplication;
 import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnRootQueue;
 import com.lanyuanxiaoyao.service.configuration.entity.zookeeper.ZookeeperNode;
-import com.lanyuanxiaoyao.service.forest.service.*;
+import com.lanyuanxiaoyao.service.forest.service.InfoService;
+import com.lanyuanxiaoyao.service.forest.service.MonitorService;
+import com.lanyuanxiaoyao.service.forest.service.QueueService;
+import com.lanyuanxiaoyao.service.forest.service.ScheduleService;
+import com.lanyuanxiaoyao.service.forest.service.YarnService;
+import com.lanyuanxiaoyao.service.forest.service.ZookeeperService;
 import com.lanyuanxiaoyao.service.web.controller.base.AmisCrudResponse;
 import com.lanyuanxiaoyao.service.web.controller.base.AmisMapResponse;
 import com.lanyuanxiaoyao.service.web.controller.base.AmisResponse;
@@ -48,14 +54,16 @@ public class OverviewController extends BaseController {
     private final QueueService queueService;
     private final ScheduleService scheduleService;
     private final ZookeeperService zookeeperService;
+    private final MonitorService monitorService;
 
     @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
-    public OverviewController(InfoService infoService, YarnService yarnService, QueueService queueService, ScheduleService scheduleService, ZookeeperService zookeeperService) {
+    public OverviewController(InfoService infoService, YarnService yarnService, QueueService queueService, ScheduleService scheduleService, ZookeeperService zookeeperService, MonitorService monitorService) {
         this.infoService = infoService;
         this.yarnService = yarnService;
         this.queueService = queueService;
         this.scheduleService = scheduleService;
         this.zookeeperService = zookeeperService;
+        this.monitorService = monitorService;
     }
 
     @GetMapping("")
@@ -207,4 +215,12 @@ public class OverviewController extends BaseController {
                 .setData("unRunningTable", unRunningTable.size())
                 .setData("unRunningTableList", unRunningTable);
     }
+
+    @GetMapping("monitor_progress")
+    public AmisMapResponse monitorProgress() {
+        return AmisCrudResponse.responseCrudData(
+                monitorService.progress()
+                        .collect(p -> new MetricsProgress(p.getName(), p.getRunning(), p.getProgress()))
+        );
+    }
 }

service-web/src/main/resources/static/common/info.js

@@ -1,6 +1,6 @@
 const commonInfo = {
-    // baseUrl: 'http://132.126.207.130:35690/hudi_services/service_web',
+    baseUrl: 'http://132.126.207.130:35690/hudi_services/service_web',
-    baseUrl: '/hudi_services/service_web',
+    // baseUrl: '/hudi_services/service_web',
     clusters: {
         // hudi同步运行集群和yarn队列名称
         sync: {

service-web/src/main/resources/static/components/overview-tab.js

@@ -408,26 +408,38 @@ function overviewTab() {
                     }
                 ]
             },
-            /*{type: 'divider'},
+            {type: 'divider'},
             {
-                type: 'service',
+                type: 'crud',
-                api: '${base}/overview/schedule_times',
+                title: '监控指标运行进度',
-                interval: 60000,
+                api: `\${base}/overview/monitor_progress`,
-                silentPolling: true,
+                ...crudCommonOptions(),
-                body: [
+                interval: 2000,
-                    '调度时间点',
+                loadDataOnce: true,
+                columns: [
                     {
-                        type: 'each',
+                        name: 'name',
-                        name: 'items',
+                        label: '名称',
-                        className: 'grid',
+                        width: 120,
-                        items: {
+                    },
-                            type: 'tag',
+                    {
-                            color: '${color}',
+                        name: 'running',
-                            label: '${DATETOSTR(TIMESTAMP(time, \'x\'), \'HH:mm:ss\')}'
+                        label: '状态',
+                        type: 'mapping',
+                        width: 50,
+                        map: {
+                            'true': '运行中',
+                            'false': '未运行',
                         }
-                    }
+                    },
+                    {
+                        label: '进度',
+                        type: 'progress',
+                        value: '${ROUND(progress * 100)}',
+                        map: 'bg-primary',
+                    },
                 ]
-            }*/
+            }
         ]
     }
 }

test/test.http

@@ -133,4 +133,8 @@ hoodie.table.checksum=989688289
 GET http://{{username}}:{{password}}@b12s10.hdp.dc:31719/task/law_enforcement?
     pulsar_url=pulsar://132.122.115.158:16650,132.122.115.159:16650,132.122.115.160:16650,132.122.115.161:16650,132.122.115.167:16650,132.122.115.168:16650&
     pulsar_topic=persistent://odcp/acct_sz/acct_item_755&start_time=1716858000000&end_time=1716861600000&
-    primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
+    primary_keys=ACCT_ITEM_ID&partition_keys=ACCT_ID
+
+### Count files
+GET http://{{username}}:{{password}}@b12s20.hdp.dc:19521/hdfs/count?
+    root=hdfs://b2/apps/datalake/hive/dws_wsyyt/external_table_hudi/dws_tb_jt_servuser_data