经济普查产品目录审核的分析及优化
**
**
**
**
ePRAS是第一次全国经济普查数据处理软件，它引入了公式的概念并建立了一套自成体系的语言结构，因此它具有第四代语言(4GL)的特征，可以方便地对ePRAS系统的各种元素，比如基层表、目录、分组等进行有效的操作。特别是它引入了自定义函数的功能后，更大大扩展了本身的能力，给应用系统的开发人员提供了广阔的发挥空间，能适用于目前多个统计专业业务工作的需要，也为将来新的统计应用打下了基础。
ePRAS软件的审核关系用公式描述，这次经济普查中，《工业企业产品生产、销售、库存》(规模以上为603表，规模以下为611表，下面以规模以上为例，简称603表)的产品目录审核是其中最复杂的审核公式之一，这个审核公式的执行时间也比一般的公式要长得多。这一方面是统计业务的要求造成的，从这个公式包含的审核关系式的条数就能看出，总共768条主产品和子产品的平衡关系和134条的不同计量单位和主项与其中项之间的关系，这种审核还要在表的8个列上分别进行。每个填报单位的产品个数虽然可能只有几种或者十几种，但是不同单位的产品不同，要在900多条审核关系中找到本单位应该审核的关系，这本身就是一件繁重、耗时的工作。另一方面，虽然这个公式在编写时已经根据产品目录和表的特点采用了相当多的技巧，但是此公式中仍有一些值得改进之处，还有相当大的潜力可挖。本文就此公式的的编写思路进行分析，根据的排序和查找的算法，对该公式进行优化设计。
一、对公式编写思路和执行流程的分析
603表审核公式(以下简称CPML_603)的大体执行流程分为以下三步：
1.对所有审核关系式做预处理,形成一个清单A,该清单可被所有参加审核单位利用；
2.读某一个填报单位的表，对它填报的所有产品通过在清单A中查找、变换等处理，形成该单位要执行的审核关系式的清单B；
3.对清单B中的审核关系式进行审核。然后转步骤2，直到所有填报单位的表审核完成。
（一）主产品和子产品的平衡关系的两种清单的形成过程
主产品和子产品的平衡关系分为三种，分别是主产品的值“等于”、“大于等于”或“大于”若干子产品的值之和。
以前 4组平衡关系为例：
"原煤=无烟煤+烟煤+褐煤"
"2.烟煤=(1)炼焦烟煤+(2)一般烟煤"
"洗精煤=1.炼焦用洗精煤+2.其他用洗精煤"
"1.炼焦用洗精煤>=其中：冶炼用洗精煤"
用产品代码表述则是：
0600003==0610002+0610003+0620002
0610003==0610004+0610005
0610006==0610007+0610009
0610007>=0610008
从上面的关系式可以看出，每组审核关系都包括一种主产品和对应的若干种子产品，主产品和子产品是相对的，如：烟煤 (0610003)在第一组审核关系中是子产品，而在第二组审核关系中是主产品。有时候一种主产品可以有多个审核关系，例如：
1711011>=1711012,
1171101>=1711014+1711015+1711018
我们的目标是，对于某个单位填报的 603表中出现的产品，不论它出现在哪个关系式中，不论它是主产品还是子产品(出现在审核关系式中关系运算符的左侧或右侧)都要进行审核，而且各单位应对每个审核关系对表的每列只进行一次审核。
公式设计者采用了几个一维数组来达到这个目的，一个数组是 rplist,它是用来记录所有关系的产品代码。
var rplist = [0600003,0610002,0600003,0610003,0600003,0620002,
0610003,0610004,0610003,0610005,
0610006,0610007,0610006,0610009,
0610007,0610008, … ]
初看这个数组的元素排列式有些特别，它的第奇数个元素都是某个关系式中的主产品代码，第偶数个元素都是在某个关系式中的子产品代码。一个关系式中有 n个子产品，则数组的元素就有n×2个，包括n个填充在奇数格的同样主产品代码，n个填充在偶数格的不同的子产品代码。
对于重复的主产品代码，公式设计者采用了在原来的代码前加重复次数序号的办法。比如 1711011第一次出现就是1711011，第2次出现就是11711011，以此类推。
另一个数组 repeatPList，以第奇数个元素为重复主产品代码、第偶数个元素为出现次数，记录了重复的产品代码情况。
var repeatPList = [1711011,2,1712002,2,2911003,2,3220003,3,3230008,2,3230018,3,
3230033,3,3230048,2,3230058,3,3230089,2,3230098,2,3230107,2, … ]
将 rplist按照偶数位代码数值增序排序，其对应左侧奇数格代码也跟着改变位置。
通过sortArray( rplist )排序后形成的数组是按照子产品代码从小到大排列的，如果某个子产品在多个关系式中出现，那么这些关系式的主产品代码都被集中在了一起。
在此基础上，下一步的工作就是对每一个单位形成一个检查表 cplist，在表中列出所有应该审核的产品关系式的关系代码(主产品代码)。cplist是一个二维数组，其第一列记录产品代码，用来和审核关系式的产品代码作判断，第二列记录主产品代码在603表中的记录号，方便在数组中读取数据, 如果主产品漏录，就表明这个审核不成立。
依次读取各填报单位的 603表的每一行，就可以取得该单位填报的每一种产品的产品代码，把它们逐个添加到cplist中。为了检查出有些单位只填子产品，而漏填了主产品的错误，最后还要对该单位填写的产品在数组rplist中作一次检索，把尚未列入的主产品补充到cplist中，得出需要检查的审核关系。检索办法是在数组rplist的偶数格检索出有没有这个产品需要检查的审核关系，如果有，则其左侧奇数格代码就是关系式的主产品代码，再在数组cplist中查找该主产品代码，如果没有查到，则将它添加到cplist的末尾。因为rplist排序后每个子产品代码对应的主产品代码都被集中在了一起，所以如果子产品代码有多个要检查的审核关系，那么它的各个主产品代码都可依次轮询得到，当轮询到一个不同子产品代码时，表明涉及前一个子产品的审核关系式已列举完成。
值得注意的是， cplist中不但记录了主产品代码，也记录了子产品代码，如果该子产品代码不是其他审核关系式的主产品代码(即：“纯”子产品代码)，则将来在和审核关系式的主产品代码作比较时必然是找不到的，因而是多余的。
（二）不同计量单位的主项与其中项逻辑关系的两种清单的形成过程
这种审核关系的数量较少，共 134条。数组erplist用来保存需检查的产品代码，用0来分隔不同的审核关系式。
var erplist = [2440003,2440004,0,2672003,2672005,2672006,2672007,0,3050003,3050004,0, … ]
数组 extmsg用来保存同一个审核关系式中第一个产品代码(主项代码)和提示信息。这里利用了ePRAS数组的一个特性，即数组的不同元素可以是不同的数据类型。
var extmsg = [2440003,"1.供儿童乘骑的带轮玩具的不同计量单位辆和千元必须同时出现",
2672003,"清洁类化妆品的其中项已填写，请填写本项." , … ]
数组 epplist用来保存不同计量单位的主项与其中项关系的主项代码，通过对某个单位填报的每一个产品代码在erplist中的检索，在形成前面的检查表cplist的同时，也构造完成了epplist。同样，epplist是一个二维数组，其第一列记录主项代码，第二列记录主项代码在603表中的记录号，这样可以方便地定位，如果主项漏录，就表明这个审核不成立。
(三)对清单中的产品依次进行两种审核关系式的检查
在前面工作的基础上，只要简单地对检查表 cplist进行遍历就可以不重不漏地对603表主产品和子产品的平衡关系进行检查；同样，对检查表epplist进行遍历，就可以不重不漏地对603表不同计量单位、主项与其中项关系进行检查。
通过前面的分析，我们可以看出，公式作者为了保证审核的完备性，提高审核效率，应用了诸多的技巧。试想一下，如果不通过这些技巧提供的审核关系过滤、筛选，对每一个 603表的填报单位直接运用所有的审核关系进行检查，将会有多少无用的数据读取和对具体单位不涉及产品的审核关系执行，将造成多少的时间消耗。
二、数据结构中排序和查找的算法
（一）排序算法
排序算法是一种基本并且常用的算法。由于实际工作中要处理的数据量巨大，所以对排序算法有很高的速度要求。而一般我们所谓的算法的性能主要是指算法的复杂度，一般用O方法来表示。O方法的定义：
若存在一个常量K和起点n 0 ，使当 n>=n 0 时，有 f(n)<=K×g(n),则f(n) = O(g(n))。
我们将按照算法的复杂度，从简单到复杂来分析算法。影响我们算法性能的主要部分是比较和交换，显然，比较和交换次数越多，性能就越差。按照关键字相同的元素在排序前后的相对位置是否改变，排序算法可以分为稳定的和不稳定的。
第一部分是简单排序算法，它们的共同点是算法复杂度为O(N 2 )。
第二部分是高级排序算法，复杂度为O(N×Log 2 (N))或小于O(N 2 )。这里我们只介绍3种算法。另外还有几种算法因为涉及树与堆的概念，这里不再详细讨论。
1、简单排序算法
（ 1）冒泡法
这是最原始，也是众所周知的最慢的算法。根据轻气泡不能在重气泡之下的原则，从下往上扫描数组 R：凡扫描到违反本原则的轻气泡，就使其向上“飘浮”。如此反复进行，直到最后任何两个气泡都是轻者在上，重者在下为止。
（ 2）选择法
这种方法提高了一点性能 (某些情况下)，这种方法类似我们人为的排序习惯：从数据中选择最小的同第一个值交换，再从剩下的部分中选择最小的与第二个交换，这样往复下去。
（ 3）插入法
插入法较为复杂，它的基本工作原理是抽出牌，在前面的牌中寻找相应的位置插入，然后继续下一张。
2、高级排序算法
（ 1）快速排序
快速排序的基本思想是基于分治策略的。将原问题分解为若干个规模更小但结构与原问题相似的子问题。递归地解这些子问题，然后将这些子问题的解组合为原问题的解。首先我们选择一个中间值 middle(程序中我们使用数组中间值)，然后把比它小的放在左边，大的放在右边(具体的实现是从两边找，找到一对后交换)。然后对两边分别使用这个过程。
（ 2）Shell排序(希尔排序)
首先需要一个递减的步长 gap，最后的步长必须是1。工作原理是首先对相隔较远的元素的所有内容排序，然后再使用同样的方法对相隔近些的元素的排序，以此类推。
（ 3）归并排序
把数据等分成两部分 , 各自用归并排序排好后再合并，它在归并时耗费O(n)的空间。
3、对排序算法的总结
(1)若n较小(如n≤40)，可采用插入排序或选择排序。当记录规模较小时，插入排序较好；反之，因为选择移动的记录数少于插人，应选选择排序为宜。(2)若文件初始状态基本有序(指正序)，则应选用插人、冒泡或随机的快速排序为宜；(3)若n较大，则应采用时间复杂度为O(N×Log 2 (N))的排序方法：快速排序、堆排序或归并排序。
快速排序是目前基于比较的内部排序中被认为是最好的方法，当待排序的关键字是随机分布时，快速排序的平均时间最短；快速排序不适合用于“几乎已排好序”或“几乎正好倒序”的数据。在此最坏情况下，时间复杂度为O(N 2 )。堆排序所需的辅助空间少于快速排序，并且不会出现快速排序可能出现的最坏情况。这两种排序都是不稳定的。归并排序是稳定的,而且适用于数据量特别大以至于无法在内存中容纳,需要通过外存来进行的外部排序。
Shell排序的时间复杂度在数学上没有解决，大致可以认为是O(n (1+￡) ), 其中￡是介于0和1之间的常数。
（二）查找算法
1.顺序查找法
即从第一个元素顺序到最后一个元素依次与待查的值进行比较，如果相等，查找成功，否则继续比较，直到所有元素都比较过了，如果还没有匹配的值，查找失败。查找适用于数据量少、不要求已经排序的数据，它的时间复杂度为 O(N)；
2.二分查找
又称折半查找，适用于数据量大、已经排序的数据，它的时间复杂度为 O(Log 2 (N))。
二分查找的基本思想是(设R[low..high]是当前的查找区间) ：
(1)首先确定该区间的中点位置：mid=(low+high)/2
(2)然后将待查的K值与R[mid].key比较：若相等，则查找成功并返回此位置，否则须确定新的查找区间，继续二分查找，具体方法如下：
①若 R[mid].key>K，则由表的有序性可知R[mid..n].keys均大于K，因此若表中存在关键字等于K的结点，则该结点必定是在位置mid左边的子表R[1..mid-1]中，故新的查找区间是左子表R[1..mid-1]。
②类似地，若 R[mid].key
1<k，则要查找的k必在mid的右子表r[mid+1..n]中，即新的查找区间是右子表r[mid+1..n]。下一次查找是针对新的查找区间进行的。 ##="" ###="" (1)="" (1)若n较小(如n≤40)，可采用顺序查找。="" (1)首先查找索引表,索引表是有序表，可采用二分查找或顺序查找，以确定待查的结点在哪一块。="" (2)然后在已确定的块中进行顺序查找,由于块内无序，只能用顺序查找。="" (2)索引表="" (2)若文件初始状态有序，且n较大，则应采用时间复杂度为o(log="" (blocking="" (n))的二分查找方法。(3)="" )="" )了，这里，数组rplist中参加排序的元素有2568个(即偶数格的子产品代码)。="" 2="" 3.分块查找="" 4.对查找算法的总结="" cpml_603公式时可以注意到，无论单位个数多少(大于1)，在进度显示中，从开始执行到单位数变成1，中间大约有20秒(在piii700mhz服务器上测试)的间隔，这个时间基本上是耗费在sortarray(="" do="" for="" function="" i="1" id[l..b]，即：id[i](1≤i≤b)中存放第i块的最大关键字及该块在表r中的起始位置。由于表r是分块有序的，所以索引表是一个递增有序表。="" if="" j="i+1" pid="0;" r[1..n]均分为b块，前b-1块中结点个数为s="n/b，第b块的结点数小于等于s；每一块中的关键字不一定有序，但前一块中的最大关键字必须小于后一块中的最小关键字，即表是“分块有序”的。" r[1..n]开始，每经过一次与当前查找区间的中点位置上的结点关键字的比较，就可确定查找是否成功，不成功则当前的查找区间就缩小一半。这一过程重复直至找到关键字为k的结点，或者直至当前的查找区间为空(即查找失败)时为止。="" rplist="" rplist[i*2]="" search)又称索引顺序查找。它是一种性能介于顺序查找和二分查找之间的查找方法。="" sortarray(="" sortarray的源代码：="" to="" totalrpnum="" totalrpnum-1="" var="" “分块有序”的线性表="" 三、对公式的进一步的优化="" 从="" 分块查找="" 分块查找的基本思想是：="" 分块查找表的存储结构由“分块有序”的线性表和索引表组成。="" 因此，从初始的查找区间="" 执行过="" 抽取各块中的最大关键字及其起始位置构成一个索引表="" 有了上边的结论，回到我们的问题，不难发现公式编写中的一些欠缺之处。我们将逐个分析它们，并提出改进的方法，以使公式的执行效率达到更优。="" 若文件初始状态分块有序="" 表="" （一）排序算法的选择="" ，且n较大，则应采用分块查找。="">rplist[j*2] then
2
3pid = rplist[j*2];
4
5rplist[j*2] = rplist[i*2];
6
7rplist[i*2] = pid;
8
9endif
10
11endfor
12
13endfor
14
15end
16
17可以看出，这是一种选择排序，时间复杂度为 O(N 2 )，比较次数为N×N /2，如果把它改成时间复杂度为O(N×Log 2 (N))的排序，将可以缩短排序时间为原来的大约Log 2 (N)/N×2，即大约9/1000。采用递归算法的快速排序或采用Shell排序来代替原来的sortArray()函数，经过测试，大约1秒都可以完成排序。
18
19根据总运行时间函数 f(u)=t 0 +t 1 u，其中u是参加审核的单位个数，t 0 是固定的，也就是开始处理每个单位前都要用的时间，实际上主要是 rplist排序和系统对数据表建立索引的时间。下面的主要工作就是如何减少每个单位处理的时间t 1 ，达到缩短总时间的目的。
20
21### （二）查找算法的选择
22
23每个单位处理的时间由二部分组成 ,一部分是形成该单位要执行的审核关系式的清单,另一部分是对清单中的审核关系式进行审核。我们要从这二方面着手减少处理的时间。
24
25findFirstRProdIndex( rplist,cplist[i,1] )函数用来在rplist中找到第一个出现的cplist[i,1]，已知元素有2568个，而且已经排好序，符合二分查找的条件，把它改为二分查找，将可以缩短查找时间为原来的大约Log 2 (N)/N×2。但是二分查找无法实现对“第一个”出现的元素的查找，为此新增了数组unicodelist用来存放单个子产品代码以及它第一次出现在rplist的位置，通过对unicodelist的二分查找得到它在rplist的位置，由于rplist已排好序，所以从中顺序挑选子产品代码存入unicodelist，unicodelist自然已经排好序了。
26
27尽管如此我们作了这些额外的事，经过测试，此项改动在单位个数为 3700，大约可以使总执行时间由180秒(已改用Shell排序)进一步缩短为130秒完成。
28
29getERProdid(prodid, erplist )函数用于查找主产品或者副产品，如果找到是副产品，返回其主产品，erplist有 440个元素，改用二分法也将有不少的收益。但原来erplist 存放需要检查计量单位同时存在的关系的产品号，不同的关系用 0分隔，排序后就失去了分隔标志，需要用一个新代码来代替，这个新代码的值应位于上一种关系最后一个产品和下一种关系第一个产品代码之间，才能在排序后保持原有的相对位置。考虑到有的上一种关系最后一个产品和下一种关系第一个产品代码是连续的，比如：3050003, 3050004, 0, 3050005, 3050006, 0。在这里，用下一种关系第一个产品代码来替代0分隔是很合适的，经过替换以后，erplist已经有序了。查找的时候，只要判断两个连续的代码是相同的，就可以确定它是主产品代码。
30
31取得提示信息的函数有 2个，它们目前也都是采用顺序查找，理论上也可以用排序后二分查找的办法提高效率，但经过实验发现，这些函数总的执行次数(时间)很少，可以忽略不计，这可能与我们采用的样本数据只有很少的错误有关。ePRAS公式的设计者在设计verify expr,ErrorMsg()语句的时候，采用了类似C语言多个条件复合表达式的短路技术(比如复合表达式(true or expr2)不管子表达式expr2的值是什么，总是返回true)，如果知道前面expr的值为true，就不再调用后面ErrorMsg()函数了。
32
33如果原始数据质量很差，错误很多，那么下面这些语句的优化还是值得的。
34
35getErrorMsg(prodid, plist, pmsg )
36
37getEXTMsg(prodid, mesg )
38
39除了在输出错误信息时对数组 pmsg或extmsg二分查找，也可以在形成清单的过程中，在数组记下各主产品的行号rn，直接引用pmsg[rn]或extmsg[rn*2]，避免二次查找。
40
41### （三）大量 case 语句的拆分 <span lang="EN-US" st</k，则要查找的k必在mid的右子表r[mid+1..n]中，即新的查找区间是右子表r[mid+1..n]。下一次查找是针对新的查找区间进行的。>