Alink漫談(七) : 如何劃分訓練數據集和測試數據集

Alink漫談(七) : 如何劃分訓練數據集和測試數據集

目錄

  • Alink漫談(七) : 如何劃分訓練數據集和測試數據集
    • 0x00 摘要
    • 0x01 訓練數據集和測試數據集
    • 0x02 Alink示例代碼
    • 0x03 批處理
      • 3.1 得到記錄數
      • 3.2 隨機選取記錄
        • 3.2.1 得到總記錄數
        • 3.2.2 決定每個task選擇記錄數
        • 3.2.3 每個task選擇記錄
      • 3.3 設置訓練數據集和測試數據集
    • 0x04 流處理
    • 0x05 參考

0x00 摘要

Alink 是阿里巴巴基於實時計算引擎 Flink 研發的新一代機器學習算法平台,是業界首個同時支持批式算法、流式算法的機器學習平台。本文將為大家展現Alink如何劃分訓練數據集和測試數據集。

0x01 訓練數據集和測試數據集

兩分法

一般做預測分析時,會將數據分為兩大部分。一部分是訓練數據,用於構建模型,一部分是測試數據,用於檢驗模型。

三分法

但有時候模型的構建過程中也需要檢驗模型/輔助模型構建,這時會將訓練數據再分為兩個部分:1)訓練數據;2)驗證數據(Validation Data)。所以這種情況下會把數據分為三部分。

  • 訓練數據(Train Data):用於模型構建。
  • 驗證數據(Validation Data):可選,用於輔助模型構建,可以重複使用。
  • 測試數據(Test Data):用於檢測模型構建,此數據只在模型檢驗時使用,用於評估模型的準確率。絕對不允許用於模型構建過程,否則會導致過渡擬合。

Training set是用來訓練模型或確定模型參數的,如ANN中權值等;

Validation set是用來做模型選擇(model selection),即做模型的最終優化及確定,如ANN的結構;

Test set則純粹是為了測試已經訓練好的模型的推廣能力。當然test set並不能保證模型的正確性,他只是說相似的數據用此模型會得出相似的結果。

實際應用

實際應用中,一般只將數據集分成兩類,即training set 和test set,大多數文章並不涉及validation set。我們這裏也不涉及。大家常用的sklearn的train_test_split函數就是將矩陣隨機劃分為訓練子集和測試子集,並返回劃分好的訓練集測試集樣本和訓練集測試集標籤。

0x02 Alink示例代碼

首先我們給出示例代碼,然後會深入剖析:

public class SplitExample {
  public static void main(String[] args) throws Exception {
    String url = "iris.csv";
    String schema = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";

    //這裡是批處理
    BatchOperator data = new CsvSourceBatchOp().setFilePath(url).setSchemaStr(schema);
    SplitBatchOp spliter = new SplitBatchOp().setFraction(0.8);
    spliter.linkFrom(data);
    BatchOperator trainData = spliter;
    BatchOperator testData = spliter.getSideOutput(0);

    // 這裡是流處理
    CsvSourceStreamOp dataS = new CsvSourceStreamOp().setFilePath(url).setSchemaStr(schema);
    SplitStreamOp spliterS = new SplitStreamOp().setFraction(0.4);
    spliterS.linkFrom(dataS);
    StreamOperator train_data = spliterS;
    StreamOperator test_data = spliterS.getSideOutput(0);
  }
}

0x03 批處理

SplitBatchOp是分割批處理的主要類,具體構建DAG的工作是在其linkFrom完成的。

總體思路比較簡單:

  1. 假定有一個採樣比例 fraction
  2. 將數據集分區,并行計算每個分區上的記錄數
  3. 把每個分區上的記錄數累積,得到所有記錄總數 totCount
  4. 從上而下計算出一個採樣總數:numTarget = totCount * fraction
  5. 因為具體選擇元素是在每個分區上做的,所以在每個分區上,分別計算出來這個分區應該採樣的記錄數,比如第n個分區上應採樣記錄數:task_n_count * fraction
  6. 把這些分區 “應該採樣的記錄數” 累積,得出來從下而上計算出的採樣總數: totSelect = task_1_count * fraction + task_2_count * fraction + ... task_n_count * fraction
  7. numTarget 和 totSelect 可能不相等,所以隨機決定把多出來的 numTarget - totSelect 加入到某一個task中。
  8. 在每個task上採樣得到具體的記錄。

