1、前言
??變步長批量處理演算法，在實作某些功能時，非常需要，如資料傳輸和資料匯入時，使用可變步長的批量處理演算法，可以極大地提高系統的性能，，
1.1、資料傳輸
??在不穩定的網路環境下，傳輸失敗的幾率提高，大的資料塊可能會傳輸失敗，如果分為小的資料塊，可以傳輸成功，但由于傳輸開銷，傳輸效率低，因此希望在網路好的時候，傳輸大的資料塊或高解析度的影像；網路差的時候，傳輸小的資料塊或對影像進行降質處理，調低影像解析度，提高壓縮比等，因此，可變概念，可以提升服務功能的質量，提升系統的可用性，
1.2、資料匯入
??資料匯入，特別是ETL，大量資料匯入，效率非常重要，對于大多數資料庫，如Myql，批量新增（如100條記錄）和單記錄新增的時間消耗相差無幾，但處理能力有百倍之差，
??曾經，筆者使用逐條資料insert，300萬條記錄匯入，化了半個小時，簡直無法忍受，于是，后來改為使用100條批量insert，但由于資料中存在這種那種的例外資料，經常出現一條例外資料導致成批(100條)的資料匯入失敗，這樣，一次匯入資料中可能有幾潭訓資料，導致幾百上千條資料沒有入庫成功，于是，再修改代碼，針對這些沒入庫成功的幾百上千條記錄里，逐潭訓入，檢測出具體的壞資料，整個程序不堪回首，
??因此，資料匯入需要可變步長演算法，這樣可用極大提升資料匯入的處理能力，
??另外，還有一種Excel資料匯入，如規定按記錄的編碼（字串型別，如身份證號、手機號、訂單編號等，唯一鍵欄位）作為記錄的特征欄位，但表格資料中有新增的，還有修改的，即如果為新的編碼值，為新增記錄，如果資料庫中編碼值已存在，則需要修改記錄，這種匯入，如果采用固定批量值匯入，新增的失敗率是很高的，如果批量一旦失敗就改為逐潭訓入的策略，也是效率不高，可變步長演算法可非常高效地解決此問題，
1.3、其它應用場景
??對這種可變的需求，具體很大的通用性，如與搜索相關的，也可以用可變步長演算法來提升處理性能，
2、演算法原理
2.1、演算法概述
??可變步長演算法，不是個新鮮的概念，區別在于變化的依據和策略，如最大似然估計、梯度下降等，以及演算法復雜度和是否簡單易用，
??本演算法可以歸于簡單的決策樹，有貪婪演算法的因子，計算量很少，介面呼叫可以內嵌方法（類似C++的指標函式）來執行批量處理作業和單條資料修正處理作業，二分法是非常經典的演算法，本演算法的核心還是二分法，也可以說依據下列公式而開發：
\[1 + 2 + 4 + ... + 2^n = 2^{n+1}-1 \]??使用QoS（Quality of Service，服務等級）的概念，將處理等級與處理能力（批量值）建立聯系，QoS等級對應的批量值為\(2^0\)到\(2^n\)的一個連續序列，如：[128,64,32,16,8,4,2,1]，上限為\(2^n\)，這里n=7，下限為約定為1，等級0對應128，等級7對應1，等級值即為等級陣列的下標，
??具體演算法如下：
- 對于長度為n的輸入資料串列，型別為泛型型別T，即資料為List
，另外提供2個T型別的串列引數，為批量處理成功的資料串列和修正處理成功的資料串列，便于外部進一步處理（如相關資料一致性處理）， - 演算法回傳處理例外記錄的日志串列，字串型別，使用者可以調整日志的格式和內容，以便定位例外資料的具體位置，
- 設定一個布爾型的例外標志值bError，用于記錄之前是否發生了批量處理的例外，初值為false，該例外標志值在單條記錄的修正處理后被復位為false，
- 設定一個資料串列下標錨點anchorIdx，表示當前正在處理的資料位置，初始為0，
- 設定一個當前等級值levelIdx，初值隨意，這里設定初值為0，即從最高等級開始，
- 如果bError為false，即當前無例外，按照當前等級對應的批量值進行批量處理，如果成功，則提升一個等級（如果已為等級上限，則維持），并更新錨點anchorIdx到新的位置；如果處理失敗，則設定bError為true，錨點anchorIdx不變，如當前等級不為等級下限，則下降一個等級；如果已是等級下限，則按照單條資料的修正處理方法進行處理，如果修正成功，則加入修正資料串列中，如果修正失敗，則加入例外日志串列中，不管修正成功與否，修正處理后，anchorIdx均加1，且設定bError為false，表示例外資料已給檢測出并按照規則進行處置了，
- 如果bError為true，即當前處于例外狀態，按照當前等級對應的批量值進行批量處理，如果成功，則下降一個等級（如果已為等級下限，則維持），并更新錨點anchorIdx到新的位置；如果失敗，且不為等級下限值，則錨點anchorIdx不變，下降一個等級；如果失敗，且當前等級為等級下限值，則按照單條資料的修正處理方法進行處理，如果修正成功，則加入修正資料串列中，如果修正失敗，則加入例外日志串列中，不管修正成功與否，修正處理后，anchorIdx均加1，且設定bError為false，
- 結束條件，anchorIdx到達n，即所有資料被處理完畢，
??這里有兩個外部方法，批量處理方法和單條修正方法，這是一個非常抽象的介面方法，具體對資料進行怎么處理，由外部根據需要自行確定，后面提供的單元測驗代碼，給出批量資料型別轉換的例子，
??當bError為true時，批量處理成功，仍然要降級，是因為前面等級發生例外時，資料范圍包括了后面等級批量值之和的范圍：
\[2^{n+1}>2^n+...+2+1 \]??當等級批量值\(2^{n+1}\)批量嘗試發生例外時，設定bError為true，然后如果\(2^n\)批量處理成功，意味著后面\(2^n\)批量必然失敗，因此必須降級，才可能避免失敗；當然如果前\(2^n\)批量處理失敗，意味著例外資料至少在這些資料中存在，也需要降級嘗試，才可能避免失敗，
2.2、演算法代碼
??演算法代碼使用java語言，很容易轉成python或其它語言，代碼在github上：https://github.com/alabo1999/framework_algoset/commit/dacefcbc9bf2ad95c05ab7671a12054178e7f90e 包含一個演算法類檔案BatchProcess.java和單元測驗檔案BatchProcessTest.java，其中BatchProcess.java只有一個方法，該方法的介面形式如下：
/**
* @methodName : varStepBatchProcess
* @description : 可變步長的批量處理演算法
* @param <T> : 泛型型別
* @param object : 提供batchProcMethod和singleProcMethod方法的類物件
* @param dataRowList : 待處理的T型別物件資料串列
* @param normalList : 正常處理的物件串列
* @param correctList : 修改處理的物件串列
* @param batchProcMethod : 正常批量處理的方法
* @param singleProcMethod : 單條修正處理的方法
* @param debugLevel : 除錯資訊輸出設定，bit0-輸出修正處理資訊，bit1-輸出詳細步驟,bit2:輸出嘗試次數
* @return : 處理程序產生的例外日志串列
*/
public static <T> List<String> varStepBatchProcess(
Object object,
List<T> dataRowList,
List<T> normalList,
List<T> correctList,
Method batchProcMethod,
Method singleProcMethod,
int debugLevel);
2.3、演算法測驗
??單元測驗檔案BatchProcessTest.java，給出了演算法測驗，代碼如下：
package com.abc.example.service;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.transaction.annotation.Transactional;
import com.abc.example.common.impexp.BatchProcess;
/**
* @className : BatchProcessTest
* @description : 批量處理測驗類
* @summary :
* @history :
* ------------------------------------------------------------------------------
* date version modifier remarks
* ------------------------------------------------------------------------------
* 2022/01/20 1.0.0 sheng.zheng 初版
*
*/
@RunWith(SpringRunner.class)
@SpringBootTest
@Transactional
public class BatchProcessTest {
@Test
// 可變步長的批量處理演算法測驗
public void varStepBatchProcessTest() {
// 構造待處理的資料，資料型別為String
List<String> dataRowList = new ArrayList<String>();
int idx = 0;
for (int i = 0; i < 1000; i++) {
String str = "";
if (i % 129 == 0) {
str = i + ".1";
idx ++;
if (idx % 2 == 0) {
str += "abc";
}
}else {
str = i + "";
}
dataRowList.add(str);
}
System.out.println(dataRowList);
// 呼叫演算法
Method method1 = getMethodByName(this,"batchProcMethod");
Method method2 = getMethodByName(this,"singleProcMethod");
// 用于存放正常批量處理的資料
List<String> normalList = new ArrayList<String>();
// 用于存放修正處理的資料
List<String> correctList = new ArrayList<String>();
// 呼叫演算法
List<String> errorList = BatchProcess.varStepBatchProcess(this, dataRowList,
normalList, correctList, method1, method2,0x05);
// 列印errorList
System.out.println("errorList: " + errorList.toString());
// 列印correctList
System.out.println("correctList: " + correctList.toString());
}
// 構造批量處理的方法
// 將串列中字串，批量轉為整型，被反射呼叫，必須是public的
public void batchProcMethod(List<String> subDataList) {
for (String item : subDataList) {
Integer.valueOf(item);
}
}
// 構造單記錄處理的方法，被反射呼叫，必須是public的
public String singleProcMethod(Exception e,String item) {
String errInfo = "";
try {
Double.valueOf(item).intValue();
}catch(Exception ex) {
errInfo = ex.getMessage();
}
return errInfo;
}
// 根據方法名稱獲取方法物件
private Method getMethodByName(Object object,String methodName) {
Class<?> class1 = object.getClass();
Method retItem = null;
Method[] methods = class1.getMethods();
for (int i = 0; i < methods.length; i++) {
Method item = methods[i];
// System.out.println(item.getName());
if (item.getName() == methodName) {
retItem = item;
break;
}
}
return retItem;
}
}
??構造了1000個字串的串列，這里使用String來作為資料型別，除了下標為129的整數倍之外的值=i+""的字串，其它的取值，如果為129的偶數倍，則為i+"0.1"的字串；如果為129的奇數倍，則為i+”0.1abc“，
??批量處理方法batchProcMethod，實作對當前批量資料的型別轉換，將字串轉為整數，當資料串列包含.1或.1abc的例外資料時，Integer.valueOf(item);會拋出例外，導致該批量處理失敗；
??單資料修正方法singleProcMethod，實作了處理"0.1"之類的double型字串轉換成，但針對".1abc"后綴的資料仍然處理失敗，
??執行測驗代碼，輸出結果如下：
single tryNum = 9, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 0,bError=true
0.1: can be fixed.
single tryNum = 24, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 129,bError=true
129.1abc: can not be fixed.
single tryNum = 39, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 258,bError=true
258.1: can be fixed.
single tryNum = 54, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 387,bError=true
387.1abc: can not be fixed.
single tryNum = 69, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 516,bError=true
516.1: can be fixed.
single tryNum = 84, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 645,bError=true
645.1abc: can not be fixed.
single tryNum = 99, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 774,bError=true
774.1: can be fixed.
single tryNum = 114, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 903,bError=true
903.1abc: can not be fixed.
tryNum: 120
errorList: [For input string: "129.1abc", For input string: "387.1abc", For input string: "645.1abc", For input string: "903.1abc"]
correctList: [0.1, 258.1, 516.1, 774.1]
??tryNum: 120，表示這個演算法總共嘗試了120次處理嘗試，包括批量處理和修正處理，需要想要看每一次處理嘗試的詳細資訊，將debugLevel設定為7，可以看到詳細的輸出（摘錄部分）：
batch tryNum = 1, levelIdx = 0,levelNum = 128,batchNum = 128,anchorIdx = 0,bError=false
java.lang.reflect.InvocationTargetException
batch tryNum = 2, levelIdx = 1,levelNum = 64,batchNum = 64,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 3, levelIdx = 2,levelNum = 32,batchNum = 32,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 4, levelIdx = 3,levelNum = 16,batchNum = 16,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 5, levelIdx = 4,levelNum = 8,batchNum = 8,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 6, levelIdx = 5,levelNum = 4,batchNum = 4,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 7, levelIdx = 6,levelNum = 2,batchNum = 2,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 8, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 0,bError=true
java.lang.reflect.InvocationTargetException
single tryNum = 9, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 0,bError=true
0.1: can be fixed.
batch tryNum = 10, levelIdx = 6,levelNum = 2,batchNum = 2,anchorIdx = 1,bError=false
batch tryNum = 11, levelIdx = 5,levelNum = 4,batchNum = 4,anchorIdx = 3,bError=false
batch tryNum = 12, levelIdx = 4,levelNum = 8,batchNum = 8,anchorIdx = 7,bError=false
batch tryNum = 13, levelIdx = 3,levelNum = 16,batchNum = 16,anchorIdx = 15,bError=false
batch tryNum = 14, levelIdx = 2,levelNum = 32,batchNum = 32,anchorIdx = 31,bError=false
batch tryNum = 15, levelIdx = 1,levelNum = 64,batchNum = 64,anchorIdx = 63,bError=false
batch tryNum = 16, levelIdx = 0,levelNum = 128,batchNum = 128,anchorIdx = 127,bError=false
java.lang.reflect.InvocationTargetException
batch tryNum = 17, levelIdx = 1,levelNum = 64,batchNum = 64,anchorIdx = 127,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 18, levelIdx = 2,levelNum = 32,batchNum = 32,anchorIdx = 127,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 19, levelIdx = 3,levelNum = 16,batchNum = 16,anchorIdx = 127,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 20, levelIdx = 4,levelNum = 8,batchNum = 8,anchorIdx = 127,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 21, levelIdx = 5,levelNum = 4,batchNum = 4,anchorIdx = 127,bError=true
java.lang.reflect.InvocationTargetException
batch tryNum = 22, levelIdx = 6,levelNum = 2,batchNum = 2,anchorIdx = 127,bError=true
batch tryNum = 23, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 129,bError=true
java.lang.reflect.InvocationTargetException
single tryNum = 24, levelIdx = 7,levelNum = 1,batchNum = 1,anchorIdx = 129,bError=true
129.1abc: can not be fixed.
3、演算法在資料匯入時的應用
??批量處理方法batchProcMethod，實作insertItems陳述句即可，注意，不要使用try/catch，否則攔截了例外，演算法會認為批量處理成功，
??單資料修正方法singleProcMethod，實作updateItems陳述句即可，注意由于此時實作按特征欄位的值，修改記錄的匯入欄位的值，實際上只修改了一條記錄，必須使用選擇性欄位更新方法，避免未匯入欄位被item物件的默認值所覆寫，
??大致形式如下：
public <T> void batchProcMethod(List<T> subDataList) {
xxxDao.insertItems(subDataList);
}
// 構造單記錄處理的方法，被反射呼叫，必須是public的
public <T> String singleProcMethod(Exception e,T item) {
String errInfo = "";
if (e instanceof DuplicateKeyException) {
// 如果唯一鍵重復，則update
// 根據匯入欄位串列，抽取相關欄位
// String[] importFieldNames = new String[]{"fieldname1","fieldname2",...,"fieldnamen"};
// Map<String,Object> map = itemToMap(item,importFieldNames);
xxxDao.updateItems(map)
}else{
errInfo = e.getMessage();
}
return errInfo;
}
??其中itemToMap方法，可使用反射方法，撰寫為靜態方法，供所有物體類使用，
4、演算法應用注意事項
??演算法代碼在github上，公開狀態，使用了apache license 2.0，如需使用，請按照許可證要求即可，
作者：阿拉伯1999 出處：http://www.cnblogs.com/alabo1999/ 本文著作權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利. 養成良好習慣，好文章隨手頂一下，轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/417069.html
標籤：其他