高效数据业务开发工具 esProc SPL

Java

public Map<String, Integer> calculateMaxConsecutiveIncreaseDays(List<StockRecord> stockRecords) {
	Map<String, List<StockRecord>> groupedByCode = stockRecords.stream()
			.collect(Collectors.groupingBy(
					StockRecord::getCode,
					Collectors.collectingAndThen(
							Collectors.toList(),
							list -> {
								list.sort(Comparator.comparing(StockRecord::getDt));
								return list;
							}
					)
			));
	Map<String, Integer> result = new HashMap<>();
	for (Map.Entry<String, List<StockRecord>> entry : groupedByCode.entrySet()) {
		String code = entry.getKey();
		List<StockRecord> records = entry.getValue();
		if (records.isEmpty()) continue;
		Map<Integer, Integer> consecutiveDaysMap = new HashMap<>();
		int cumulativeSum = 0;
		for (int i = 0; i < records.size(); i++) {
			StockRecord current = records.get(i);
			int flag;
			if (i == 0) {
				flag = 1; 
			} else {
				StockRecord prev = records.get(i - 1);
				flag = current.getCl() > prev.getCl() ? 0 : 1;
			}
			cumulativeSum += flag; 
			consecutiveDaysMap.merge(cumulativeSum, 1, Integer::sum);
		}
		int maxDays = consecutiveDaysMap.values().stream()
				.max(Comparator.naturalOrder())
				.orElse(0);

		result.put(code, maxDays);
	}
	return result;
}

SQL

SELECT CODE, MAX(con_rise) AS max_increase_days
FROM (
    SELECT CODE, COUNT(*) AS con_rise
    FROM (
        SELECT CODE, DT,  SUM(updown_flag) OVER (PARTITION BY CODE ORDER BY CODE, DT) AS no_up_days
        FROM (
            SELECT CODE, DT, 
                    CASE WHEN CL > LAG(CL) OVER (PARTITION BY CODE ORDER BY CODE, DT)  THEN 0
                    ELSE 1 END AS updown_flag
            FROM stock
        )
    )
    GROUP BY CODE, no_up_days
)
GROUP BY CODE

Python

Simport pandas as pd
stock_file = "StockRecords.txt"
stock_info = pd.read_csv(stock_file,sep="\t")
stock_info.sort_values(by=['CODE','DT'],inplace=True)
stock_group = stock_info.groupby(by='CODE')
stock_info['label'] = stock_info.groupby('CODE')['CL'].diff().fillna(0).le(0).astype(int).cumsum()
max_increase_days = {}
for code, group in stock_info.groupby('CODE'):
    max_increase_days[code] = group.groupby('label').size().max() – 1
max_rise_df = pd.DataFrame(list(max_increase_days.items()), columns=['CODE', 'max_increase_days'])

SPL

stock.sort(DT).group(CODE;~.group@i(CL< CL[-1]).max(~.len()):mids)

轻量

• JDK1.8 及以上JVM
• 核心部署包不足15M，易分发
• 任何操作系统，包括VM和Container，甚至Android
• 硬件要求低，PC就能跑

与各类Java框架/报表工具配合

标准JDBC调用

报表查询数据准备层

• SPL敏捷计算提升开发效率，避免复杂SQL和存储过程
• 不依赖于数据库的计算能力，代码跨库移植
• 解释执行，天然热切换，报表模块与应用解耦
• 低成本应对没完没了

Java数据逻辑/微服务实现

• 纯Java，和主应用一起打包享受Java成熟框架优势
• 敏捷开发，全面替代Stream/Kotlin/ORM
• 不依赖于数据库的计算能力，代码跨库移植
• 解释执行，热切换，低耦合

存储过程替代

• 纯Java，和主应用一起打包享受Java成熟框架优势，克服存储过程缺点
• 强大过程计算，易于调试，更高开发效率
• 库外计算，不依赖于数据库，天然可移植
• 无须编译存储过程权限，同时避免应用间耦合，提升安全性和可靠性

数据库减负/消灭中间表

• 非关键中间数据移出数据库存入文件，减少数据库存储负担
• 树状目录更易于管理，降低应用间耦合
• 计算任务移出库外实现，减少数据库计算负担
• 文件访问性能更高，运算性能大幅提高

多样数据源混算/T+0实时全量统计

• 丰富数据源支持：RDB,NoSQL,File，HTTP,...；json等多层数据
• 直接计算，无须入库，保持实时性
• 异构数据库跨库计算，生产库+分析库混合计算实现T+0统计
• 不依赖于数据源的计算能力，天然可移植

嵌入/边缘计算引擎

• 小体积全嵌入，可用于边缘计算场景
• 全面计算功能，包括数学类库，大部分任务无须其它组件
• 简单文件存储，无须数据库
• 可接驳远程大型数据源和存储装置

应用外数据清洗准备

• 数据源支持丰富而且一致，方便访问各种无SQL的数据源
• 语言能力强，描述复杂运算比Python更为简捷
• 并行计算，处理大数据时方便性和速度远超Python
• 强集成性，必要时可转移成应用内计算

数据科学家探索分析

• 语言能力强，描述复杂运算比SQL和Python更为简捷
• 比SQL和Python更强的交互性，调试更方便
• 文件存储，数据独立便携，无须数据库就可以分析手边数据
• 并行计算，处理大数据量时方便性和速度远超Python

冷数据的热计算

• 历史冷数据装入数据库占用空间，导致运维复杂化，成本高昂，使用频率却很低，但不装入又不能计算
• 临时装入效率低，装入时间远超计算时间

• SPL直接基于文件计算，无需入库，数据不再“冷” ，低廉存储成本提供热计算