3.1 得到記錄數

如果要分割數據,首先必須知道數據集的記錄數。比如這個DataSet的記錄是1萬個?還是十萬個?因為數據集可能會很大,所以這一步操作也使用了并行處理,即把數據分區,然後通過mapPartition操作得到每一個分區上元素的數目。

DataSet<Tuple2<Integer, Long>> countsPerPartition = DataSetUtils.countElementsPerPartition(rows); //返回哪個task有哪些記錄數

DataSet<long[]> numPickedPerPartition = countsPerPartition
    .mapPartition(new CountInPartition(fraction)) //計算總數
    .setParallelism(1)
    .name("decide_count_of_each_partition");

因為每個分區就對應了一個task,所以我們也可以認為,這是獲取了每個task的記錄數。

具體工作是在 DataSetUtils.countElementsPerPartition 中完成的。返回類型是<index of this subtask, record count in this subtask>,比如3號task擁有30個記錄。

public static <T> DataSet<Tuple2<Integer, Long>> countElementsPerPartition(DataSet<T> input) {
   return input.mapPartition(new RichMapPartitionFunction<T, Tuple2<Integer, Long>>() {
      @Override
      public void mapPartition(Iterable<T> values, Collector<Tuple2<Integer, Long>> out) throws Exception {
         long counter = 0;
         for (T value : values) {
            counter++; //計算本task的記錄總數
         }
         out.collect(new Tuple2<>(getRuntimeContext().getIndexOfThisSubtask(), counter));
      }
   });
}

計算總數的工作其實是在下一階段算子中完成的。

3.2 隨機選取記錄

接下來的工作主要是在 CountInPartition.mapPartition 完成的,其作用是隨機決定每個task選擇多少個記錄。

這時候就不需要并行了,所以 .setParallelism(1)

3.2.1 得到總記錄數

得到了每個分區記錄數之後,我們遍歷每個task的記錄數,然後累積得到總記錄數 totCount(就是從上而下計算出來的總數)。

public void mapPartition(Iterable<Tuple2<Integer, Long>> values, Collector<long[]> out) throws Exception {
	long totCount = 0L;
	List<Tuple2<Integer, Long>> buffer = new ArrayList<>();
	for (Tuple2<Integer, Long> value : values) { //遍歷輸入的所有分區記錄
    totCount += value.f1; //f1是Long類型的記錄數
    buffer.add(value);
	}
  ...
  //後續代碼在下面分析。  
}

3.2.2 決定每個task選擇記錄數

然後CountInPartition.mapPartition函數中會隨機決定每個task會選擇的記錄數。mapPartition的參數 Iterable<Tuple2<Integer, Long>> values 就是前一階段的結果 :一個元祖<task id, 每個task的記錄數目>。

把這些元祖結合在一起,記錄在buffer這個列表中。

buffer = {ArrayList@8972}  size = 4
 0 = {Tuple2@8975} "(3,38)" // 3號task,其對應的partition記錄數是38個。
 1 = {Tuple2@8976} "(2,0)"
 2 = {Tuple2@8977} "(0,38)"
 3 = {Tuple2@8978} "(1,74)"

系統的task數目就是buffer大小。

int npart = buffer.size(); // num tasks

然後,根據”記錄總數“計算出來 “隨機訓練數據的個數numTarget”。比如總數1萬,應該隨機分配20%,於是numTarget就應該是2千。這個数字以後會用到。

long numTarget = Math.round((totCount * fraction));

得到每個task的記錄數目,比如是上面buffer中的 38,0,38,還是74,記錄在 eachCount 中。

