Quartz + Tablesaw 报表统计

Posted 2020-10-21 devywb

场景

在12 月份做的报表功能中，直接从 ES 查询一个月的数据。当数据量特别大时，查询速度会非常缓慢甚至查询失败。解决方案是使用定时任务，在每天凌晨指定时间自动查询前一天的数据，然后写入 CSV 文件中，每天追加。生成报表文件时，就不用再查询 ES，而是读取 CSV 文件，统计一个月每天数据的总和。

一、定时任务

定时任务使用的是 Quartz 框架。

Quartz 是什么

Quartz 是一个开源的作业调度框架,由 java 编写,在.NET 平台为 Quartz.Net,通过 Quart 可以快速完成任务调度的工作。

Quartz 使用场景

如定时发送邮件、定时统计数据生成报表等等

在项目中使用 Quartz

添加依赖

<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.3.0</version>
</dependency>

spring-quartz.xml 配置文件

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

<beans>
    <!-- 要调用的工作类 -->
    <bean id="quartzJob" class="com.devywb.quartzJob"></bean>
    <!-- 定义调用的对象和方法 -->
    <bean id="jobTask" class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
        <!-- 调用的类 -->
        <property name="targetObject">
            <ref bean="quartzJob"></ref>
        </property>
        <!-- 调用的方法 -->
        <property name="targetMethod">
            <value>work</value>
        </property>
    </bean>
    
    <!-- 定义触发的时间 -->
    <bean id="doTime" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
      <property name="jobDetail">
                <ref bean="jobTask"/>
      </property>
      <!-- cron表达式 -->
      <property name="cronExpression">
        <!-- 每隔20秒钟执行一次 -->
        <!-- <value>*/20 * * * * ?</value> -->
        <!-- 每天凌晨五点执行 -->
        <value>0 0 5 * * ?</value>
      </property>
    </bean>
    
     <!-- 总管理类 -->
     <bean id="startQuertz" lazy-init="false" autowire="no" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
         <property name="triggers">
             <list>
                 <ref bean="doTime"/>
             </list>
         </property>
     </bean>
</beans>

二、Tablesaw

Tablesaw是一套内存内数据表，其中包含多种数据工具与面向列的存储格式。其设计思路认为没人会面向小型任务执行分布式分析，而大家可以在单一服务器上对200万行级别的表进行交互。
大家能够利用Tablesaw执行各种规则，从而检查显示布局、数据优先级或者针对数据显示及交互向特定用户提供扩展控制范围。在它的帮助下，我们可以利用RDBMS与CSV文件导入数据，添加及删除列，执行映射与规约操作或者将表保存在经过压缩的列式存储格式当中。

添加依赖

<!-- https://mvnrepository.com/artifact/tech.tablesaw/tablesaw-core-->
<dependency>
    <groupId>tech.tablesaw</groupId>
    <artifactId>tablesaw-core</artifactId>
    <version>0.11.2</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-math3</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.0</version>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-math3 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-math3</artifactId>
    <version>3.0</version>
</dependency>

在项目中添加 Tablesaw 的依赖后，项目无法启动，原因是 commons-lang3 和 commons-math3 两个依赖的版本太高了。后来采用的解决方法是排除依赖再另外引入低版本依赖。

常用 API

// 读取 csv 文件
Table table = Table.read().csv(String filePath);

// 获取所有列名
List<String> columnNames = table.columnNames();

// sum
table.sum("列名").get();

// 排序
table.sortDescendingOn("列名"); // 降序
table.sortAscendingOn("列名"); // 升序

// 分组
table.groupBy(columns);

// 前多少条
table.first(nRows)

// 生成 saw 文件
Table table = Table.read().csv(contents, tableName);
table.save(folder);

// 读取 saw 文件
Table table = Table.readTable(tableNameAndPath)

踩过的坑

写入数据至 CSV 文件时，当数据中某个字段包含分隔符或其他特殊符号时，程序会报错。下面列出两种解决方案。

第一种，使用第三方库，例如 openCSV，它底层对特殊符号做了处理；

第二种，手动处理字段中的特殊符号；

/**
 * list 转 CSV 字符串
 * @param header
 * @param data
 * @return
 */
public static String list2CsvStr(Map<String, String> header, List<Map<String, Object>> data) {
    String content = "";
    if (header != null && header.size() > 0) {
        if (data != null && data.size() > 0) {
            StringBuilder sb = new StringBuilder();
            for (Map<String, Object> map : data) {
                Set<Entry<String, String>> entrySet = header.entrySet();
                int i = 0;
                for (Entry<String, String> entry : entrySet) {
                    String key = entry.getKey();
                    if(i > 0) sb.append(SEPARATOR); // 不是第一列时，添加分隔符
                    if(map.containsKey(key)) {
                        Object value = map.get(key);
                        String valueStr = value != null ? value.toString() : "";
                        if(valueStr.contains(SEPARATOR)) { // 如果数据包含分割符
                            if(valueStr.contains("\"")) { // 如果字段中包含双引号，替换成两个
                                valueStr.replace("\"", "\"\"");
                            }
                            // 如果字段包含分割符，则用双引号括起来
                            valueStr = "\"" + valueStr +"\"";
                        }
                        sb.append(valueStr);
                    } else {
                        sb.append("");
                    }
                    i ++ ;
                }
                sb.append("\n");
            }
            content = sb.toString();
        }
    }
    return content;
}

/**
 * map 转 CSV 字符串
 * @param header
 * @param data
 * @return
 */
public static String map2CsvStr(Map<String, String> header, Map<String, Object> data) {
    String content = "";
    if (header != null && header.size() > 0) {
        if (data != null && data.size() > 0) {
            StringBuilder sb = new StringBuilder();
            int i = 0;
            Set<Entry<String, String>> entrySet = header.entrySet();
            for (Entry<String, String> entry : entrySet) {
                String key = entry.getKey();
                if(i > 0) sb.append(SEPARATOR);
                if(data.containsKey(key)) {
                    Object value = data.get(key);
                    String valueStr = value != null ? value.toString() : "";
                    if(valueStr.contains(SEPARATOR)) { // 如果数据包含分割符
                        if(valueStr.contains("\"")) { // 如果字段中包含双引号，替换成两个
                            valueStr.replace("\"", "\"\"");
                        }
                        // 如果字段包含分割符，则用双引号括起来
                        valueStr = "\"" + valueStr +"\"";
                    }
                    sb.append(valueStr);
                } else {
                    sb.append("");
                }
                i ++;
            }
            sb.append("\n");
            content = sb.toString();
        }
    }
    return content;
}