Skip to content
项目
群组
代码片段
帮助
当前项目
正在载入...
登录 / 注册
切换导航面板
H
huazheng-project-flink
概览
概览
详情
活动
周期分析
版本库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
统计图
问题
0
议题
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
CI / CD
CI / CD
流水线
作业
日程表
图表
维基
Wiki
代码片段
代码片段
成员
成员
折叠边栏
关闭边栏
活动
图像
聊天
创建新问题
作业
提交
问题看板
Open sidebar
huazheng
huazheng-project-flink
Commits
048d9e65
提交
048d9e65
authored
2月 24, 2021
作者:
think
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
likp任务
上级
5703a8cb
显示空白字符变更
内嵌
并排
正在显示
4 个修改的文件
包含
85 行增加
和
1 行删除
+85
-1
GPMapper.java
.../java/com/huazheng/project/greenplum/mapper/GPMapper.java
+1
-0
DeleteUpdateJobServiceImpl.java
...ct/greenplum/service/impl/DeleteUpdateJobServiceImpl.java
+69
-0
GPMapper_greenplum.xml
src/main/resources/mapper/greenplum/GPMapper_greenplum.xml
+5
-1
spring-init.xml
src/main/resources/spring-init.xml
+10
-0
没有找到文件。
src/main/java/com/huazheng/project/greenplum/mapper/GPMapper.java
浏览文件 @
048d9e65
...
...
@@ -233,6 +233,7 @@ public interface GPMapper {
@CacheEvict
(
key
=
"'selectLikp'+':'+#p0.vbeln+','+#p0.mandt"
)
public
void
updateLikp
(
Likp
element
);
public
List
<
Likp
>
selectLikpCheck
(
Likp
build
);
public
Long
selectLikpMaxRowNum
();
@Cacheable
(
key
=
"#root.method.name+':'+#p0.mandt+','+#p0.bukrs+','+#p0.belnr+','+#p0.gjahr"
,
unless
=
"#result == null"
)
public
Bkpf
selectBkpf
(
Bkpf
bkpf
);
// 查询替代删除
...
...
src/main/java/com/huazheng/project/greenplum/service/impl/DeleteUpdateJobServiceImpl.java
浏览文件 @
048d9e65
...
...
@@ -20,6 +20,7 @@ import com.huazheng.project.hana.model.Afvc;
import
com.huazheng.project.hana.model.Aufm
;
import
com.huazheng.project.hana.model.Knkk
;
import
com.huazheng.project.hana.model.Knvp
;
import
com.huazheng.project.hana.model.Likp
;
import
com.huazheng.project.hana.model.Mkpf
;
import
com.huazheng.project.hana.model.Mseg
;
import
com.huazheng.project.hana.model.Mska
;
...
...
@@ -1723,4 +1724,72 @@ public class DeleteUpdateJobServiceImpl {
}
}
public
void
selectLikpCheck
()
{
try
{
ValueOperations
<
String
,
String
>
opsForValue
=
redis1Template
.
opsForValue
();
opsForValue
.
setIfAbsent
(
"huazheng:check:Likp:rowNum"
,
"0"
);
String
rowNum
=
opsForValue
.
get
(
"huazheng:check:Likp:rowNum"
);
// 一次性任务的终止位
if
(
redis1Template
.
hasKey
(
"huazheng:check:Likp:stopNum"
)
==
false
)
{
// 查询最大计数器
String
maxRowNum
=
gpMapper
.
selectLikpMaxRowNum
().
toString
();
opsForValue
.
setIfAbsent
(
"huazheng:check:Likp:stopNum"
,
maxRowNum
);
}
Long
stopNum
=
Long
.
parseLong
(
opsForValue
.
get
(
"huazheng:check:Likp:stopNum"
));
if
(
stopNum
<
Long
.
parseLong
(
rowNum
))
{
JobKey
jobKey
=
JobKey
.
jobKey
(
"selectLikpCheck_once"
,
"DEFAULT"
);
// 删除任务
quartzScheduler
.
deleteJob
(
jobKey
);
return
;
}
Likp
build
=
Likp
.
builder
().
rowNum
(
rowNum
).
build
();
List
<
Likp
>
list
=
gpMapper
.
selectLikpCheck
(
build
);
// 从数仓中查询一组数据
if
(
list
.
size
()
==
0
)
{
JobKey
jobKey
=
JobKey
.
jobKey
(
"selectLikpCheck_once"
,
"DEFAULT"
);
// 删除任务
quartzScheduler
.
deleteJob
(
jobKey
);
return
;
}
list
.
forEach
(
target
->
{
// 遍历要检查的数据
Likp
source
=
sapMapper
.
selectLikpById
(
target
);
// 根据主键查询源库中的数据
String
operator
=
"none"
;
if
(
source
==
null
)
{
// 如果源库中没有数据
gpMapper
.
deleteLikp
(
target
);
// 删除数仓中的数据
operator
=
"delete"
;
}
else
{
// 源库中有数据
String
shash
=
SecureUtil
.
md5
(
JSONUtil
.
toJsonStr
(
source
));
// 源库中数据的hash结果
String
thash
=
target
.
getHashResult
();
// 数仓中数据的hash结果
if
(!
shash
.
equals
(
thash
))
{
// 如果hash结果不一致
source
.
setHashResult
(
shash
);
// ===============================
String
wadat_ist
=
source
.
getWadat_ist
();
String
lfuhr
=
source
.
getLfuhr
();
if
(
"00000000"
.
equals
(
wadat_ist
))
{
source
.
setWadat_ist2lfuhr
(
null
);
}
else
{
Date
date
=
DateUtil
.
parse
(
wadat_ist
+
lfuhr
,
"yyyyMMddHHmmss"
);
source
.
setWadat_ist2lfuhr
(
date
);
}
source
.
setErdat1
(
SomeUtils
.
caDate
(
source
.
getErdat
()));
// 日期00000000格式转换,已处理异常
source
.
setWadat_ist1
(
SomeUtils
.
caDate
(
source
.
getWadat_ist
()));
// 日期00000000格式转换,已处理异常
// ===============================
while
(
true
)
{
try
{
gpMapper
.
updateLikp
(
source
);
// 更新数据到数仓中
break
;
}
catch
(
RuntimeException
e
)
{
log
.
error
(
e
.
getMessage
());
ThreadUtil
.
safeSleep
(
500
);
}
}
ThreadUtil
.
safeSleep
(
500
);
}
}
redis1Template
.
opsForValue
().
set
(
"huazheng:check:Likp:rowNum"
,
target
.
getRowNum
());
if
(!
operator
.
equals
(
"none"
))
{
log
.
info
(
String
.
format
(
"selectLikpCheck --> rowNum:%s, operator:%s"
,
target
.
getRowNum
(),
operator
));
}
});
}
catch
(
Exception
e
)
{
redis1Template
.
opsForValue
().
set
(
"huazheng:checkError:Likp:rowNum"
,
getErrorInfoFromException
(
e
));
}
}
}
src/main/resources/mapper/greenplum/GPMapper_greenplum.xml
浏览文件 @
048d9e65
...
...
@@ -908,6 +908,10 @@
<select
id=
"selectLikpCheck"
parameterType=
"com.huazheng.project.hana.model.Likp"
resultType=
"com.huazheng.project.hana.model.Likp"
>
select * from Likp where rownum
>
#{rowNum} order by rownum limit 20
</select>
<select
id=
"selectLikpMaxRowNum"
resultType=
"long"
>
select max(rowNum) from Likp
</select>
<select
id=
"selectBkpf"
parameterType=
"com.huazheng.project.hana.model.Bkpf"
resultType=
"com.huazheng.project.hana.model.Bkpf"
>
select * from Bkpf where mandt = #{mandt} and bukrs = #{bukrs} and belnr = #{belnr} and gjahr = #{gjahr}
...
...
@@ -1195,7 +1199,7 @@
select * from Vbep where rownum
>
#{rowNum} order by rownum limit 20
</select>
<select
id=
"selectVbepMaxRowNum"
resultType=
"long"
>
select max(rowNum) from
Zsd06
select max(rowNum) from
Vbep
</select>
...
...
src/main/resources/spring-init.xml
浏览文件 @
048d9e65
...
...
@@ -1645,6 +1645,16 @@
</property>
<property
name=
"cronExpression"
value=
"* * * * * ?"
/>
</bean>
<bean
class=
"org.springframework.scheduling.quartz.CronTriggerFactoryBean"
>
<property
name=
"jobDetail"
>
<bean
parent=
"methodJobDetail"
>
<property
name=
"name"
value=
"selectLikpCheck_once"
/>
<property
name=
"targetObject"
ref=
"deleteUpdateJobServiceImpl"
/>
<property
name=
"targetMethod"
value=
"selectLikpCheck"
/>
</bean>
</property>
<property
name=
"cronExpression"
value=
"* * * * * ?"
/>
</bean>
</list>
...
...
编写
预览
Markdown
格式
0%
重试
或
添加新文件
添加附件
取消
您添加了
0
人
到此讨论。请谨慎行事。
请先完成此评论的编辑!
取消
请
注册
或者
登录
后发表评论