集算器实现外部数据并行计算

一时失言乱红尘 2022-01-12 15:57 110阅读 0赞

##   文本并行 ##

SPL可将文本文件按体积大致分为N段，只读取其中一段。比如cardInfo.txt存储着一千万条人口信息，将其分为十份，取第二份，代码可以写作：

按体积大致分段，而不是按行数精确分段，目的是提高分段性能。比如在IDE中观察A2或A3的前几个字段，可以看到行数并非精确的100万（与具体数据有关）：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">index</td> 
   <td style="vertical-align:top;">cardNo</td> 
   <td style="vertical-align:top;">name</td> 
   <td style="vertical-align:top;">gender</td> 
   <td style="vertical-align:top;">province</td> 
   <td style="vertical-align:top;">mobile</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">1</td> 
   <td style="vertical-align:top;">308200310180525</td> 
   <td style="vertical-align:top;">Alison Clinton</td> 
   <td style="vertical-align:top;">female</td> 
   <td style="vertical-align:top;">Idaho&nbsp;</td> 
   <td style="vertical-align:top;">1024627490</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">2</td> 
   <td style="vertical-align:top;">709198311300191</td> 
   <td style="vertical-align:top;">Abby Wood</td> 
   <td style="vertical-align:top;">female</td> 
   <td style="vertical-align:top;">Kansas&nbsp;</td> 
   <td style="vertical-align:top;">19668466</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">3</td> 
   <td style="vertical-align:top;">1005199807060610</td> 
   <td style="vertical-align:top;">George Bush</td> 
   <td style="vertical-align:top;">male</td> 
   <td style="vertical-align:top;">California&nbsp;</td> 
   <td style="vertical-align:top;">1019879226</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">…</td> 
   <td style="vertical-align:top;">…</td> 
   <td style="vertical-align:top;">…</td> 
   <td style="vertical-align:top;">…</td> 
   <td style="vertical-align:top;">…</td> 
   <td style="vertical-align:top;">…</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">1000005</td> 
   <td style="vertical-align:top;">405199907050256</td> 
   <td style="vertical-align:top;">Mark Rowswell</td> 
   <td style="vertical-align:top;">male</td> 
   <td style="vertical-align:top;">Idaho&nbsp;</td> 
   <td style="vertical-align:top;">1168620176</td> 
  </tr> 
 </tbody> 
</table>

分段读取可应用于多线程计算，从而提高读取性能。比如用2个线程分别读取cardInfo.txt，各线程计算本段行数，最后合并为总行数，可用如下代码：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">5</td> 
   <td style="vertical-align:top;">fork to(2)</td> 
   <td style="vertical-align:top;">=A1.cursor@t(;A5:2).total(count(1))</td> 
   <td style="vertical-align:top;">/2线程分段</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">6</td> 
   <td style="vertical-align:top;">=A5.sum()</td> 
   <td style="vertical-align:top;"><br></td> 
   <td style="vertical-align:top;">/合并结果</td> 
  </tr> 
 </tbody> 
</table>

语句fork语句适合算法较复杂的情况，当算法比较简单时，可用cursor@m直接分段读取。比如前面的代码可以改写如下：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">7</td> 
   <td style="vertical-align:top;">=A1.cursor@tm(;2).total(count(1))</td> 
   <td style="vertical-align:top;">/2线程分段</td> 
  </tr> 
 </tbody> 
</table>

上述代码指定了线程数，如果省略线程数，则用配置文件中的“parallet limit”当做默认线程数。假设parallet limit=2，则上述代码可以改写成：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">8</td> 
   <td style="vertical-align:top;">=A1.cursor@tm().total(count(1))</td> 
   <td style="vertical-align:top;">/默认线程数分段</td> 
  </tr> 
 </tbody> 
</table>

为了验证分段读取前后的性能差异，下面设计一个算法，分别用单线程和2线程计算cardInfo.txt的总行数，可以看到性能显著提升：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">11</td> 
   <td style="vertical-align:top;">=now()</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">12</td> 
   <td style="vertical-align:top;">=A1.cursor@t().total(count(1))</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">13</td> 
   <td style="vertical-align:top;">=interval@ms(A11,now())</td> 
   <td style="vertical-align:top;">/未分段，20882ms</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">14</td> 
   <td style="vertical-align:top;"><br></td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">15</td> 
   <td style="vertical-align:top;">=now()</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">16</td> 
   <td style="vertical-align:top;">=A1.cursor@tm(;2).total(count(1))</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">17</td> 
   <td style="vertical-align:top;">=interval@ms(A15,now())</td> 
   <td style="vertical-align:top;">/2线程分段，12217ms</td> 
  </tr> 
 </tbody> 
</table>

##   JDBC 并行 ##

通过JDBC取数时，有时会遇到数据库负载虽然不重，但取数性能仍然较差的情况，这种情况下可以用并行取数提高性能。

比如Oracle数据库有一张通话记录表callrecord，记录数100万条，索引字段是callTime，且数据基本按该字段平均分布。采用非并行取数时，可以发现性能不够理想，代码如下：