for (Tuple2<Integer, Long> value : buffer) {
    eachCount[value.f0] = value.f1;
}

得到每個task中隨機選中的訓練記錄數,記錄在 eachSelect 中。就是每個task目前 “記錄数字 * fraction”。比如3號task記錄數是38個,應該選20%,則38*20%=8個。

然後把這些task自己的“隨機訓練記錄數”再累加起來得到 totSelect(就是從下而上計算出來的總數)。

long totSelect = 0L;
for (int i = 0; i < npart; i++) {
    eachSelect[i] = Math.round(Math.floor(eachCount[i] * fraction));
    totSelect += eachSelect[i];
}

請注意,這時候 totSelect 和 之前計算的numTarget就有具體細微出入了,就是理論上的一個数字,但是我們 從上而下 計算 和 從下而上 計算,其結果可能不一樣。通過下面我們可以看出來。

numTarget = all count * fraction

totSelect = task_1_count * fraction + task_2_count * fraction + ...

所以我們下一步要處理這個細微出入,就得到remain,這是”總體算出來的隨機數目” numTarget 和 “從所有task選中的隨機訓練記錄數累積” totSelect 的差。

if (totSelect < numTarget) {
    long remain = numTarget - totSelect;
    remain = Math.min(remain, totCount - totSelect);

如果剛好個數相等,則就正常分配。

if (remain == totCount - totSelect) {

如果數目不等,隨機決定把”多出來的remain”加入到eachSelect數組中的隨便一個記錄上。

for (int i = 0; i < Math.min(remain, npart); i++) {
    int taskId = shuffle.get(i);
    while (eachSelect[taskId] >= eachCount[taskId]) {
          taskId = (taskId + 1) % npart;
    }
    eachSelect[taskId]++;
}

最後給出所有信息

long[] statistics = new long[npart * 2];
for (int i = 0; i < npart; i++) {
    statistics[i] = eachCount[i];
    statistics[i + npart] = eachSelect[i];
}
out.collect(statistics);

// 我們這裡是4核,所以前面四項是eachCount,後面是eachSelect
statistics = {long[8]@9003} 
 0 = 38 //eachCount
 1 = 38
 2 = 36
 3 = 38
   
 4 = 31 //eachSelect
 5 = 31
 6 = 28
 7 = 30

這些信息是作為廣播變量存儲起來的,馬上下面就會用到。

 .withBroadcastSet(numPickedPerPartition, "counts")

3.2.3 每個task選擇記錄

CountInPartition.PickInPartition函數中會隨機在每個task選擇記錄。

首先得到task數目 和 之前存儲的廣播變量(就是之前剛剛存儲的)。

int npart = getRuntimeContext().getNumberOfParallelSubtasks();
List<long[]> bc = getRuntimeContext().getBroadcastVariable("counts");

分離count和select。

long[] eachCount = Arrays.copyOfRange(bc.get(0), 0, npart);
long[] eachSelect = Arrays.copyOfRange(bc.get(0), npart, npart * 2);

得到總task數目

int taskId = getRuntimeContext().getIndexOfThisSubtask();

得到自己 task 對應的 count, select

long count = eachCount[taskId];
long select = eachSelect[taskId];

添加本task對應的記錄,隨機洗牌打亂順序

for (int i = 0; i < count; i++) {
     shuffle.add(i); //就是把count內的数字加到數組
}
Collections.shuffle(shuffle, new Random(taskId)); //洗牌打亂順序

// suffle舉例
shuffle = {ArrayList@8987}  size = 38
 0 = {Integer@8994} 17
 1 = {Integer@8995} 8
 2 = {Integer@8996} 33
 3 = {Integer@8997} 34
 4 = {Integer@8998} 20
 5 = {Integer@8999} 0
 6 = {Integer@9000} 26
 7 = {Integer@9001} 27
 8 = {Integer@9002} 23
 9 = {Integer@9003} 28
 10 = {Integer@9004} 9
 11 = {Integer@9005} 16
 12 = {Integer@9006} 13
 13 = {Integer@9007} 2
 14 = {Integer@9008} 5
 15 = {Integer@9009} 31
 16 = {Integer@9010} 15
 17 = {Integer@9011} 22
 18 = {Integer@9012} 18
 19 = {Integer@9013} 35
 20 = {Integer@9014} 36
 21 = {Integer@9015} 12
 22 = {Integer@9016} 7
 23 = {Integer@9017} 21
 24 = {Integer@9018} 14
 25 = {Integer@9019} 1
 26 = {Integer@9020} 10
 27 = {Integer@9021} 30
 28 = {Integer@9022} 29
 29 = {Integer@9023} 19
 30 = {Integer@9024} 25
 31 = {Integer@9025} 32
 32 = {Integer@9026} 37
 33 = {Integer@9027} 4
 34 = {Integer@9028} 11
 35 = {Integer@9029} 6
 36 = {Integer@9030} 3
 37 = {Integer@9031} 24

隨機選擇,把選擇后的再排序回來

for (int i = 0; i < select; i++) {
    selected[i] = shuffle.get(i); //這時候select看起來是按照順序選擇,但是實際上suffle裏面已經是亂序
}
Arrays.sort(selected); //這次再排序

// selected舉例,一共30個
selected = {int[30]@8991} 
 0 = 0
 1 = 1
 2 = 2
 3 = 5
 4 = 7
 5 = 8
 6 = 9
 7 = 10
 8 = 12
 9 = 13
 10 = 14
 11 = 15
 12 = 16
 13 = 17
 14 = 18
 15 = 19
 16 = 20
 17 = 21
 18 = 22
 19 = 23
 20 = 26
 21 = 27
 22 = 28
 23 = 29
 24 = 30
 25 = 31
 26 = 33
 27 = 34
 28 = 35
 29 = 36

發送選擇的數據

if (numEmits < selected.length && iRow == selected[numEmits]) {
    out.collect(row);
    numEmits++;
}

3.3 設置訓練數據集和測試數據集

output是訓練數據集,SideOutput是測試數據集。因為這兩個數據集在Alink內部都是Table類型,所以直接使用了SQL算子 minusAll 來完成分割。

this.setOutput(out, in.getSchema());
this.setSideOutputTables(new Table[]{in.getOutputTable().minusAll(this.getOutputTable())});

0x04 流處理

訓練是在SplitStreamOp類完成的,其通過linkFrom完成了模型的構建。

流處理依賴SplitStream 和 SelectTransformation 這兩個類來完成分割流。具體並沒有建立一個物理操作,而只是影響了上游算子如何與下游算子聯繫,如何選擇記錄

SplitStream <Row> splited = in.getDataStream().split(new RandomSelectorOp(getFraction()));

首先,用RandomSelectorOp來隨機決定輸出時候選擇哪個流。我們可以看到,這裏就是隨便起了”a”, “b” 這兩個名字而已。

class RandomSelectorOp implements OutputSelector <Row> {
   private double fraction;
   private Random random = null;
   @Override
   public Iterable <String> select(Row value) {
      if (null == random) {
         random = new Random(System.currentTimeMillis());
      }
      List <String> output = new ArrayList <String>(1);
      output.add((random.nextDouble() < fraction ? "a" : "b")); //隨機選取数字分配,隨意起的名字
      return output;
   }
}

其次,得到那兩個隨機生成的流。

DataStream <Row> partA = splited.select("a");
DataStream <Row> partB = splited.select("b");

最後把這兩個流分別設置為output和sideOutput。

this.setOutput(partA, in.getSchema()); //訓練集
this.setSideOutputTables(new Table[]{
DataStreamConversionUtil.toTable(getMLEnvironmentId(), partB, in.getSchema())}); //驗證集

最後返回本身,這時候SplitStreamOp擁有兩個成員變量:

this.output就是訓練集。

this.sideOutPut就是驗證集。

return this;

0x05 參考

訓練數據,驗證數據和測試數據分析

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

網頁設計公司推薦不同的風格,搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

您可能也會喜歡…