Skip to content

H
huazheng-project-flink
提交 2e159274 作者: think
操作

增加表 Mkpf Mseg

上级 9fffd16d
正在显示 包含 578 行增加 和 1 行删除
src/main/java/com/huazheng/project/HZDataStream.java
......@@ -31,6 +31,8 @@ import com.huazheng.project.greenplum.source.hana.LikpSource;
import com.huazheng.project.greenplum.source.hana.LipsSource;
import com.huazheng.project.greenplum.source.hana.MaktSource;
import com.huazheng.project.greenplum.source.hana.MaraSource;
import com.huazheng.project.greenplum.source.hana.MkpfSource;
import com.huazheng.project.greenplum.source.hana.MsegSource;
import com.huazheng.project.greenplum.source.hana.MskaSource;
import com.huazheng.project.greenplum.source.hana.Pa0002Source;
import com.huazheng.project.greenplum.source.hana.S066Source;
......@@ -78,6 +80,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -193,6 +197,7 @@ redis-cli -n 1 --raw keys "huazheng*ikp*" | xargs redis-cli -n 1 del
private static Kna1Source kna1Source;
private static KnvvSource knvvSource;
private static AfvcSource afvcSource;
private static MkpfSource mkpfSource;
private static KnkkSource knkkSource;
private static MaraSource maraSource;
private static MaktSource maktSource;
......@@ -224,6 +229,7 @@ redis-cli -n 1 --raw keys "huazheng*ikp*" | xargs redis-cli -n 1 del
private static SysSAPreturnNoSource sysSAPreturnNoSource;
private static TasksSource tasksSource;
private static MsegSource msegSource;
static {
context = new ClassPathXmlApplicationContext(new String[] { "spring-init.xml" });
......@@ -255,7 +261,8 @@ redis-cli -n 1 --raw keys "huazheng*ikp*" | xargs redis-cli -n 1 del
kna1Source = (Kna1Source) context.getBean("kna1Source");
knvvSource = (KnvvSource) context.getBean("knvvSource");
afvcSource = (AfvcSource) context.getBean("afvcSource");
knkkSource = (KnkkSource) context.getBean("knkkSource");
mkpfSource = (MkpfSource) context.getBean("mkpfSource");
knkkSource = (KnkkSource) context.getBean("knkkSource");
maraSource = (MaraSource) context.getBean("maraSource");
maktSource = (MaktSource) context.getBean("maktSource");
zsd06Source = (Zsd06Source) context.getBean("zsd06Source");
......@@ -286,6 +293,7 @@ redis-cli -n 1 --raw keys "huazheng*ikp*" | xargs redis-cli -n 1 del
sysSAPreturnNoSource = (SysSAPreturnNoSource) context.getBean("sysSAPreturnNoSource");
tasksSource = (TasksSource) context.getBean("tasksSource");
msegSource = (MsegSource) context.getBean("msegSource");
}
......@@ -899,6 +907,8 @@ redis-cli -n 1 --raw keys "huazheng*ikp*" | xargs redis-cli -n 1 del
env.addSource(personCompSource).setParallelism(1).name("输入PersonComp队列").flatMap(greenPlumFlatMapFunction).returns(PersonComp.class).setParallelism(1).name("拉取PersonComp数据").addSink(greenPlumRichSinkFunction).setParallelism(1).name("输出PersonComp数据");
env.addSource(afvcSource).setParallelism(1).name("输入Afvc队列").flatMap(greenPlumFlatMapFunction).returns(Afvc.class).setParallelism(1).name("拉取Afvc数据").addSink(greenPlumRichSinkFunction).setParallelism(1).name("输出Afvc数据");
env.addSource(mkpfSource).setParallelism(1).name("输入Mkpf队列").flatMap(greenPlumFlatMapFunction).returns(Mkpf.class).setParallelism(1).name("拉取Mkpf数据").addSink(greenPlumRichSinkFunction).setParallelism(1).name("输出Mkpf数据");
env.addSource(msegSource).setParallelism(1).name("输入Mseg队列").flatMap(greenPlumFlatMapFunction).returns(Mseg.class).setParallelism(1).name("拉取Mseg数据").addSink(greenPlumRichSinkFunction).setParallelism(1).name("输出Mseg数据");
env.execute("华正数据迁移任务");
}
......
src/main/java/com/huazheng/project/greenplum/mapper/GPMapper.java
......@@ -24,6 +24,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -64,6 +66,24 @@ import com.huazheng.project.mysql.model.TransformNewNode;
@CacheConfig(cacheNames = "huazheng")
public interface GPMapper {
@Cacheable(key = "#root.method.name+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr+','+#p0.zeile", unless="#result == null")
public Mseg selectMseg(Mseg mseg); // 查询替代删除
public void insertMseg(Mseg element);
@CacheEvict(key = "'selectMseg'+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr+','+#p0.zeile")
public void deleteMseg(Mseg item);
@CacheEvict(key = "'selectMseg'+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr+','+#p0.zeile")
public void updateMseg(Mseg element);
public List<Mseg> selectMsegCheck(Mseg build);
@Cacheable(key = "#root.method.name+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr", unless="#result == null")
public Mkpf selectMkpf(Mkpf mkpf); // 查询替代删除
public void insertMkpf(Mkpf element);
@CacheEvict(key = "'selectMkpf'+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr")
public void deleteMkpf(Mkpf item);
@CacheEvict(key = "'selectMkpf'+':'+#p0.mandt+','+#p0.mblnr+','+#p0.mjahr")
public void updateMkpf(Mkpf element);
public List<Mkpf> selectMkpfCheck(Mkpf build);
@Cacheable(key = "#root.method.name+':'+#p0.mandt+','+#p0.aufnr", unless="#result == null")
public Aufk selectAufk(Aufk aufk); // 查询替代删除
public void insertAufk(Aufk element);
......
src/main/java/com/huazheng/project/greenplum/service/impl/DeleteUpdateJobServiceImpl.java
......@@ -32,6 +32,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -956,6 +958,8 @@ public class DeleteUpdateJobServiceImpl {
selectZsdfhzlCheck(); // 40
selectTvkbtCheck(); //
selectAfvcCheck();
selectMkpfCheck();
selectMsegCheck();
}
private void selectAfkoCheck() {
try {
......@@ -1376,6 +1380,93 @@ public class DeleteUpdateJobServiceImpl {
redis1Template.opsForValue().set("huazheng:checkError:Afvc:rowNum", getErrorInfoFromException(e));
}
}
private void selectMkpfCheck() {
try {
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:check:Mkpf:rowNum", "0");
String rowNum = opsForValue.get("huazheng:check:Mkpf:rowNum");
Mkpf build = Mkpf.builder().rowNum(rowNum).build();
List<Mkpf> list = gpMapper.selectMkpfCheck(build); // 从数仓中查询一组数据
if (list.size() == 0) {
redis1Template.opsForValue().set("huazheng:check:Mkpf:rowNum", "0"); // 计数器复位
ThreadUtil.sleep(1000); // 没有数据了，休眠一下
}
list.forEach(target -> { // 遍历要检查的数据
Mkpf source = sapMapper.selectMkpfById(target); // 根据主键查询源库中的数据
String operator = "none";
if (source == null) { // 如果源库中没有数据
gpMapper.deleteMkpf(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);
while (true) {
try {
gpMapper.updateMkpf(source); // 更新数据到数仓中
break;
} catch (RuntimeException e) {
log.error(e.getMessage());ThreadUtil.safeSleep(500);
}
}
ThreadUtil.safeSleep(500);
}
}
redis1Template.opsForValue().set("huazheng:check:Mkpf:rowNum", target.getRowNum());
if (!operator.equals("none")) {
log.info(String.format("selectMkpfCheck --> rowNum:%s, operator:%s", target.getRowNum(), operator));
}
});
} catch (Exception e) {
redis1Template.opsForValue().set("huazheng:checkError:Mkpf:rowNum", getErrorInfoFromException(e));
}
}
private void selectMsegCheck() {
try {
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:check:Mseg:rowNum", "0");
String rowNum = opsForValue.get("huazheng:check:Mseg:rowNum");
Mseg build = Mseg.builder().rowNum(rowNum).build();
List<Mseg> list = gpMapper.selectMsegCheck(build); // 从数仓中查询一组数据
if (list.size() == 0) {
redis1Template.opsForValue().set("huazheng:check:Mseg:rowNum", "0"); // 计数器复位
ThreadUtil.sleep(1000); // 没有数据了，休眠一下
}
list.forEach(target -> { // 遍历要检查的数据
Mseg source = sapMapper.selectMsegById(target); // 根据主键查询源库中的数据
String operator = "none";
if (source == null) { // 如果源库中没有数据
gpMapper.deleteMseg(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);
while (true) {
try {
gpMapper.updateMseg(source); // 更新数据到数仓中
break;
} catch (RuntimeException e) {
log.error(e.getMessage());ThreadUtil.safeSleep(500);
}
}
ThreadUtil.safeSleep(500);
}
}
redis1Template.opsForValue().set("huazheng:check:Mseg:rowNum", target.getRowNum());
if (!operator.equals("none")) {
log.info(String.format("selectMsegCheck --> rowNum:%s, operator:%s", target.getRowNum(), operator));
}
});
} catch (Exception e) {
redis1Template.opsForValue().set("huazheng:checkError:Mseg:rowNum", getErrorInfoFromException(e));
}
}
// 数据量过大不再执行任务
private void selectKonvCheck() {
try {
......
src/main/java/com/huazheng/project/greenplum/service/impl/GPServiceImpl.java
......@@ -27,6 +27,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -1972,4 +1974,58 @@ public class GPServiceImpl {
}
}
public void processMkpf(Mkpf data, Collector<Mkpf> out) {
try {
Mkpf exist = gpMapper.selectMkpf(data);
if (exist != null) {
data.setExist(true); // 已经在库
}
out.collect(data);
} catch (Exception e) {
redis1Template.opsForHash().put("huazheng:Mkpf:error", "processMkpf", getErrorInfoFromException(e));
log.error(e.getMessage());
}
}
public void sinkMkpf(Mkpf element) {
try {
if (element.isExist() == false) {
log.debug("GPServiceImpl.sinkMkpf()");
gpMapper.insertMkpf(element);
}
} catch (RuntimeException e) {
redis1Template.opsForHash().put("huazheng:Mkpf:error", "sinkMkpf", getErrorInfoFromException(e));
log.error(e.getMessage());
} catch (Exception e) {
redis1Template.opsForHash().put("huazheng:Mkpf:error", "sinkMkpf", getErrorInfoFromException(e));
log.error(e.getMessage());
}
}
public void processMseg(Mseg data, Collector<Mseg> out) {
try {
Mseg exist = gpMapper.selectMseg(data);
if (exist != null) {
data.setExist(true); // 已经在库
}
out.collect(data);
} catch (Exception e) {
redis1Template.opsForHash().put("huazheng:Mseg:error", "processMseg", getErrorInfoFromException(e));
log.error(e.getMessage());
}
}
public void sinkMseg(Mseg element) {
try {
if (element.isExist() == false) {
log.debug("GPServiceImpl.sinkMseg()");
gpMapper.insertMseg(element);
}
} catch (RuntimeException e) {
redis1Template.opsForHash().put("huazheng:Mseg:error", "sinkMseg", getErrorInfoFromException(e));
log.error(e.getMessage());
} catch (Exception e) {
redis1Template.opsForHash().put("huazheng:Mseg:error", "sinkMseg", getErrorInfoFromException(e));
log.error(e.getMessage());
}
}
}
src/main/java/com/huazheng/project/greenplum/service/impl/JobServiceImpl.java
......@@ -29,6 +29,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -122,6 +124,8 @@ public class JobServiceImpl {
selectAfkoNew();
selectAfpoNew();
selectAfvcNew();
selectMkpfNew();
selectMsegNew();
selectVbepNew();
selectT023tNew();
selectKna1New();
......@@ -562,6 +566,59 @@ public class JobServiceImpl {
}
}
}
private void selectMkpfNew() {
DefaultRedisScript<String> script = new DefaultRedisScript<String>();
script.setResultType(String.class);
script.setScriptSource(new ResourceScriptSource(new ClassPathResource("luascript/vbap4send.lua")));
List<String> keys = Arrays.asList("huazheng:Mkpf:sendcount", "huazheng:Mkpf:rowids", "huazheng:list:Mkpf");
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:Mkpf:sendcount", "0"); // 不存在则创建，存在则么有操作
opsForValue.setIfAbsent("huazheng:Mkpf:receivecount", "0"); // 不存在则创建，存在则么有操作
opsForValue.setIfAbsent("huazheng:Mkpf:rowids", "0"); // 不存在则创建，存在则么有操作
Long sendcount = Long.valueOf(opsForValue.get("huazheng:Mkpf:sendcount"));
Long receivecount = Long.valueOf(opsForValue.get("huazheng:Mkpf:receivecount"));
if (sendcount - receivecount <= 20) { // 如果发送数和消费数的差小于5则往队列中写数据
String rowids = opsForValue.get("huazheng:Mkpf:rowids"); // 标记id
Mkpf mkpf = Mkpf.builder().rowids(Long.valueOf(rowids)).build();
List<Mkpf> list = sapMapper.selectMkpfNew(mkpf);
if (!list.isEmpty()) {
list.forEach(item -> {
JSONObject json = JSONUtil.parseObj(item, false);
String execute = redis1Template.execute(script, keys, item.getRowids().toString(), json.toString());
log.info("标记时间回写 --> " + execute);
});
}
}
}
private void selectMsegNew() {
DefaultRedisScript<String> script = new DefaultRedisScript<String>();
script.setResultType(String.class);
script.setScriptSource(new ResourceScriptSource(new ClassPathResource("luascript/vbap4send.lua")));
List<String> keys = Arrays.asList("huazheng:Mseg:sendcount", "huazheng:Mseg:rowids", "huazheng:list:Mseg");
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:Mseg:sendcount", "0"); // 不存在则创建，存在则么有操作
opsForValue.setIfAbsent("huazheng:Mseg:receivecount", "0"); // 不存在则创建，存在则么有操作
opsForValue.setIfAbsent("huazheng:Mseg:rowids", "0"); // 不存在则创建，存在则么有操作
Long sendcount = Long.valueOf(opsForValue.get("huazheng:Mseg:sendcount"));
Long receivecount = Long.valueOf(opsForValue.get("huazheng:Mseg:receivecount"));
if (sendcount - receivecount <= 20) { // 如果发送数和消费数的差小于5则往队列中写数据
String rowids = opsForValue.get("huazheng:Mseg:rowids"); // 标记id
Mseg mseg = Mseg.builder().rowids(Long.valueOf(rowids)).build();
List<Mseg> list = sapMapper.selectMsegNew(mseg);
if (!list.isEmpty()) {
list.forEach(item -> {
JSONObject json = JSONUtil.parseObj(item, false);
String execute = redis1Template.execute(script, keys, item.getRowids().toString(), json.toString());
log.info("标记时间回写 --> " + execute);
});
}
}
}
private void selectAfpoNew() {
DefaultRedisScript<String> script = new DefaultRedisScript<String>();
script.setResultType(String.class);
......
src/main/java/com/huazheng/project/greenplum/source/hana/MkpfSource.java 0 → 100644
package com.huazheng.project.greenplum.source.hana;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Arrays;
import java.util.List;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import com.huazheng.project.HZDataStream;
import com.huazheng.project.hana.model.Mkpf;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class MkpfSource implements SourceFunction<Mkpf> {
private static final long serialVersionUID = 1L;
public String getErrorInfoFromException(Exception e) {
try {
StringWriter sw = new StringWriter();
PrintWriter pw = new PrintWriter(sw);
e.printStackTrace(pw);
return "\r\n" + sw.toString() + "\r\n";
} catch (Exception e2) {
return "bad getErrorInfoFromException";
}
}
@Override
public void run(SourceContext<Mkpf> ctx) throws Exception {
DefaultRedisScript<String> script = new DefaultRedisScript<String>();
script.setResultType(String.class);
script.setScriptSource(new ResourceScriptSource(new ClassPathResource("luascript/vbap.lua")));
List<String> keys = Arrays.asList("huazheng:Mkpf:sendcount", "huazheng:Mkpf:id", "huazheng:list:Mkpf", "huazheng:Mkpf:receivecount");
while (true) {
try {
String value = HZDataStream.redis1Template.execute(script, keys, "");
String[] values = value.toString().split("==========");
String checkString = values[0];
String[] split = checkString.split(", ");
boolean check = split[0].split(":")[1].equals(split[1].split(":")[1]);
if (values.length > 1) { // 有数据字符串
// log.info(msg + " " + value.toString() + " " + check);
log.info(checkString + " " + check);
Mkpf data = JSONUtil.toBean(values[1], Mkpf.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了，休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Mkpf:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
src/main/java/com/huazheng/project/greenplum/source/hana/MsegSource.java 0 → 100644
package com.huazheng.project.greenplum.source.hana;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Arrays;
import java.util.List;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import com.huazheng.project.HZDataStream;
import com.huazheng.project.hana.model.Mseg;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class MsegSource implements SourceFunction<Mseg> {
private static final long serialVersionUID = 1L;
public String getErrorInfoFromException(Exception e) {
try {
StringWriter sw = new StringWriter();
PrintWriter pw = new PrintWriter(sw);
e.printStackTrace(pw);
return "\r\n" + sw.toString() + "\r\n";
} catch (Exception e2) {
return "bad getErrorInfoFromException";
}
}
@Override
public void run(SourceContext<Mseg> ctx) throws Exception {
DefaultRedisScript<String> script = new DefaultRedisScript<String>();
script.setResultType(String.class);
script.setScriptSource(new ResourceScriptSource(new ClassPathResource("luascript/vbap.lua")));
List<String> keys = Arrays.asList("huazheng:Mseg:sendcount", "huazheng:Mseg:id", "huazheng:list:Mseg", "huazheng:Mseg:receivecount");
while (true) {
try {
String value = HZDataStream.redis1Template.execute(script, keys, "");
String[] values = value.toString().split("==========");
String checkString = values[0];
String[] split = checkString.split(", ");
boolean check = split[0].split(":")[1].equals(split[1].split(":")[1]);
if (values.length > 1) { // 有数据字符串
// log.info(msg + " " + value.toString() + " " + check);
log.info(checkString + " " + check);
Mseg data = JSONUtil.toBean(values[1], Mseg.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了，休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Mseg:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
src/main/java/com/huazheng/project/hana/mapper/SapMapper.java
......@@ -17,6 +17,8 @@ import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Mkpf;
import com.huazheng.project.hana.model.Mseg;
import com.huazheng.project.hana.model.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
......@@ -69,7 +71,11 @@ public interface SapMapper {
public List<T001w> selectT001wNew(T001w t001w);
public List<Mska> selectMskaNew(Mska mska);
public List<Zsdfhzl> selectZsdfhzlNew(Zsdfhzl zsdfhzl);
public List<Mkpf> selectMkpfNew(Mkpf mkpf);
public List<Mseg> selectMsegNew(Mseg mseg);
public Mseg selectMsegById(Mseg target);
public Mkpf selectMkpfById(Mkpf target);
public Afko selectAfkoById(Afko target);
public Afpo selectAfpoById(Afpo target);
public Afvc selectAfvcById(Afvc target);
......
src/main/java/com/huazheng/project/hana/model/Mkpf.java 0 → 100644
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.util.Date;
import org.springframework.format.annotation.DateTimeFormat;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Builder
public class Mkpf implements Serializable {
private static final long serialVersionUID = 1L;
/* === 组合键 === */
private String mandt; // 集团
private String mblnr; // 物料凭证编号
private String mjahr; // 物料凭证年度
/* === 组合键 === */
@DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss") // 页面写入数据库时格式化
@JSONField(format="yyyy-MM-dd HH:mm:ss") // 数据库导出页面时json格式化
private String bldat; // 凭证中的凭证日期
@DateTimeFormat(pattern="HH:mm:ss") // 页面写入数据库时格式化
@JSONField(format="HH:mm:ss") // 数据库导出页面时json格式化
private String cputm; // 输入时间
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记，不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记，不是字段
}
src/main/java/com/huazheng/project/hana/model/Mseg.java 0 → 100644
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.util.Date;
import org.springframework.format.annotation.DateTimeFormat;
import com.alibaba.fastjson.annotation.JSONField;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Builder
public class Mseg implements Serializable {
private static final long serialVersionUID = 1L;
/* === 组合键 === */
private String mandt; // 集团
private String mblnr; // 物料凭证编号
private String mjahr; // 物料凭证年度
private String zeile; // 凭证结构
/* === 组合键 === */
private String bwart; // 凭证类型
private String charg; // 批号
private String menge; // 数量
@DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss") // 页面写入数据库时格式化
@JSONField(format="yyyy-MM-dd HH:mm:ss") // 数据库导出页面时json格式化
private String budat_mkpf; // 凭证中的过帐日期
@DateTimeFormat(pattern="HH:mm:ss") // 页面写入数据库时格式化
@JSONField(format="HH:mm:ss") // 数据库导出页面时json格式化
private String cputm_mkpf; // 输入时间
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记，不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记，不是字段
}
src/main/resources/devtools/table/hana/mkpf.sql 0 → 100644
drop table mkpf;
CREATE TABLE mkpf (
mandt text,
mblnr text,
mjahr text,
bldat timestamp,
cputm time,
hashResult text,
rowNum serial,
PRIMARY KEY (mandt,mblnr,mjahr)
)
Distributed by (mandt,mblnr,mjahr);
src/main/resources/devtools/table/hana/mseg.sql 0 → 100644
drop table mseg;
CREATE TABLE mseg (
mandt text,
mblnr text,
mjahr text,
zeile text,
bwart text,
charg text,
menge text,
budat_mkpf timestamp,
cputm_mkpf time,
hashResult text,
rowNum serial,
PRIMARY KEY (mandt,mblnr,mjahr,zeile)
)
Distributed by (mandt,mblnr,mjahr,zeile);
src/main/resources/mapper/greenplum/GPMapper_greenplum.xml
......@@ -631,6 +631,44 @@
select * from Afvc where rownum &gt; #{rowNum} order by rownum limit 20
</select>
<select id="selectMseg" parameterType="com.huazheng.project.hana.model.Mseg" resultType="com.huazheng.project.hana.model.Mseg">
select * from Mseg where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr} and zeile = #{zeile}
</select>
<insert id="insertMseg" parameterType="com.huazheng.project.hana.model.Mseg">
insert into Mseg (mandt, mblnr, mjahr, zeile, bwart, charg, menge, budat_mkpf, cputm_mkpf, hashResult)
values(#{mandt},#{mblnr},#{mjahr},#{zeile},#{bwart},#{charg},#{menge},#{budat_mkpf},#{cputm_mkpf},#{hashResult})
</insert>
<delete id="deleteMseg" parameterType="com.huazheng.project.hana.model.Mseg">
delete from Mseg where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr} and zeile = #{zeile}
</delete>
<update id="updateMseg" parameterType="com.huazheng.project.hana.model.Mseg">
update Mseg set
mandt = #{mandt}, mblnr = #{mblnr}, mjahr = #{mjahr}, zeile = #{zeile}, bwart = #{bwart}, charg = #{charg}, menge = #{menge}, budat_mkpf = #{budat_mkpf}, cputm_mkpf = #{cputm_mkpf}, hashResult = #{hashResult}
where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr} and zeile = #{zeile}
</update>
<select id="selectMsegCheck" parameterType="com.huazheng.project.hana.model.Mseg" resultType="com.huazheng.project.hana.model.Mseg">
select * from Mseg where rownum &gt; #{rowNum} order by rownum limit 20
</select>
<select id="selectMkpf" parameterType="com.huazheng.project.hana.model.Mkpf" resultType="com.huazheng.project.hana.model.Mkpf">
select * from Mkpf where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr}
</select>
<insert id="insertMkpf" parameterType="com.huazheng.project.hana.model.Mkpf">
insert into Mkpf (mandt,mblnr,mjahr,bldat,cputm,hashResult)
values(#{mandt},#{mblnr},#{mjahr},#{bldat},#{cputm},#{hashResult})
</insert>
<delete id="deleteMkpf" parameterType="com.huazheng.project.hana.model.Mkpf">
delete from Mkpf where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr}
</delete>
<update id="updateMkpf" parameterType="com.huazheng.project.hana.model.Mkpf">
update Mkpf set
mandt = #{mandt}, mblnr = #{mblnr}, mjahr = #{mjahr}, bldat = #{bldat}, cputm = #{cputm}, hashResult = #{hashResult}
where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr}
</update>
<select id="selectMkpfCheck" parameterType="com.huazheng.project.hana.model.Mkpf" resultType="com.huazheng.project.hana.model.Mkpf">
select * from Mkpf where rownum &gt; #{rowNum} order by rownum limit 20
</select>
<select id="selectKonv" parameterType="com.huazheng.project.hana.model.Konv" resultType="com.huazheng.project.hana.model.Konv">
select * from Konv
where mandt = #{mandt} and knumv = #{knumv} and kposn = #{kposn} and stunr = #{stunr} and zaehk = #{zaehk}
......
src/main/resources/mapper/hana/SapMapper_hana.xml
......@@ -199,6 +199,23 @@
where "$rowid$" &gt; #{rowids} ${hana_mandt}
order by "$rowid$"
</select>
<select id="selectMkpfNew" parameterType="Mkpf" resultType="Mkpf">
select top 20 "$rowid$" as rowids,
mandt, mblnr, mjahr, bldat, cputm
from ${hana_user}.Mkpf
where "$rowid$" &gt; #{rowids} ${hana_mandt}
order by "$rowid$"
</select>
<select id="selectMsegNew" parameterType="Mseg" resultType="Mseg">
select top 20 "$rowid$" as rowids,
mandt, mblnr, mjahr, zeile, bwart, charg, menge, budat_mkpf, cputm_mkpf
from ${hana_user}.Mseg
where "$rowid$" &gt; #{rowids} ${hana_mandt}
order by "$rowid$"
</select>
<select id="selectKnkkNew" parameterType="Knkk" resultType="Knkk">
select top 20 "$rowid$" as rowids,
mandt,kunnr,kkber,klimk,skfor,ssobl,aedat
......@@ -329,6 +346,20 @@
from ${hana_user}.Afvc
where mandt = #{mandt} and aufpl = #{aufpl} and aplzl = #{aplzl}
</select>
<select id="selectMkpfById" parameterType="Mkpf" resultType="Mkpf">
select mandt, mblnr, mjahr, bldat, cputm
from ${hana_user}.Mkpf
where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr}
</select>
<select id="selectMsegById" parameterType="Mseg" resultType="Mseg">
select mandt, mblnr, mjahr, zeile, bwart, charg, menge, budat_mkpf, cputm_mkpf
from ${hana_user}.Mseg
where mandt = #{mandt} and mblnr = #{mblnr} and mjahr = #{mjahr} and zeile = #{zeile}
</select>
<select id="selectKonvById" parameterType="Konv" resultType="Konv">
select mandt,knumv,kposn,stunr,zaehk,kntyp,kbetr
from ${hana_user}.Konv
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论