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

refactor(info-query): 优化跨天信息和基本信息查询

Browse Source
This commit is contained in:
兰缘小妖
2023-06-13 10:45:33 +08:00
parent c1a9e5b24b
commit 04a23431f7
6 changed files with 229 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/controller/InfoController.java
View File

@@ -37,26 +37,6 @@ public class InfoController {
this.infoService = infoService;
}
@GetMapping("table_total")
public Long tableTotal() {
return infoService.tableTotal();
}
@GetMapping("hudi_total")
public Long hudiTotal() {
return infoService.hudiTotal();
}
@GetMapping("focus_count")
public Long focusCount() {
return infoService.focusCount();
}
@GetMapping("normal_count")
public Long normalCount() {
return infoService.normalCount();
}
@GetMapping("/job_id_alias")
public PageResponse<JobIdAndAlias> jobIdAndAlias(
@RequestParam(value = "page", defaultValue = "1") Integer page,
@@ -145,14 +125,9 @@ public class InfoController {
return infoService.nonUpdatedVersionTables();
}
@GetMapping("/un_scheduled_normal_table_count")
public Long unScheduledNormalTableCount(@RequestParam("version") String version) {
return infoService.unScheduledNormalTableCount(version);
}
@GetMapping("/un_scheduled_focus_table_count")
public Long unScheduledFocusTableCount(@RequestParam("version") String version) {
return infoService.unScheduledFocusTableCount(version);
@GetMapping("/un_receive_version_normal_table")
public ImmutableList<JobIdAndAlias> unReceiveVersionNormalTable(@RequestParam("version") String version) {
return infoService.unReceiveVersionNormalTable(version);
}
@GetMapping("/un_receive_version_normal_table_count")
@@ -160,8 +135,53 @@ public class InfoController {
return infoService.unReceiveVersionNormalTableCount(version);
}
@GetMapping("/un_receive_version_focus_table")
public ImmutableList<JobIdAndAlias> unReceiveVersionFocusTable(@RequestParam("version") String version) {
return infoService.unReceiveVersionFocusTable(version);
}
@GetMapping("/un_receive_version_focus_table_count")
public Long unReceiveVersionFocusTableCount(@RequestParam("version") String version) {
return infoService.unReceiveVersionFocusTableCount(version);
}
@GetMapping("/un_scheduled_normal_table")
public ImmutableList<JobIdAndAlias> unScheduledNormalTable(@RequestParam("version") String version) {
return infoService.unScheduledNormalTable(version);
}
@GetMapping("/un_scheduled_normal_table_count")
public Long unScheduledNormalTableCount(@RequestParam("version") String version) {
return infoService.unScheduledNormalTableCount(version);
}
@GetMapping("/un_scheduled_focus_table")
public ImmutableList<JobIdAndAlias> unScheduledFocusTable(@RequestParam("version") String version) {
return infoService.unScheduledFocusTable(version);
}
@GetMapping("/un_scheduled_focus_table_count")
public Long unScheduledFocusTableCount(@RequestParam("version") String version) {
return infoService.unScheduledFocusTableCount(version);
}
@GetMapping("/table_count")
public Long tableCount() {
return infoService.tableCount();
}
@GetMapping("/table_focus_count")
public Long tableFocusCount() {
return infoService.tableFocusCount();
}
@GetMapping("/hudi_count")
public Long hudiCount() {
return infoService.hudiCount();
}
@GetMapping("/hudi_focus_count")
public Long hudiFocusCount() {
return infoService.hudiFocusCount();
}
}

198
service-info-query/src/main/java/com/lanyuanxiaoyao/service/info/service/InfoService.java
View File

@@ -343,119 +343,189 @@ public class InfoService {
});
}
private SqlBuilder generateUnReceiveVersionNormalTableSql(SelectSqlBuilder builder, String version) {
return builder
.from(TABLE_INFO)
.whereLt(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.andNotIn(
StrUtil.format("concat({}, {})", TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS),
SqlBuilder.select(StrUtil.format("concat({}, {})", TABLE_VERSION_FLINK_JOB_ID, TABLE_VERSION_ALIAS))
.from(TABLE_VERSION)
.whereEq(TABLE_VERSION_VERSION, version)
);
}
@Cacheable(value = "un-receive-version-normal-table", sync = true)
@Retryable(Throwable.class)
public ImmutableList<JobIdAndAlias> unReceiveVersionNormalTable(String version) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateUnReceiveVersionNormalTableSql(SqlBuilder.select(TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS), version)
.build(),
(rs, row) -> new JobIdAndAlias(rs.getLong(1), rs.getString(2))
)
);
}
@Cacheable(value = "un-receive-version-normal-table-count", sync = true)
@Retryable(Throwable.class)
public Long unReceiveVersionNormalTableCount(String version) {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select(COUNT)
.from(TABLE_INFO)
.whereLt(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.andNotIn(
StrUtil.format("concat({}, {})", TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS),
SqlBuilder.select(StrUtil.format("concat({}, {})", TABLE_VERSION_FLINK_JOB_ID, TABLE_VERSION_ALIAS))
.from(TABLE_VERSION)
.whereEq(TABLE_VERSION_VERSION, version)
)
generateUnReceiveVersionNormalTableSql(SqlBuilder.select(COUNT), version)
.build(),
Long.class
);
}
private SqlBuilder generateUnReceiveVersionFocusTable(SelectSqlBuilder builder, String version) {
return builder
.from(TABLE_INFO)
.whereGe(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.andNotIn(
StrUtil.format("concat({}, {})", TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS),
SqlBuilder.select(StrUtil.format("concat({}, {})", TABLE_VERSION_FLINK_JOB_ID, TABLE_VERSION_ALIAS))
.from(TABLE_VERSION)
.whereEq(TABLE_VERSION_VERSION, version)
);
}
@Cacheable(value = "un-receive-version-focus-table", sync = true)
@Retryable(Throwable.class)
public ImmutableList<JobIdAndAlias> unReceiveVersionFocusTable(String version) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateUnReceiveVersionFocusTable(SqlBuilder.select(TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS), version)
.build(),
(rs, row) -> new JobIdAndAlias(rs.getLong(1), rs.getString(2))
)
);
}
@Cacheable(value = "un-receive-version-focus-table-count", sync = true)
@Retryable(Throwable.class)
public Long unReceiveVersionFocusTableCount(String version) {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select(COUNT)
.from(TABLE_INFO)
.whereGe(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.andNotIn(
StrUtil.format("concat({}, {})", TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS),
SqlBuilder.select(StrUtil.format("concat({}, {})", TABLE_VERSION_FLINK_JOB_ID, TABLE_VERSION_ALIAS))
.from(TABLE_VERSION)
.whereEq(TABLE_VERSION_VERSION, version)
)
generateUnReceiveVersionFocusTable(SqlBuilder.select(COUNT), version)
.build(),
Long.class
);
}
private SqlBuilder generateUnScheduledNormalTableSql(SelectSqlBuilder builder, String version) {
return builder
.from(TABLE_INFO)
.join(TABLE_VERSION)
.onEq(TABLE_INFO_FLINK_JOB_ID, Column.as(TABLE_VERSION_FLINK_JOB_ID))
.andEq(TABLE_INFO_ALIAS, Column.as(TABLE_VERSION_ALIAS))
.whereLt(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_VERSION_SCHEDULED, false)
.andEq(TABLE_VERSION_VERSION, version)
.andEq(TABLE_INFO_STATUS, "y");
}
@Cacheable(value = "un-scheduled-normal-table", sync = true)
@Retryable(Throwable.class)
public ImmutableList<JobIdAndAlias> unScheduledNormalTable(String version) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateUnScheduledNormalTableSql(SqlBuilder.select(TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS), version)
.build(),
(rs, row) -> new JobIdAndAlias(rs.getLong(1), rs.getString(2))
)
);
}
@Cacheable(value = "un_scheduled_normal_table_count", sync = true)
@Retryable(Throwable.class)
public Long unScheduledNormalTableCount(String version) {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select(COUNT)
.from(TABLE_INFO)
.join(TABLE_VERSION)
.onEq(TABLE_INFO_FLINK_JOB_ID, Column.as(TABLE_VERSION_FLINK_JOB_ID))
.andEq(TABLE_INFO_ALIAS, Column.as(TABLE_VERSION_ALIAS))
.whereLt(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_VERSION_SCHEDULED, false)
.andEq(TABLE_VERSION_VERSION, version)
.andEq(TABLE_INFO_STATUS, "y")
generateUnScheduledNormalTableSql(SqlBuilder.select(COUNT), version)
.build(),
Long.class
);
}
private SqlBuilder generateUnScheduledFocusTableSql(SelectSqlBuilder builder, String version) {
return builder
.from(TABLE_INFO)
.join(TABLE_VERSION)
.onEq(TABLE_INFO_FLINK_JOB_ID, Column.as(TABLE_VERSION_FLINK_JOB_ID))
.andEq(TABLE_INFO_ALIAS, Column.as(TABLE_VERSION_ALIAS))
.whereGe(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_VERSION_SCHEDULED, false)
.andEq(TABLE_VERSION_VERSION, version)
.andEq(TABLE_INFO_STATUS, "y");
}
@Cacheable(value = "un-scheduled-focus-table", sync = true)
@Retryable(Throwable.class)
public ImmutableList<JobIdAndAlias> unScheduledFocusTable(String version) {
return Lists.immutable.ofAll(
mysqlJdbcTemplate.query(
generateUnScheduledFocusTableSql(SqlBuilder.select(TABLE_INFO_FLINK_JOB_ID, TABLE_INFO_ALIAS), version)
.build(),
(rs, row) -> new JobIdAndAlias(rs.getLong(1), rs.getString(2))
)
);
}
@Cacheable(value = "un_scheduled_focus_table_count", sync = true)
@Retryable(Throwable.class)
public Long unScheduledFocusTableCount(String version) {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select(COUNT)
.from(TABLE_INFO)
.join(TABLE_VERSION)
.onEq(TABLE_INFO_FLINK_JOB_ID, Column.as(TABLE_VERSION_FLINK_JOB_ID))
.andEq(TABLE_INFO_ALIAS, Column.as(TABLE_VERSION_ALIAS))
.whereGe(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_VERSION_SCHEDULED, false)
.andEq(TABLE_VERSION_VERSION, version)
.andEq(TABLE_INFO_STATUS, "y")
generateUnScheduledFocusTableSql(SqlBuilder.select(COUNT), version)
.build(),
Long.class
);
}
@Cacheable(value = "table-total", sync = true)
@Cacheable(value = "table-count", sync = true)
@Retryable(Throwable.class)
public Long tableTotal() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct concat(src_schema, src_table))").from(TABLE_INFO).whereEq(TABLE_INFO_STATUS, "y").build(),
Long.class
);
}
@Cacheable(value = "hudi-total", sync = true)
@Retryable(Throwable.class)
public Long hudiTotal() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct tgt_hdfs_path) as count").from(TABLE_INFO).whereEq(TABLE_INFO_STATUS, "y").build(),
Long.class
);
}
@Cacheable(value = "focus-count", sync = true)
@Retryable(Throwable.class)
public Long focusCount() {
public Long tableCount() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct concat(src_schema, src_table))")
.from(TABLE_INFO)
.whereGe(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.whereEq(TABLE_INFO_STATUS, "y")
.build(),
Long.class
);
}
@Cacheable(value = "normal-count", sync = true)
@Cacheable(value = "table-focus-count", sync = true)
@Retryable(Throwable.class)
public Long normalCount() {
public Long tableFocusCount() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct concat(src_schema, src_table))")
.from(TABLE_INFO)
.whereLt(TABLE_INFO_PRIORITY, 10000)
.andEq(TABLE_INFO_STATUS, "y")
.whereEq(TABLE_INFO_STATUS, "y")
.andGe(TABLE_INFO_PRIORITY, 10000)
.build(),
Long.class
);
}
@Cacheable(value = "hudi-count", sync = true)
@Retryable(Throwable.class)
public Long hudiCount() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct tgt_hdfs_path) as count")
.from(TABLE_INFO)
.whereEq(TABLE_INFO_STATUS, "y")
.build(),
Long.class
);
}
@Cacheable(value = "hudi-focus-count", sync = true)
@Retryable(Throwable.class)
public Long hudiFocusCount() {
return mysqlJdbcTemplate.queryForObject(
SqlBuilder.select("count(distinct tgt_hdfs_path) as count")
.from(TABLE_INFO)
.whereEq(TABLE_INFO_STATUS, "y")
.andGe(TABLE_INFO_PRIORITY, 10000)
.build(),
Long.class
);