提交 afcbbc9e 作者: guofeng

Initial commit

上级
target/
!.mvn/wrapper/maven-wrapper.jar
### STS ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
### NetBeans ###
nbproject/private/
build/
nbbuild/
dist/
nbdist/
.nb-gradle/
#redis.hostName=172.100.103.112
#redis.port=6379
#redis.database=10
#redis.maxTotal=200
#redis.maxIdle=200
#redis.maxWaitMillis=2000
#kafka.tianjin.serverIn=101.101.140.30:9092
#kafka.tianjin.topicIn=eriinfo
#kafka.tianjin.groupId=ehualu-bigdata-flink
#kafka.tianjin.serverOut=192.168.8.133:9092
#kafka.tianjin.topicOut=eriinfo-out
#kafka.jcbk.serverIn=172.100.103.112:9092
#kafka.jcbk.topicIn=jcbk
#kafka.jcbk.groupId=ehualu-bigdata-flink
#
#kafka.jcbk.serverOut=172.100.103.112:9092
#kafka.jcbk.topicOut=jcbk-out
#kafka.jcbk.topicOut1=jcbk-out-statistics4warn
#kafka.jcbk.topicOut2=jcbk-out-monitor4bayName
#kafka.jcbk.topicOut3=jcbk-out-monitor4trajectory
# ----------------------------------------------+
# OFF 0 | WARN 3 | TRACE 6 |
# FATAL 1 | INFO 4 | ALL 7 |
# ERROR 2 | DEBUG 5 | |
# ----------------------------------------------+
log4j.rootLogger=INFO, console
log4j.logger.com.mengdee=INFO
# ============================== console =====================================
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %5p %c{1}:%L - %m %n
#log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
#log4j.appender.file.layout=org.apache.log4j.PatternLayout
#log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %5p %c{1}:%L - %m %n
#log4j.appender.file.file=/usr/local/spark-app/logs/spark-area-security-manage-data.log
#log4j.appender.file.Threshold=WARN
<?xml version="1.0" encoding="UTF-8"?>
<Configuration package="log4j.test"
status="WARN">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
</Console>
</Appenders>
<Loggers>
<Logger name="log4j.test.Log4jTest" level="trace">
<AppenderRef ref="Console"/>
</Logger>
<Root level="info">
<AppenderRef ref="Console"/>
</Root>
</Loggers>
</Configuration>
\ No newline at end of file
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="com.huazheng.project.stream.service" />
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<list>
<value>classpath:application.properties</value>
</list>
</property>
</bean>
<bean id="proper" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
<property name="locations">
<array>
<value>classpath:application.properties</value>
</array>
</property>
</bean>
<!-- 定义连接池 -->
<bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
<property name="maxTotal" value="${redis.maxTotal}" />
<property name="maxIdle" value="${redis.maxIdle}" />
<property name="MaxWaitMillis" value="${redis.maxWaitMillis}" />
<property name="testOnBorrow" value="true" />
</bean>
<!-- 定义连接工厂 -->
<bean id="jedisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="hostName" value="${redis.hostName}" />
<property name="port" value="${redis.port}" />
<property name="database" value="${redis.database}" />
<property name="usePool" value="true" />
<property name="poolConfig" ref="jedisPoolConfig" />
</bean>
<!-- 定义模板 -->
<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
<property name="connectionFactory" ref="jedisConnectionFactory" />
<property name="keySerializer">
<bean class="org.springframework.data.redis.serializer.StringRedisSerializer" />
</property>
<property name="valueSerializer">
<bean class="org.springframework.data.redis.serializer.StringRedisSerializer" />
</property>
<property name="hashKeySerializer">
<bean class="org.springframework.data.redis.serializer.StringRedisSerializer" />
</property>
<property name="hashValueSerializer">
<bean class="org.springframework.data.redis.serializer.StringRedisSerializer" />
</property>
</bean>
</beans>
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.huazheng.project</groupId>
<artifactId>huazheng-project-flink</artifactId>
<name>Flink Quickstart Job</name>
<version>0.0.1-SNAPSHOT</version>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.eclipse.m2e</groupId>
<artifactId>lifecycle-mapping</artifactId>
<version>1.0.0</version>
<configuration>
<lifecycleMappingMetadata>
<pluginExecutions>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<versionRange>[3.1.1,)</versionRange>
<goals>
<goal>shade</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore />
</action>
</pluginExecution>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<versionRange>[3.1,)</versionRange>
<goals>
<goal>testCompile</goal>
<goal>compile</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore />
</action>
</pluginExecution>
</pluginExecutions>
</lifecycleMappingMetadata>
</configuration>
</plugin>
</plugins>
</pluginManagement>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<version>3.1.1</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>org.apache.logging.log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer>
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer>
<resource>META-INF/spring.schemas</resource>
</transformer>
<transformer>
<resource>META-INF/spring.tooling</resource>
</transformer>
<transformer>
<mainClass>com.huazheng.project.HZDataStream</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<releases>
<enabled>false</enabled>
</releases>
<snapshots />
<id>apache.snapshots</id>
<name>Apache Development Snapshot Repository</name>
<url>https://repository.apache.org/content/repositories/snapshots/</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>cdh.repo</id>
<name>Cloudera Repositories</name>
<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
</repository>
<repository>
<releases>
<enabled>false</enabled>
</releases>
<snapshots />
<id>cdh.snapshots.repo</id>
<name>Cloudera Snapshots Repository</name>
<url>https://repository.cloudera.com/artifactory/libs-snapshot-local</url>
</repository>
<repository>
<releases />
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>central</id>
<url>http://repo1.maven.org/maven2/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.11.1</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>flink-core</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>commons-lang3</artifactId>
<groupId>org.apache.commons</groupId>
</exclusion>
<exclusion>
<artifactId>commons-math3</artifactId>
<groupId>org.apache.commons</groupId>
</exclusion>
<exclusion>
<artifactId>jsr305</artifactId>
<groupId>com.google.code.findbugs</groupId>
</exclusion>
<exclusion>
<artifactId>force-shading</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.11.1</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>flink-runtime_2.11</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>flink-shaded-guava</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>flink-core</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>commons-math3</artifactId>
<groupId>org.apache.commons</groupId>
</exclusion>
<exclusion>
<artifactId>jsr305</artifactId>
<groupId>com.google.code.findbugs</groupId>
</exclusion>
<exclusion>
<artifactId>force-shading</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.11.1</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<artifactId>flink-optimizer_2.11</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>commons-cli</artifactId>
<groupId>commons-cli</groupId>
</exclusion>
<exclusion>
<artifactId>flink-runtime_2.11</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>flink-core</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
<exclusion>
<artifactId>jsr305</artifactId>
<groupId>com.google.code.findbugs</groupId>
</exclusion>
<exclusion>
<artifactId>force-shading</artifactId>
<groupId>org.apache.flink</groupId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<properties>
<java.version>1.8</java.version>
<scala.binary.version>2.11</scala.binary.version>
<log4j.version>2.12.1</log4j.version>
<flink.version>1.11.1</flink.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.target>${java.version}</maven.compiler.target>
</properties>
</project>
差异被折叠。 点击展开。
package com.huazheng.project.greenplum.flatmap;
import java.io.Serializable;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;
import com.huazheng.project.greenplum.service.impl.GPServiceImpl;
import cn.hutool.core.util.ReflectUtil;
@Service
public class GreenPlumFlatMapFunction<T, O> implements FlatMapFunction<T, O>, ApplicationContextAware, Serializable {
private static final long serialVersionUID = 1L;
private static GPServiceImpl gpserviceImpl;
@Override
public void flatMap(T value, Collector<O> out) throws Exception {
String simpleName = value.getClass().getSimpleName();
ReflectUtil.invoke(gpserviceImpl, "process" + simpleName, value, out);
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
gpserviceImpl = (GPServiceImpl) applicationContext.getBean("GPServiceImpl");
}
}
package com.huazheng.project.greenplum.richsink;
import java.io.Serializable;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;
import com.huazheng.project.greenplum.service.impl.GPServiceImpl;
import cn.hutool.core.util.ReflectUtil;
@Service
public class GreenPlumRichSinkFunction<T> extends RichSinkFunction<T> implements ApplicationContextAware, Serializable {
private static final long serialVersionUID = 1L;
private static GPServiceImpl gpserviceImpl;
@Override
public void invoke(T value) throws Exception {
String simpleName = value.getClass().getSimpleName();
ReflectUtil.invoke(gpserviceImpl, "sink" + simpleName, value);
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
gpserviceImpl = (GPServiceImpl) applicationContext.getBean("GPServiceImpl");
}
}
This source diff could not be displayed because it is too large. You can view the blob instead.
package com.huazheng.project.greenplum.service.impl;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import com.huazheng.project.greenplum.mapper.GPMapper;
import com.huazheng.project.hana.model.CAVThisMonth;
import com.huazheng.project.mssql3.mapper.Cinderellaw2Mapper;
import com.huazheng.project.mssql3.model.PersonComp;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class OtherJobServiceImpl {
@Autowired
private Cinderellaw2Mapper cinderellaw2Mapper;
@Autowired
private RedisTemplate<String, String> redis1Template;
@Autowired
private GPMapper gpMapper;
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";
}
}
// 同步人员对照数据,非flink流程
public void caPersonComp() {
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:PersonComp:sendcount", "huazheng:PersonComp:rowids", "huazheng:list:PersonComp");
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:PersonComp:sendcount", "0"); // 不存在则创建,存在则么有操作
opsForValue.setIfAbsent("huazheng:PersonComp:receivecount", "0"); // 不存在则创建,存在则么有操作
opsForValue.setIfAbsent("huazheng:PersonComp:rowids", "0"); // 不存在则创建,存在则么有操作
Long sendcount = Long.valueOf(opsForValue.get("huazheng:PersonComp:sendcount"));
Long receivecount = Long.valueOf(opsForValue.get("huazheng:PersonComp:receivecount"));
if (sendcount - receivecount <= 20) { // 如果发送数和消费数的差小于5则往队列中写数据
String rowids = opsForValue.get("huazheng:PersonComp:rowids"); // 标记id
PersonComp personComp = PersonComp.builder().rowids(Long.valueOf(rowids)).build();
List<PersonComp> list = cinderellaw2Mapper.selectPersonCompNew(personComp);
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);
});
}
}
}
// 计算回款
public void caPayment() {
// 查询客户每月回款列表
List<CAVThisMonth> groupList = gpMapper.selectCAVThisMonthByEveryMonth();
// 按照客户和而货币分组
Map<String, List<CAVThisMonth>> groupItemsCollect = groupList.stream().collect(Collectors.groupingBy(e -> e.getKunnr() + "#" + e.getWaers()));
groupItemsCollect.forEach((kunnr2Waers, groupItems) -> {
BigDecimal byhkTotal = new BigDecimal(0); // 当月当前累积回款
BigDecimal byxseTotal = new BigDecimal(0); // 当月当前累积销售额
// 遍历每个客户的每月回款清单,开始计算
for (int i = 0; i < groupItems.size(); i++) {
CAVThisMonth item = groupItems.get(i);
String kunnr = item.getKunnr(); // 客户
String waers = item.getWaers(); // 货币
String budatb = item.getBudatb(); // 上个月
// 抽取符合条件的上月的数据
List<CAVThisMonth> collect = groupItems.stream() //
.filter(pp -> pp.getKunnr().equals(kunnr)) //
.filter(pp -> pp.getWaers().equals(waers)) //
.filter(pp -> pp.getBudatb().equals(budatb)) //
.collect(Collectors.toList());
BigDecimal ljsyysk = new BigDecimal(0); // 上月,累积剩余应收款
// 如果上月有数据
if (!collect.isEmpty() && collect.get(0).getLjsyysk() != null) {
ljsyysk = ljsyysk.add(collect.get(0).getLjsyysk()); // 上月,累积剩余应收款
}
BigDecimal byhk = item.getByhk(); // 当月回款
// BigDecimal byxse = item.getByxse(); // 当月销售额
item.setDyjhk(byhk.subtract(ljsyysk)); // 计算,当月净回款
// item.setLjsyysk(byxse.subtract(item.getDyjhk())); // 计算,累积剩余应收款
byhkTotal = byhkTotal.add(item.getByhk()); // 当月回款
byxseTotal = byxseTotal.add(item.getByxse()); // 当月销售额
item.setLjsyysk(byxseTotal.subtract(byhkTotal)); // 计算,累积剩余应收款
}
// 当月销售额0修正
List<CAVThisMonth> zeroCollect = groupItems.stream().filter(pp -> pp.getByxse().compareTo(BigDecimal.ZERO) == 0).collect(Collectors.toList());
if (!zeroCollect.isEmpty()) {
zeroCollect.forEach(item -> {
item.setDyjhk(new BigDecimal(0)); // 计算,当月净回款
item.setLjsyysk(new BigDecimal(0)); // 计算,累积剩余应收款
});
}
});
// 数据输出
groupList.forEach(item -> {
CAVThisMonth target = gpMapper.selectCAVThisMonth(item); // 先查询数据是否存在
String kunnr = item.getKunnr(); // 客户
String waers = item.getWaers(); // 货币
String budatd = item.getBudatd();
String budatb = item.getBudatb();
if (target == null) { // 并不存在
gpMapper.insertCAVThisMonth(item); // 数据入库
log.info("caPayment -> insert {}, {}, {}, {}", kunnr, waers, budatd, budatb);
} else {
String shash = SecureUtil.md5(JSONUtil.toJsonStr(item)); // 源库中数据的hash结果
String thash = target.getHashResult(); // 数仓中数据的hash结果
if (!shash.equals(thash)) { // 如果hash结果不一致
item.setHashResult(shash);
gpMapper.updateCAVThisMonth(item); // 更新数据到数仓中
log.info("caPayment -> update {}, {}, {}, {}", kunnr, waers, budatd, budatb);
}
}
});
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class AfkoSource implements SourceFunction<String> {
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<String> 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:Afko:sendcount", "huazheng:Afko:id", "huazheng:list:Afko", "huazheng:Afko: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Afko:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class AfpoSource implements SourceFunction<String> {
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<String> 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:Afpo:sendcount", "huazheng:Afpo:id", "huazheng:list:Afpo", "huazheng:Afpo: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Afpo:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class AufkSource implements SourceFunction<String> {
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<String> 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:Aufk:sendcount", "huazheng:Aufk:id", "huazheng:list:Aufk", "huazheng:Aufk: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Aufk:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class AufmSource implements SourceFunction<String> {
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<String> 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:Aufm:sendcount", "huazheng:Aufm:id", "huazheng:list:Aufm", "huazheng:Aufm: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Aufm:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class BsadSource implements SourceFunction<String> {
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<String> 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:bsad:sendcount", "huazheng:bsad:id", "huazheng:list:bsad", "huazheng:bsad: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:bsad:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class BsidSource implements SourceFunction<String> {
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<String> 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:bsid:sendcount", "huazheng:bsid:id", "huazheng:list:bsid", "huazheng:bsid: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:bsid:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class Kna1Source implements SourceFunction<String> {
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<String> 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:Kna1:sendcount", "huazheng:Kna1:id", "huazheng:list:Kna1", "huazheng:Kna1: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Kna1:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class KnkkSource implements SourceFunction<String> {
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<String> 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:Knkk:sendcount", "huazheng:Knkk:id", "huazheng:list:Knkk", "huazheng:Knkk: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Knkk:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class KnvvSource implements SourceFunction<String> {
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<String> 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:Knvv:sendcount", "huazheng:Knvv:id", "huazheng:list:Knvv", "huazheng:Knvv: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Knvv:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class KonvSource implements SourceFunction<String> {
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<String> 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:Konv:sendcount", "huazheng:Konv:id", "huazheng:list:Konv", "huazheng:Konv: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Konv:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class LikpSource implements SourceFunction<String> {
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<String> 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:Likp:sendcount", "huazheng:Likp:id", "huazheng:list:Likp", "huazheng:Likp: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Likp:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class LipsSource implements SourceFunction<String> {
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<String> 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:Lips:sendcount", "huazheng:Lips:id", "huazheng:list:Lips", "huazheng:Lips: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Lips:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class MaktSource implements SourceFunction<String> {
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<String> 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:Makt:sendcount", "huazheng:Makt:id", "huazheng:list:Makt", "huazheng:Makt: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Makt:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class MaraSource implements SourceFunction<String> {
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<String> 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:Mara:sendcount", "huazheng:Mara:id", "huazheng:list:Mara", "huazheng:Mara: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Mara:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class MskaSource implements SourceFunction<String> {
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<String> 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:Mska:sendcount", "huazheng:Mska:id", "huazheng:list:Mska", "huazheng:Mska: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Mska:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class Pa0002Source implements SourceFunction<String> {
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<String> 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:Pa0002:sendcount", "huazheng:Pa0002:id", "huazheng:list:Pa0002", "huazheng:Pa0002: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Pa0002:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class S066Source implements SourceFunction<String> {
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<String> 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:S066:sendcount", "huazheng:S066:id", "huazheng:list:S066", "huazheng:S066: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:S066:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class S067Source implements SourceFunction<String> {
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<String> 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:S067:sendcount", "huazheng:S067:id", "huazheng:list:S067", "huazheng:S067: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:S067:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class T001wSource implements SourceFunction<String> {
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<String> 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:T001w:sendcount", "huazheng:T001w:id", "huazheng:list:T001w", "huazheng:T001w: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:T001w:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class T023tSource implements SourceFunction<String> {
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<String> 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:T023t:sendcount", "huazheng:T023t:id", "huazheng:list:T023t", "huazheng:T023t: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:T023t:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TvkbtSource implements SourceFunction<String> {
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<String> 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:Tvkbt:sendcount", "huazheng:Tvkbt:id", "huazheng:list:Tvkbt", "huazheng:Tvkbt: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Tvkbt:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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.Vbak;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbakSource implements SourceFunction<Vbak> {
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<Vbak> 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:Vbak:sendcount", "huazheng:Vbak:id", "huazheng:list:Vbak", "huazheng:Vbak: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);
Vbak data = JSONUtil.toBean(values[1], Vbak.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbak:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbapSource implements SourceFunction<String> {
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<String> 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:vbap:sendcount", "huazheng:vbap:flagDate", "huazheng:list:vbap", "huazheng:vbap: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:vbap:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbepSource implements SourceFunction<String> {
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<String> 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:Vbep:sendcount", "huazheng:Vbep:id", "huazheng:list:Vbep", "huazheng:Vbep: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbep:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbpaSource implements SourceFunction<String> {
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<String> 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:Vbpa:sendcount", "huazheng:Vbpa:id", "huazheng:list:Vbpa", "huazheng:Vbpa: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbpa:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbrkSource implements SourceFunction<String> {
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<String> 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:Vbrk:sendcount", "huazheng:Vbrk:id", "huazheng:list:Vbrk", "huazheng:Vbrk: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbrk:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbrpSource implements SourceFunction<String> {
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<String> 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:Vbrp:sendcount", "huazheng:Vbrp:id", "huazheng:list:Vbrp", "huazheng:Vbrp: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbrp:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class VbukSource implements SourceFunction<String> {
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<String> 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:Vbuk:sendcount", "huazheng:Vbuk:id", "huazheng:list:Vbuk", "huazheng:Vbuk: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Vbuk:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class ZmdpcSource implements SourceFunction<String> {
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<String> 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:Zmdpc:sendcount", "huazheng:Zmdpc:id", "huazheng:list:Zmdpc", "huazheng:Zmdpc: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Zmdpc:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class Zsd06Source implements SourceFunction<String> {
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<String> 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:Zsd06:sendcount", "huazheng:Zsd06:id", "huazheng:list:Zsd06", "huazheng:Zsd06: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Zsd06:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
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.Zsdfhzl;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class ZsdfhzlSource implements SourceFunction<Zsdfhzl> {
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<Zsdfhzl> 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:Zsdfhzl:sendcount", "huazheng:Zsdfhzl:id", "huazheng:list:Zsdfhzl", "huazheng:Zsdfhzl: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);
Zsdfhzl data = JSONUtil.toBean(values[1], Zsdfhzl.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Zsdfhzl:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SalesContractProcessSource implements SourceFunction<String> {
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<String> 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:SalesContractProcess:sendcount", "huazheng:SalesContractProcess:id", "huazheng:list:SalesContractProcess", "huazheng:SalesContractProcess: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SalesContractProcess:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SalesForecastProcessSource implements SourceFunction<String> {
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<String> 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:SalesForecastProcess:sendcount", "huazheng:SalesForecastProcess:id", "huazheng:list:SalesForecastProcess", "huazheng:SalesForecastProcess: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SalesForecastProcess:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SampleApplicationProcessMingXiSource implements SourceFunction<String> {
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<String> 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:SampleApplicationProcessMingXi:sendcount", "huazheng:SampleApplicationProcessMingXi:id", "huazheng:list:SampleApplicationProcessMingXi", "huazheng:SampleApplicationProcessMingXi: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SampleApplicationProcessMingXi:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SampleApplicationProcessSource implements SourceFunction<String> {
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<String> 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:SampleApplicationProcess:sendcount", "huazheng:SampleApplicationProcess:id", "huazheng:list:SampleApplicationProcess", "huazheng:SampleApplicationProcess: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SampleApplicationProcess:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SampleClosingProcessMingXiSource implements SourceFunction<String> {
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<String> 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:SampleClosingProcessMingXi:sendcount", "huazheng:SampleClosingProcessMingXi:id", "huazheng:list:SampleClosingProcessMingXi", "huazheng:SampleClosingProcessMingXi: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SampleClosingProcessMingXi:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SampleClosingProcessSource implements SourceFunction<String> {
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<String> 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:SampleClosingProcess:sendcount", "huazheng:SampleClosingProcess:id", "huazheng:list:SampleClosingProcess", "huazheng:SampleClosingProcess: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SampleClosingProcess:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql2;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class SysSAPreturnNoSource implements SourceFunction<String> {
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<String> 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:SysSAPreturnNo:sendcount", "huazheng:SysSAPreturnNo:id", "huazheng:list:SysSAPreturnNo", "huazheng:SysSAPreturnNo: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:SysSAPreturnNo:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql2;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TKeHuSource implements SourceFunction<String> {
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<String> 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:TKeHu:sendcount", "huazheng:TKeHu:flagDate", "huazheng:list:TKeHu", "huazheng:TKeHu: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:TKeHu:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mssql2;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TKeHuXiaoShouZZYeWuYSource implements SourceFunction<String> {
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<String> 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:TKeHuXiaoShouZZYeWuY:sendcount", "huazheng:TKeHuXiaoShouZZYeWuY:flagDate", "huazheng:list:TKeHuXiaoShouZZYeWuY", "huazheng:TKeHuXiaoShouZZYeWuY: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:TKeHuXiaoShouZZYeWuY:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
//package com.huazheng.project.greenplum.source.mssql4;
//
//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 cn.hutool.core.thread.ThreadUtil;
//import lombok.extern.log4j.Log4j2;
//
//@Log4j2
//@Service
//public class TasksSource implements SourceFunction<String> {
//
// 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<String> 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:Tasks:sendcount", "huazheng:Tasks:id", "huazheng:list:Tasks", "huazheng:Tasks: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);
// ctx.collect(values[1]);
// } else { // 没有数据字符串
// ThreadUtil.sleep(1000); // 没有数据了,休眠一下
// }
// } catch (Exception e) {
// HZDataStream.redis1Template.opsForHash().put("huazheng:Tasks:error", "receivecount_elseerror", getErrorInfoFromException(e));
// }
// }
// }
//
// @Override
// public void cancel() {
//
// }
//
//}
package com.huazheng.project.greenplum.source.mssql3;
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.mssql3.model.PersonComp;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class PersonCompSource implements SourceFunction<PersonComp> {
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<PersonComp> 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:PersonComp:sendcount", "huazheng:PersonComp:id", "huazheng:list:PersonComp", "huazheng:PersonComp: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);
PersonComp data = JSONUtil.toBean(values[1], PersonComp.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:PersonComp:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
//package com.huazheng.project.greenplum.source.mssql4;
//
//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 cn.hutool.core.thread.ThreadUtil;
//import lombok.extern.log4j.Log4j2;
//
//@Log4j2
//@Service
//public class TasksSource implements SourceFunction<String> {
//
// 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<String> 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:Tasks:sendcount", "huazheng:Tasks:id", "huazheng:list:Tasks", "huazheng:Tasks: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);
// ctx.collect(values[1]);
// } else { // 没有数据字符串
// ThreadUtil.sleep(1000); // 没有数据了,休眠一下
// }
// } catch (Exception e) {
// HZDataStream.redis1Template.opsForHash().put("huazheng:Tasks:error", "receivecount_elseerror", getErrorInfoFromException(e));
// }
// }
// }
//
// @Override
// public void cancel() {
//
// }
//
//}
package com.huazheng.project.greenplum.source.mssql4;
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.mssql4.model.Tasks;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TasksSource implements SourceFunction<Tasks> {
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<Tasks> 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:Tasks:sendcount", "huazheng:Tasks:id", "huazheng:list:Tasks", "huazheng:Tasks: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);
Tasks data = JSONUtil.toBean(values[1], Tasks.class);
ctx.collect(data);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Tasks:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mysql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class HandoverSource implements SourceFunction<String> {
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<String> 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:Handover:sendcount", "huazheng:Handover:handover_id", "huazheng:list:Handover", "huazheng:Handover: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:Handover:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mysql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class HandoverTaskSource implements SourceFunction<String> {
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<String> 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:HandoverTask:sendcount", "huazheng:HandoverTask:task_id", "huazheng:list:HandoverTask", "huazheng:HandoverTask: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:HandoverTask:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mysql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TransformHistoryNodeSource implements SourceFunction<String> {
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<String> 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:TransformHistoryNode:sendcount", "huazheng:TransformHistoryNode:id", "huazheng:list:TransformHistoryNode", "huazheng:TransformHistoryNode: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:TransformHistoryNode:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.greenplum.source.mysql;
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 cn.hutool.core.thread.ThreadUtil;
import lombok.extern.log4j.Log4j2;
@Log4j2
@Service
public class TransformNewNodeSource implements SourceFunction<String> {
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<String> 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:TransformNewNode:sendcount", "huazheng:TransformNewNode:id", "huazheng:list:TransformNewNode", "huazheng:TransformNewNode: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);
ctx.collect(values[1]);
} else { // 没有数据字符串
ThreadUtil.sleep(1000); // 没有数据了,休眠一下
}
} catch (Exception e) {
HZDataStream.redis1Template.opsForHash().put("huazheng:TransformNewNode:error", "receivecount_elseerror", getErrorInfoFromException(e));
}
}
}
@Override
public void cancel() {
}
}
package com.huazheng.project.hana.mapper;
import java.util.List;
import com.huazheng.project.hana.model.Afko;
import com.huazheng.project.hana.model.Afpo;
import com.huazheng.project.hana.model.Aufk;
import com.huazheng.project.hana.model.Aufm;
import com.huazheng.project.hana.model.Bsad;
import com.huazheng.project.hana.model.Bsid;
import com.huazheng.project.hana.model.Kna1;
import com.huazheng.project.hana.model.Knkk;
import com.huazheng.project.hana.model.Knvv;
import com.huazheng.project.hana.model.Konv;
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.Mska;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.S066;
import com.huazheng.project.hana.model.S067;
import com.huazheng.project.hana.model.T001w;
import com.huazheng.project.hana.model.T023t;
import com.huazheng.project.hana.model.Tvkbt;
import com.huazheng.project.hana.model.Vbak;
import com.huazheng.project.hana.model.Vbap;
import com.huazheng.project.hana.model.Vbep;
import com.huazheng.project.hana.model.Vbpa;
import com.huazheng.project.hana.model.Vbrk;
import com.huazheng.project.hana.model.Vbrp;
import com.huazheng.project.hana.model.Vbuk;
import com.huazheng.project.hana.model.Zmdpc;
import com.huazheng.project.hana.model.Zsd06;
import com.huazheng.project.hana.model.Zsdfhzl;
public interface SapMapper {
public List<Vbap> selectVbapNew(Vbap vbap);
public List<Bsid> selectBsidNew(Bsid bsid);
public List<Bsad> selectBsadNew(Bsad bsad);
public List<Pa0002> selectPa0002New(Pa0002 pa0002);
public List<Vbpa> selectVbpaNew(Vbpa vbpa);
public List<Tvkbt> selectTvkbtNew(Tvkbt tvkbt);
public List<Aufm> selectAufmNew(Aufm aufm);
public List<Vbrk> selectVbrkNew(Vbrk vbrk);
public List<Vbrp> selectVbrpNew(Vbrp vbrp);
public List<S066> selectS066New(S066 s066);
public List<S067> selectS067New(S067 s067);
public List<Vbuk> selectVbukNew(Vbuk vbuk);
public List<Likp> selectLikpNew(Likp likp);
public List<Konv> selectKonvNew(Konv konv);
public List<Lips> selectLipsNew(Lips lips);
public List<Afko> selectAfkoNew(Afko afko);
public List<Afpo> selectAfpoNew(Afpo afpo);
public List<Vbep> selectVbepNew(Vbep vbep);
public List<T023t> selectT023tNew(T023t t023t);
public List<Kna1> selectKna1New(Kna1 kna1);
public List<Knvv> selectKnvvNew(Knvv knvv);
public List<Knkk> selectKnkkNew(Knkk knkk);
public List<Mara> selectMaraNew(Mara mara);
public List<Makt> selectMaktNew(Makt makt);
public List<Zsd06> selectZsd06New(Zsd06 zsd06);
public List<Zmdpc> selectZmdpcNew(Zmdpc zmdpc);
public List<Vbak> selectVbakNew(Vbak vbak);
public List<Aufk> selectAufkNew(Aufk aufk);
public List<T001w> selectT001wNew(T001w t001w);
public List<Mska> selectMskaNew(Mska mska);
public List<Zsdfhzl> selectZsdfhzlNew(Zsdfhzl zsdfhzl);
public Afko selectAfkoById(Afko target);
public Afpo selectAfpoById(Afpo target);
public Aufk selectAufkById(Aufk target);
public Aufm selectAufmById(Aufm target);
public Kna1 selectKna1ById(Kna1 target);
public Knkk selectKnkkById(Knkk target);
public Knvv selectKnvvById(Knvv target);
public Konv selectKonvById(Konv target);
public Likp selectLikpById(Likp target);
public Lips selectLipsById(Lips target);
public Makt selectMaktById(Makt target);
public Mara selectMaraById(Mara target);
public Mska selectMskaById(Mska target);
public Pa0002 selectPa0002ById(Pa0002 target);
public S066 selectS066ById(S066 target);
public S067 selectS067ById(S067 target);
public T001w selectT001wById(T001w target);
public T023t selectT023tById(T023t target);
public Vbak selectVbakById(Vbak target);
public Vbep selectVbepById(Vbep target);
public Vbpa selectVbpaById(Vbpa target);
public Vbrk selectVbrkById(Vbrk target);
public Vbrp selectVbrpById(Vbrp target);
public Vbuk selectVbukById(Vbuk target);
public Zmdpc selectZmdpcById(Zmdpc target);
public Zsd06 selectZsd06ById(Zsd06 target);
public Zsdfhzl selectZsdfhzlById(Zsdfhzl target);
public Bsad selectBsadById(Bsad target);
public Bsid selectBsidById(Bsid target);
public Vbap selectVbapById(Vbap target);
public Tvkbt selectTvkbtById(Tvkbt target);
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
/**
* 订单表头数据 PP 订单
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Builder
public class Afko implements Serializable {
private static final long serialVersionUID = 1L;
private String aufnr; // 订单号
/* === 组合键 === */
private String mandt; // 集团
// private String aufnr; // 订单号
/* === 组合键 === */
private String gltrp ; // 基本完成日期
private BigDecimal gamng ; // 订单数量总计
private String plnbez ; // 物料号
private String gmein ; // 基本计量单位
private String plnnr ; // 任务清单组码
private String plnal ; // 组计数器
private String stlnr ; // 物料单
private String cy_seqnr; // 顺序号订单
private String gstrp ; // 订单开始日期
private String aufpl ; // 订单中工序的工艺路线号
private String aplzt ; // 订单的通用计数器
private Date gltrp1 ; // 基本完成日期,转换yyyy-MM-dd
private Date gstrp1 ; // 订单开始日期,转换yyyy-MM-dd
private Vbap vbap;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
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 Afpo implements Serializable {
private static final long serialVersionUID = 1L;
private String kdauf; // 销售订单号
private String kdpos; // 销售订单行项
private String aufnr; // 订单号
/* === 组合键 === */
private String mandt; // 集团
// private String aufnr; // 订单号
private String posnr; // 销售凭证项目
/* === 组合键 === */
private BigDecimal psmng; // 订单项数量
private BigDecimal wemng; // 此订单项的收货数量
private String meins; // 基本计量单位
private String matnr; // 订单的物料编号
private Object uebto; // 过量交货限度
private Object untto; // 交货不足限度
private String pwerk; // 订单的计划工厂
private String verid; // 生产版本
private String dwerk; // 工厂
private String dauat; // 订单类型
private String krsnr; // 预留/相关需求的编号
private String sernr; // BOM展开号
private String plnum; // 计划订单号
private Vbap vbap;
private VbapAdv vbapAdv;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.util.Date;
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 Aufk implements Serializable {
private static final long serialVersionUID = 1L;
private String kdauf; // 销售订单数
private String kdpos; // 销售订单中的项目编号
/* === 组合键 === */
private String mandt; // 集团
private String aufnr; // 订单号
/* === 组合键 === */
private String erdat; // 创建日期
private String erfzeit; // 创建时间
private String ernam; // 输入者
private String aenam; // 最后更改人
private String bukrs; // 公司代码
private String loekz; // 删除标志
private Date erdat1; // 转换 erdat 创建日期yyyy-MM-dd
private Date erdat2; // 转换 erdat 创建日期yyyy-MM-dd hh:mm:ss
private Vbap vbap;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.util.Date;
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 Aufm implements Serializable {
private static final long serialVersionUID = 1L;
private String mblnr; // 物料凭证编号
private String mjahr; // 物料凭证年度
private String zeile; // 物料凭证中的项目
private String bldat; // 凭证中的凭证日期
private String budat; // 凭证中的过帐日期
private String bwart; // 移动类型(库存管理)
private String matnr; // 物料号
private String werks; // 工厂
private String lgort; // 库存地点
private String charg; // 批号
private String sobkz; // 特殊库存标识
private String lifnr; // 供应商帐户号
private String kdauf; // 销售订单数
private String kdpos; // 销售订单中的项目编号
private String shkzg; // 借方/贷方标识
private String menge; // 数量
private String bwtar; // 评估类型
private String aufnr; // 订单号
private Date bldat1; // 凭证中的凭证日期
private Date budat1; // 凭证中的过帐日期
private String mandt; // 集团
private Vbap vbap;
private VbapAdv vbapAdv;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
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 Bsad implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String bukrs; // 公司代码
private String kunnr; // 客户编号
private String umsks; // 特殊总分类帐事务类型
private String umskz; // 特殊总帐标识
private Date augdt; // 清帐日期,转换目标
private String augbl; // 清算单据的单据号码
private String zuonr; // 分配编号
private Integer gjahr; // 会计年度
private String belnr; // 会计凭证编号
private Integer buzei; // 会计凭证中的行项目数
private Date budat; // 过账日期,转换目标
private String blart; // 凭证类型
private String waers; // 货币码
private BigDecimal wrbtr; // 凭证货币金额 13,2
private String shkzg; // 借方/贷方标识
private String zterm; // 付款条件代码
private Date zfbdt; // 用于到期日计算的基准日期,转换目标
private Date dqsjc; // 到期时间:budat > 26号 :zfbdt+2个月,budat < 26号 :zfbdt+1个月
private String budatc; // 过账日期所在的月份,转换出来的
private String budatd; // 根据过账日期判断日是否>26,大于的话,就在过账日期的基础上加一个月,小于等于的话就是不变
// 用于hana到目标库转换,目标库没有此字段
private String augdts; // 清帐日期,源数据
private String budats; // 过账日期,源数据
private String zfbdts; // 用于到期日计算的基准日期,源数据
private String buzeis;
private String pernr; // 人员编码
private String vbeln;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
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 Bsid implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String bukrs; // 公司代码
private String kunnr; // 客户编号
private String umsks; // 特殊总分类帐事务类型
private String umskz; // 特殊总帐标识
private Date augdt; // 清帐日期,转换目标
private String augbl; // 清算单据的单据号码
private String zuonr; // 分配编号
private Integer gjahr; // 会计年度
private String belnr; // 会计凭证编号
private Integer buzei; // 会计凭证中的行项目数
private Date budat; // 过账日期,转换目标
private String blart; // 凭证类型
private String waers; // 货币码
private BigDecimal wrbtr; // 凭证货币金额 13,2
private String shkzg; // 借方/贷方标识
private String zterm; // 付款条件代码
private Date zfbdt; // 用于到期日计算的基准日期,转换目标
private Date dqsjc; // 到期时间:budat > 26号 :zfbdt+2个月,budat < 26号 :zfbdt+1个月
private String budatc; // 过账日期所在的月份,转换出来的
private String budatd; // 根据过账日期判断日是否>26,大于的话,就在过账日期的基础上加一个月,小于等于的话就是不变
// 用于hana到目标库转换,目标库没有此字段
private String augdts; // 清帐日期,源数据
private String budats; // 过账日期,源数据
private String zfbdts; // 用于到期日计算的基准日期,源数据
private String buzeis;
private String pernr; // 人员编码
private String vbeln;
private Long rowids;
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
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 Bsid2Bsad implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String bukrs; // 公司代码
private String kunnr; // 客户编号
private String umsks; // 特殊总分类帐事务类型
private String umskz; // 特殊总帐标识
private Date augdt; // 清帐日期,转换目标
private String augbl; // 清算单据的单据号码
private String zuonr; // 分配编号
private Integer gjahr; // 会计年度
private String belnr; // 会计凭证编号
private Integer buzei; // 会计凭证中的行项目数
private Date budat; // 过账日期,转换目标
private String blart; // 凭证类型
private String waers; // 货币码
private BigDecimal wrbtr; // 凭证货币金额 13,2
private String shkzg; // 借方/贷方标识
private String zterm; // 付款条件代码
private Date zfbdt; // 用于到期日计算的基准日期,转换目标
private Date dqsjc; // 到期时间:budat > 26号 :zfbdt+2个月,budat < 26号 :zfbdt+1个月
private String budatc; // 过账日期所在的月份,转换出来的
private String budatd; // 根据过账日期判断日是否>26,大于的话,就在过账日期的基础上加一个月,小于等于的话就是不变
// 用于hana到目标库转换,目标库没有此字段
private String augdts; // 清帐日期,源数据
private String budats; // 过账日期,源数据
private String zfbdts; // 用于到期日计算的基准日期,源数据
private String vbeln;
private boolean exist; // 用于标记,不是字段
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
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 CAVThisMonth implements Serializable {
private static final long serialVersionUID = 1L;
private String kunnr;
private String waers;
private String budatd; // 过账日期计算
private String budatb; // 过账日期计算上个月,非字段用于查询的结果返回
private BigDecimal byxse; // 当月销售额
private BigDecimal byhk; // 当月回款
private BigDecimal dyjhk; // 当月净回款,非字段用于计算的结果返回
private BigDecimal ljsyysk; // 累积剩余应收款,非字段用于计算的结果返回
private String hashResult;
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
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 Kna1 implements Serializable {
private static final long serialVersionUID = 1L;
private String kunnr; // 客户编号
private String name1; // 名称1
private String name2; // 名称2
private String mandt;
private String ktokd; // 客户帐户组
private Vbap vbap;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
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 Knkk implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String kunnr; // 产品组
private String kkber; // 贷方控制范围
private BigDecimal klimk; // 授信额度
private BigDecimal skfor; // 应收总额 (用于信贷限额检查)
private BigDecimal ssobl; // 检查信贷限额的相关特别债务
private String aedat; // 更新日期
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
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 Knvv implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String kunnr; // 产品组
private String vkorg; // 销售组织
private String vtweg; // 分销渠道
private String spart; // 产品组
private String waers; // 货币
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
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 Konv implements Serializable {
private static final long serialVersionUID = 1L;
private String mandt; // 集团
private String knumv; // 单据条件数
private String kposn; // 条件项目号
private String stunr; // 步骤编号
private String zaehk; // 条件计数器
private String kntyp; // 条件类别( 例如: 税,货运,价格,成本 )
private BigDecimal kbetr; // 价格( 条件金额或百分数 ) 11 2
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
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;
/**
* SD凭证:交货抬头数据
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Builder
public class Likp implements Serializable {
private static final long serialVersionUID = 1L;
private String vbeln; // 交货
/* === 组合键 === */
private String mandt; // 集团
// private String vbeln; // 交货
/* === 组合键 === */
private String erdat ; // 创建日期
private String wadat_ist; // 实际货物移动日期
private String ctlpc ; // 风险类型
private String kunnr ; // 送达方
private Date erdat1 ; // 创建日期
private Date wadat_ist1; // 实际货物移动日期
private String aedat; // 更新日期
private String lfuhr; // 交货时间
@DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss") // 页面写入数据库时格式化
@JSONField(format="yyyy-MM-dd HH:mm:ss") // 数据库导出页面时json格式化
private Date wadat_ist2lfuhr; // 交货时间
private Vbap vbap;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
package com.huazheng.project.hana.model;
import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
/**
* SD凭证:交货:项目数据
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Builder
public class Lips implements Serializable {
private static final long serialVersionUID = 1L;
private String vbeln; // 交货
private String posnr; // 交货项目
private String vgbel; // 参考单据的单据编号
private String vgpos; // 参考项目的项目号
/* === 组合键 === */
private String mandt; // 集团
// private String vbeln; // 交货
// private String posnr; // 交货项目
/* === 组合键 === */
private String matnr; // 物料
private String matkl; // 物料组
private String arktx; // 物料描述
private String werks; // 工厂
private String lgort; // 库存地点
private String charg; // 批次
private BigDecimal lfimg; // 基本计量单位
private Object uebto; // 过量交货限度
private Object untto; // 交货不足限度
private String erdat; // 记录的创建日期
private String erzet; // 输入时间
private Date erdat1; // 记录的创建日期
private Date erdat2; // 记录的创建日期
private String aedat; // 更新日期
private Vbap vbap;
private VbapAdv vbapAdv;
private Long rowids; // sap那边的rowid
private boolean exist; // 用于标记,不是字段
private String hashResult; // 数据hash标记
private String rowNum; // 用于标记,不是字段
}
This source diff could not be displayed because it is too large. You can view the blob instead.
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论