改为2线程并行取数后，可以看到性能提升明显，代码如下：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">6</td> 
   <td style="vertical-align:top;">=now()</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">7</td> 
   <td colspan="2" style="vertical-align:top;">=connect("orcl").query@x("select &nbsp; min(callTime),max(callTime) from callrecordA")</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">8</td> 
   <td style="vertical-align:top;">=2.(range(A7.#1,elapse@s(A7.#2,1),~:2))</td> 
   <td style="vertical-align:top;">/时间区间参数列表</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">9</td> 
   <td style="vertical-align:top;">fork A8</td> 
   <td style="vertical-align:top;">=connect("orcl")</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">10</td> 
   <td style="vertical-align:top;"><br></td> 
   <td style="vertical-align:top;">=B9.query@x("select * from &nbsp; callrecordA where callTime&gt;=? and callTime&lt;?",A9(1),A9(2))</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">11</td> 
   <td style="vertical-align:top;">=A9.conj()</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">12</td> 
   <td style="vertical-align:top;">=interval@ms(A6,now())</td> 
   <td style="vertical-align:top;">/并行取数，10045ms</td> 
  </tr> 
 </tbody> 
</table>

既然要并行取数，就要把源数据分成多个区间，使每区间的数据量大致相等。在这个例子中，索引字段是时间类型callTime，所以先用A7求出callTime的数据范围，再用A8将该范围平均分成2个时间区间。之后在A9进行并行计算，每个线程以各自的时间区间为参数执行SQL，取数结果将大致相等。最后合并多线程的取数结果，作为最终结果。

函数range非常适合对数据分段。该函数可将某范围平均分为N个区间，获得第i个区间，且可根据范围的数据类型自动调整区间的数据类型。本例的范围类型是datetime，则函数range将范围按秒均分，返回类型也是datetime。如果范围类型是date，则函数range按天均分；如果范围类型是整数，则函数range按整数均分。

上面例子中，分段字段是索引，如果没有建立索引，则查询性能会出现下降。在这种情况下，并行取数仍然可以带来明显的性能提升，所以可以用相同的方法。

上面例子中，源数据基本按callTime平均分布，因此容易使各区间的数据量大致相等，如果源数据分布很不平均，可以考虑按行号分段。每种数据库都有生成行号的方法，比如oralce可用rownum。

除了单表单SQL并行取数，SPL也支持多表多SQL并行取数。比如某报表格式较复杂，需要SPL执行多个SQL，并按一定的格式拼出结果集。当采用非并行取数时，可以发现性能不够理想，代码如下：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;"><br></td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">A</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">B</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">1</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=now()</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=connect("orcl")</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">2</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyy')='2015'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A2)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">3</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201501'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A3)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">4</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201502'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A4)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">5</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201503'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A5)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">6</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyy')='2016'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A6)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">7</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201601'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A7)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">8</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201602'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A8)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">9</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">select count(1)&nbsp; from&nbsp; &nbsp; callrecordA where to_char(calltime,'yyyyMM')='201603'</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.query(A9)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">10</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);"><br></td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=B1.close()</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">11</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=[B2:B9].new(~.#1:data)</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);text-align:center;font-size:16px;">12</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">=interval@ms(A1,now())</td> 
   <td style="vertical-align:top;border:1px solid rgb(212,212,212);">/非并行取数，2195毫秒</td> 
  </tr> 
 </tbody> 
</table>

改为4线程并行取数后，可以看到性能提升明显，代码如下：

<table> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;">14</td> 
   <td style="vertical-align:top;">=now()</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">15</td> 
   <td style="vertical-align:top;">fork [A2:A9]</td> 
   <td style="vertical-align:top;">=connect("orcl")</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">16</td> 
   <td style="vertical-align:top;"><br></td> 
   <td style="vertical-align:top;">=B15.query@x(A15)</td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">17</td> 
   <td style="vertical-align:top;">=A15.new(~.#1:data)</td> 
   <td style="vertical-align:top;"><br></td> 
  </tr> 
  <tr> 
   <td style="vertical-align:top;">18</td> 
   <td style="vertical-align:top;">=interval@ms(A14,now())</td> 
   <td style="vertical-align:top;">/4并行取数,1320毫秒</td> 
  </tr> 
 </tbody> 
</table>

需要注意的是，并行取数时任务数可大于并行数。比如上面代码共8个任务，但同时执行的任务只有4个，其他待执行的任务排在队列中，如果某个小任务先执行完成，SPL会从队列中取下一个任务并执行它。可以看到，当任务数较多时，即使各任务负载相差较大，也能充分发挥硬件性能。

##   混合并行 ##

当数据量太大时，除了分库计算，还可以进行混合数据源并行计算，后者性能更高。具体做法是：把数据分为两部分（或多部分），一部分存储在数据库中，通常是当前实时数据，一部分存储在组文件，通常是历史数据，再对两种数据源进行并行计算，从而获得更高性能。

比如历史订单存储在orders.ctx中，当前订单存储在数据库orcl中，请按年、月分组，对各组数据的amount字段求和。SPL代码如下：

注意fork……fork……的用法。如果fork语句块下接非fork语句块，则两者顺序执行，如果fork语句块下接fork语句块，则两者并行执行。

转载于:https://blog.51cto.com/12749034/2394415