由于算力的限制，有时我们无法使用足够大的batchsize，此时该如何使用BN呢？本文将介绍两种在小batchsize也可以发挥BN性能的方法。
 
前言
 
BN（Batch Normalization）几乎是目前神经网络的必选组件，但是使用BN有两个前提要求：
 
 
batchsize不能太小；
 
 
每一个minibatch和整体数据集同分布。
 
不然的话，非但不能发挥BN的优势，甚至会适得其反。但是由于算力的限制，有时我们无法使用足够大的batchsize，此时该如何使用BN呢？本文介绍两篇在小batchsize也可以发挥BN性能的方法。解决思路为：既然batchsize太小的情况下，无法保证当前minibatch收集到的数据和整体数据同分布。那么能否多收集几个batch的数据进行统计呢？这两篇工作分别分别是：
 
 
BRN：Batch Renormalization: Towards Reducing Minibatch Dependence in Batch-Normalized Models
 
 
CBN：Cross-Iteration Batch Normalization
 
另外，本文也会给出代码解析，帮助大家理解。
 
batchsize过小的场景
 
通常情况下，大家对CNN任务的研究一般为公开的数据集指标负责。分类任务为ImageNet数据集负责，其尺度为224X224。检测任务为coco数据集负责，其尺度为640X640左右。分割任务一般为coco或PASCAL VOC数据集负责，后者的尺度大概在500X500左右。再加上例如resize的前处理操作，真正送入网络的图片的分辨率都不算太大。一般性能的GPU也很容易实现大的batchsize(例如大于32)的支持。
 
但是实际的项目中，经常遇到需要处理的图片尺度过大的场景，例如我们使用500w像素甚至2000w像素的工业相机进行数据采集，500w的相机采集的图片尺度就是2500X2000左右。而对于微小的缺陷检测、高精度的关键点检测或小物体的目标检测等任务，我们一般不太想粗暴降低输入图片的分辨率，这样违背了我们使用高分辨率相机的初衷，也可能导致丢失有用特征。在算力有限的情况下，我们的batchsize就无法设置太大，甚至只能为1或2。小的batchsize会带来很多训练上的问题，其中BN问题就是最突出的。虽然大batchsize训练是一个共识，但是现实中可能无法具有充足的资源，因此我们需要一些处理手段。
 
BN回顾
 
 
首先Batch Normalization 中的Normalization被称为标准化，通过将数据进行平和缩放拉到一个特定的分布。BN就是在batch维度上进行数据的标准化。BN的引入是用来解决 internal covariate shift 问题，即训练迭代中网络激活的分布的变化对网络训练带来的破坏。BN通过在每次训练迭代的时候，利用minibatch计算出的当前batch的均值和方差，进行标准化来缓解这个问题。虽然How Does Batch Normalization Help Optimization 这篇文章探究了BN其实和Internal Covariate Shift （ICS）问题关系不大，本文不深入讨论，这个会在以后的文章中细说。
 
一般来说，BN有两个优点：
 
 
降低对初始化、学习率等超参的敏感程度，因为每层的输入被BN拉成相对稳定的分布，也能加速收敛过程。
 
 
应对梯度饱和和梯度弥散，主要是对于使用sigmoid和tanh的激活函数的网络。
 
当然，BN的使用也有两个前提：
 
 
minibatch和全部数据同分布。因为训练过程每个minibatch从整体数据中均匀采样，不同分布的话minibatch的均值和方差和训练样本整体的均值和方差是会存在较大差异的，在测试的时候会严重影响精度。
 
 
batchsize不能太小，否则效果会较差，论文给的一般性下限是32。
 
再来回顾一下BN的具体做法：
 
 
 
训练的时候：使用当前batch统计的均值和方差对数据进行标准化，同时优化优化gamma和beta两个参数。另外利用指数滑动平均收集全局的均值和方差。
 
 
测试的时候：使用训练时收集全局均值和方差以及优化好的gamma和beta进行推理。
 
可以看出，要想BN真正work，就要保证训练时当前batch的均值和方差逼近全部数据的均值和方差。
 
BRN
 
论文题目：Batch Renormalization: Towards Reducing Minibatch Dependence in Batch-Normalized Models
 
论文地址： https://arxiv.org/pdf/1702.03275.pdf
 
代码地址： https://github.com/ludvb/batchrenorm
 
核心解析：
 
本文的核心思想就是：训练过程中，由于batchsize较小，当前minibatch统计到的均值和方差与全部数据有差异，那么就对当前的均值和方差进行修正。修正的方法主要是利用到通过滑动平均收集到的全局均值和标准差。看公式：
 
上面公式中，i表示网络的第i层。μ和σ表示网络推理时的均值和标准差，也就是训练过程通过滑动平均收集的到均值和方差。μB和σb表示当前训练迭代过程中的实际统计到的均值和标准差。BN在小batch不work的根本原因就是这两组参数存在较大的差异。通过r和d对训练过程中数据进行线性变换，在该变化下，上公式左右两端就严格相等了。其实标准的BN就是r=1，d=0的一种情况。对于某一个特定的minibatch，其中r和d可以看成是固定的，是直接计算出来的，不需要梯度优化的。
 
具体流程：
 
 
 
统计当前batch数据的均值和标注差，和标准BN做法一致。
 
 
根据当前batch的均值和标准差结合全局的均值和标准差利用上面的公式计算r和d；注意该运算是不参与梯度反向传播的。另外，r和d需要增加一个限制，直接clip操作就好。
 
 
利用当前的均值和标准差对当前数据执行Normalization操作，利用上面计算得到的r和d对当前batch进行线性变换。
 
 
滑动平均收集全局均值和标注差。
 
测试过程和标准BN一样。其实本质上，就是训练的过程中使用全局的信息进行更新当前batch的数据。间接利用了全局的信息，而非当前这一个batch的信息。
 
实验效果：
 
在较大的batchsize(32)的时候，与标准BN相比，不会丢失效果，训练过程一如既往稳定高效。如下：
 
 
在小的batchsize(4)下, 本文做法依然接近batchsize为32的时候，可见在小batchsize下是work的。
 
 
代码解析：
 
def forward(self, x):    if x.dim() > 2:      x = x.transpose(1, -1)    if self.training: # 训练过程      dims = [i for i in range(x.dim() - 1)      batch_mean = x.mean(dims) # 计算均值      batch_std = x.std(dims, unbiased=False) + self.eps # 计算标准差      # 按照公式计算r和d      r = (batch_std.detach() / self.running_std.view_as(batch_std)).clamp_(1 / self.rmax, self.rmax)      d = ((batch_mean.detach() - self.running_mean.view_as(batch_mean))            / self.running_std.view_as(batch_std)).clamp_(-self.dmax, self.dmax)      # 对当前数据进行标准化和线性变换      x = (x - batch_mean) / batch_std * r + d      # 滑动平均收集全局均值和标注差      self.running_mean += self.momentum * (batch_mean.detach() - self.running_mean)      self.running_std += self.momentum * (batch_std.detach() - self.running_std)      self.num_batches_tracked += 1    else: # 测试过程    x = (x - self.running_mean) / self.running_std  return x
 
CBN
 
论文题目：Cross-Iteration Batch Normalization
 
论文地址：https://arxiv.org/abs/2002.05712
 
代码地址：https://github.com/Howal/Cross-iterationBatchNorm
 
 
本文认为BRN的问题在于它使用的全局均值和标准差不是当前网络权重下获取的，因此不是exactly正确的，所以batchsize再小一点，例如为1或2时就不太work了。本文使用泰勒多项式逼近原理来修正当前的均值和标准差，同样也是间接利用了全局的均值和方差信息。简述就是：当前batch的均值和方差来自之前的K次迭代均值和方差的平均，由于网络权重一直在更新，所以不能直接粗暴求平均。本文而是利用泰勒公式估计前面的迭代在当前权重下的数值。
 
泰勒公式：
 
泰勒公式是一个用函数在某点的信息描述其附近取值的公式。如果函数满足一定的条件，泰勒公式可以用函数在某一点的各阶导数值做系数构建一个多项式来近似表达这个函数。教科书介绍如下：
 
 
核心解析：
 
本文做法，由于网络一般使用SGD更新权重，因此网络权重的变化是平滑的，所以适用泰勒公式。如下，t为训练过程中当前迭代时刻，t-τ为t时刻向前τ时刻。θ为网络权重，权重下标代表该权重的时刻。μ为当前minibatch均值，v为当强minibatch平方的均值，是为了计算标准差。因此直接套用泰勒公式得到：
 
上面这两个公式就是为了估计在t-τ时刻，t时刻的权重下的均值和方差的参数估计。BRN可以看作没有进行该方法估计，使用的依然是t-τ时刻权重的参数估计。其中O为高阶项，因为该式主要由一阶项控制，因此高阶项目可以忽略。上面的公式还要进一步简化，主要是偏导项的求法。假设当前层为l，实际上∂μ/ ∂θ 和 ∂ν/∂θ依赖与所有l层之前层的权重，求导计算量极大。不过经验观察到，l层之前层的偏数下降很快，因此可以忽略掉，仅仅计算当前层的权重偏导。
 
因此化简为如下，可以看出，求偏导的部分，只考虑对当前层的偏导数，注意上标l表示网络层的意思。至此，之前时刻在当前权重下的均值和方差已经估计出来了。
 
下面穿插代码解析整个计算过程。
 
首先是统计计算当前batch的数据，和标准BN没有差别。代码为：
 
cur_mu = y.mean(dim=1)  # 当前层的均值cur_meanx2 = torch.pow(y, 2).mean(dim=1)  # 当前值平方的均值，计算标准差使用cur_sigma2 = y.var(dim=1)  # 当前值的方差
 
对当前网络层求偏导，直接使用torch的内置函数。代码：​​​​​​​
 
# 注意 grad_outputs = self.ones : 不同值的梯度对结果影响程度不同，类似torch.sum()的作用。dmudw = torch.autograd.grad(cur_mu, weight, self.ones, retain_graph=True)[0]dmeanx2dw = torch.autograd.grad(cur_meanx2, weight, self.ones, retain_graph=True)[0]
 
使用公式（7）和（8）继续下面的计算，也就是向前累计K次估计数值，更新到当前batch的均值和方差的计算上，这里引入了一个超参就是k的大小，它表示当前的迭代向后回溯到多长的步长的迭代。实验探究k=8是一个比较折中的选择。k=1的时候，RBN退化成了原始的BN：
 
代码如下，其中这里的self.pre_mu, self.pre_dmudw, self.pre_weight是前面每次迭代收集到了窗口k大小的数值，分别代表均值、均值对权重的偏导、权重。self.pre_meanx2, self.pre_dmeanx2dw, self.pre_weight同理，是对应平方均值的。​​​​​​​
 
# 利用泰勒公式估计mu_all = torch.stack \  ([cur_mu, ] + [tmp_mu + (self.rho * tmp_d * (weight.data - tmp_w)).sum(1).sum(1).sum(1) for tmp_mu, tmp_d, tmp_w in zip(self.pre_mu, self.pre_dmudw, self.pre_weight)])
meanx2_all = torch.stack \  ([cur_meanx2, ] + [tmp_meanx2 + (self.rho * tmp_d * (weight.data - tmp_w)).sum(1).sum(1).sum(1) for tmp_meanx2, tmp_d, tmp_w in zip(self.pre_meanx2, self.pre_dmeanx2dw, self.pre_weight)])
 
上面所说的变量收集迭代过程如下：​​​​​​​
 
# 动态维护buffer_num长度的均值、均值平方、偏导、权重self.pre_mu = [cur_mu.detach(), ] + self.pre_mu[:(self.buffer_num - 1)]self.pre_meanx2 = [cur_meanx2.detach(), ] + self.pre_meanx2[:(self.buffer_num - 1)]self.pre_dmudw = [dmudw.detach(), ] + self.pre_dmudw[:(self.buffer_num - 1)]self.pre_dmeanx2dw = [dmeanx2dw.detach(), ] + self.pre_dmeanx2dw[:(self.buffer_num - 1)]tmp_weight = torch.zeros_like(weight.data)tmp_weight.copy_(weight.data)self.pre_weight = [tmp_weight.detach(), ] + self.pre_weight[:(self.buffer_num - 1)]
 
计算获取当前batch的均值和方差，取修正后的K次迭代数据的平均即可。​​​​​​​
 
# 利用收集到的一定窗口长度的均值和平方均值，计算当前均值和方差sigma2_all = meanx2_all - torch.pow(mu_all, 2)re_mu_all = mu_all.clone()re_meanx2_all = meanx2_all.clone()re_mu_all[sigma2_all < 0] = 0re_meanx2_all[sigma2_all < 0] = 0count = (sigma2_all >= 0).sum(dim=0).float()mu = re_mu_all.sum(dim=0) / count # 平均操作sigma2 = re_meanx2_all.sum(dim=0) / count - torch.pow(mu, 2)
 
均值和方差使用过程，和标准BN没有区别。​​​​​​​
 
# 标准化过程，和原始BN没有区别y = y - mu.view(-1, 1)if self.out_p:  # 仅仅控制开平方的位置  y = y / (sigma2.view(-1, 1) + self.eps) ** .5else:  y = y / (sigma2.view(-1, 1) ** .5 + self.eps)
 
最后再理解一下：
 
mu_0是当前batch统计获取的均值，mu_1是上一batch统计获取的均值。当前batch计算BN的时候也想利用到mu_1，但是统计mu_1的时候利用到网络的权重也是上一次的，直接使用肯定有问题，所以本文使用泰勒公式估计出mu_1在当前权重下应该是什么样子。方差估计同理。
 
实验效果：
 
这里的Naive CBN 是上一篇论文BRN的做法，可以认为是CBN不使用泰勒估计的一种特例。在batchsize下降的过程中，CBN指标依然坚挺，甚至超过了GN(不过也侧面反应了GN确实厉害)。而原始BN和其改进版BRN在batchsize更小的时候都不太work了。
 

Python循环产生批量数据batch
 
 
 
目录
 
Python循环产生批量数据batch
 
一、Python循环产生批量数据batch
 
二、TensorFlow循环产生批量数据batch 
 
(1) tf.train.slice_input_producer
 
(2) tf.train.batch和tf.train.shuffle_batch
 
(3) TF循环产生批量数据batch 的完整例子
 
三、更加实用的方法：数据巨大的情况
 

一、Python循环产生批量数据batch
 
   在机器学习中，经常需要产生一个batch的数据用于训练模型，比如tensorflow的接口tf.train.batch就可以实现数据批量读取的操作。本博客将不依赖TensorFlow，实现一个类似于tensorflow接口tf.train.batch的方法，循环产生批量数据batch。实现的代码和测试的代码如下：
 
   TXT文本如下，格式：图片名 label1 label2 ，注意label可以多个
 
1.jpg 1 11
2.jpg 2 12
3.jpg 3 13
4.jpg 4 14
5.jpg 5 15
6.jpg 6 16
7.jpg 7 17
8.jpg 8 18
 
    要想产生batch数据，关键是要用到Python的关键字yield，实现一个batch一个batch的返回数据，代码实现主要有两个方法：
 
 
 
def get_data_batch(inputs, batch_size=None, shuffle=False):
    '''
    循环产生批量数据batch
    :param inputs: list数据
    :param batch_size: batch大小
    :param shuffle: 是否打乱inputs数据
    :return: 返回一个batch数据
    '''
 
 
def get_next_batch(batch):
    return batch.__next__()
 
 
    使用时，将数据传到 get_data_batch( )方法，然后使用get_next_batch( )获得一个batch数据，完整的Python代码如下：
 
# -*-coding: utf-8 -*-
"""
    @Project: create_batch_data
    @File   : create_batch_data.py
    @Author : panjq
    @E-mail : pan_jinquan@163.com
    @Date   : 2017-10-27 18:20:15
"""
import math
import random
import os
import glob
import numpy as np


def get_data_batch(inputs, batch_size=None, shuffle=False):
    '''
    循环产生批量数据batch
    :param inputs: list类型数据，多个list,请[list0,list1,...]
    :param batch_size: batch大小
    :param shuffle: 是否打乱inputs数据
    :return: 返回一个batch数据
    '''
    rows = len(inputs[0])
    indices = list(range(rows))
    # 如果输入是list,则需要转为list
    if shuffle:
        random.seed(100)
        random.shuffle(indices)
    while True:
        batch_indices = np.asarray(indices[0:batch_size])  # 产生一个batch的index
        indices = indices[batch_size:] + indices[:batch_size]  # 循环移位，以便产生下一个batch
        batch_data = []
        for data in inputs:
            data = np.asarray(data)
            temp_data=data[batch_indices] #使用下标查找，必须是ndarray类型类型
            batch_data.append(temp_data.tolist())
        yield batch_data

def get_data_batch2(inputs, batch_size=None, shuffle=False):
    '''
    循环产生批量数据batch
    :param inputs: list类型数据，多个list,请[list0,list1,...]
    :param batch_size: batch大小
    :param shuffle: 是否打乱inputs数据
    :return: 返回一个batch数据
    '''
    # rows,cols=inputs.shape
    rows = len(inputs[0])
    indices = list(range(rows))
    if shuffle:
        random.seed(100)
        random.shuffle(indices)
    while True:
        batch_indices = indices[0:batch_size]  # 产生一个batch的index
        indices = indices[batch_size:] + indices[:batch_size]  # 循环移位，以便产生下一个batch
        batch_data = []
        for data in inputs:
            temp_data = find_list(batch_indices, data)
            batch_data.append(temp_data)
        yield batch_data



def get_data_batch_one(inputs, batch_size=None, shuffle=False):
    '''
    产生批量数据batch,非循环迭代
    迭代次数由:iter_nums= math.ceil(sample_nums / batch_size)
    :param inputs: list类型数据，多个list,请[list0,list1,...]
    :param batch_size: batch大小
    :param shuffle: 是否打乱inputs数据
    :return: 返回一个batch数据
    '''
    # rows,cols=inputs.shape
    rows = len(inputs[0])
    indices = list(range(rows))
    if shuffle:
        random.seed(100)
        random.shuffle(indices)
    while True:
        batch_data = []
        cur_nums = len(indices)
        batch_size = np.where(cur_nums > batch_size, batch_size, cur_nums)
        batch_indices = indices[0:batch_size]  # 产生一个batch的index
        indices = indices[batch_size:]
        # indices = indices[batch_size:] + indices[:batch_size]  # 循环移位，以便产生下一个batch
        for data in inputs:
            temp_data = find_list(batch_indices, data)
            batch_data.append(temp_data)
        yield batch_data


def find_list(indices, data):
    out = []
    for i in indices:
        out = out + [data[i]]
    return out


def get_list_batch(inputs, batch_size=None, shuffle=False):
    '''
    循环产生batch数据
    :param inputs: list数据
    :param batch_size: batch大小
    :param shuffle: 是否打乱inputs数据
    :return: 返回一个batch数据
    '''
    if shuffle:
        random.shuffle(inputs)
    while True:
        batch_inouts = inputs[0:batch_size]
        inputs = inputs[batch_size:] + inputs[:batch_size]  # 循环移位，以便产生下一个batch
        yield batch_inouts


def load_file_list(text_dir):
    text_dir = os.path.join(text_dir, '*.txt')
    text_list = glob.glob(text_dir)
    return text_list


def get_next_batch(batch):
    return batch.__next__()


def load_image_labels(finename):
    '''
    载图txt文件，文件中每行为一个图片信息，且以空格隔开：图像路径 标签1 标签1，如：test_image/1.jpg 0 2
    :param test_files:
    :return:
    '''
    images_list = []
    labels_list = []
    with open(finename) as f:
        lines = f.readlines()
        for line in lines:
            # rstrip：用来去除结尾字符、空白符(包括\n、\r、\t、' '，即：换行、回车、制表符、空格)
            content = line.rstrip().split(' ')
            name = content[0]
            labels = []
            for value in content[1:]:
                labels.append(float(value))
            images_list.append(name)
            labels_list.append(labels)
    return images_list, labels_list


if __name__ == '__main__':
    filename = './training_data/test.txt'
    images_list, labels_list = load_image_labels(filename)

    # 若输入为np.arange数组，则需要tolist()为list类型，如：
    # images_list = np.reshape(np.arange(8*3), (8,3))
    # labels_list = np.reshape(np.arange(8*3), (8,3))
    # images_list=images_list.tolist()
    # labels_list=labels_list.tolist()

    iter = 5  # 迭代3次，每次输出一个batch个
    # batch = get_data_batch([images_list, labels_list], batch_size=3, shuffle=False)
    batch = get_data_batch2(inputs=[images_list,labels_list], batch_size=5, shuffle=True)

    for i in range(iter):
        print('**************************')
        batch_images, batch_labels = get_next_batch(batch)
        print('batch_images:{}'.format(batch_images))
        print('batch_labels:{}'.format(batch_labels))


 
   运行输出结果为：
 
 
 
**************************
 batch_images:['1.jpg', '2.jpg', '3.jpg']
 batch_labels:[[1.0, 11.0], [2.0, 12.0], [3.0, 13.0]]
 **************************
 batch_images:['4.jpg', '5.jpg', '6.jpg']
 batch_labels:[[4.0, 14.0], [5.0, 15.0], [6.0, 16.0]]
 **************************
 batch_images:['7.jpg', '8.jpg', '1.jpg']
 batch_labels:[[7.0, 17.0], [8.0, 18.0], [1.0, 11.0]]
 **************************
 batch_images:['2.jpg', '3.jpg', '4.jpg']
 batch_labels:[[2.0, 12.0], [3.0, 13.0], [4.0, 14.0]]
 **************************
 batch_images:['5.jpg', '6.jpg', '7.jpg']
 batch_labels:[[5.0, 15.0], [6.0, 16.0], [7.0, 17.0]]
 
 
Process finished with exit code 0
 
 
二、TensorFlow循环产生批量数据batch 
 
    使用TensorFlow实现产生批量数据batch，需要几个接口，
 
(1) tf.train.slice_input_producer
 
 
 
tf.train.slice_input_producer是一个tensor生成器，作用是按照设定，每次从一个tensor列表中按顺序或者随机抽取出一个tensor放入文件名队列。
 
 
slice_input_producer(tensor_list,
                     num_epochs=None,
                     shuffle=True,
                     seed=None,
                     capacity=32,
                     shared_name=None,
                     name=None)
# 第一个参数
#           tensor_list：包含一系列tensor的列表，表中tensor的第一维度的值必须相等，即个数必须相等，有多少个图像，就应该有多少个对应的标签。
# 第二个参数num_epochs: 可选参数，是一个整数值，代表迭代的次数，如果设置
#           num_epochs = None, 生成器可以无限次遍历tensor列表，如果设置为
#           num_epochs = N，生成器只能遍历tensor列表N次。
# 第三个参数shuffle： bool类型，设置是否打乱样本的顺序。一般情况下，如果shuffle = True，生成的样本顺序就被打乱了，在批处理的时候不需要再次打乱样本，使用
#           tf.train.batch函数就可以了;
#           如果shuffle = False, 就需要在批处理时候使用
#           tf.train.shuffle_batch函数打乱样本。
# 第四个参数seed: 可选的整数，是生成随机数的种子，在第三个参数设置为shuffle = True的情况下才有用。
# 第五个参数capacity：设置tensor列表的容量。
# 第六个参数shared_name：可选参数，如果设置一个‘shared_name’，则在不同的上下文环境（Session）中可以通过这个名字共享生成的tensor。
# 第七个参数name：可选，设置操作的名称
 
    tf.train.slice_input_producer定义了样本放入文件名队列的方式，包括迭代次数，是否乱序等，要真正将文件放入文件名队列，还需要调用tf.train.start_queue_runners 函数来启动执行文件名队列填充的线程，之后计算单元才可以把数据读出来，否则文件名队列为空的，计算单元就会处于一直等待状态，导致系统阻塞。
 
    例子：
 
import tensorflow as tf
 
images = ['img1', 'img2', 'img3', 'img4', 'img5']
labels= [1,2,3,4,5]
 
epoch_num=8
 
f = tf.train.slice_input_producer([images, labels],num_epochs=None,shuffle=False)
 
with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    coord = tf.train.Coordinator()
    threads = tf.train.start_queue_runners(sess=sess, coord=coord)
    for i in range(epoch_num):
        k = sess.run(f)
        print '************************'
        print (i,k)
 
    coord.request_stop()
    coord.join(threads)
 
(2) tf.train.batch和tf.train.shuffle_batch
 
    tf.train.batch是一个tensor队列生成器，作用是按照给定的tensor顺序，把batch_size个tensor推送到文件队列，作为训练一个batch的数据，等待tensor出队执行计算。
 
tf.train.batch(tensors, 
               batch_size, 
               num_threads=1, 
               capacity=32,
               enqueue_many=False, 
               shapes=None, 
               dynamic_pad=False,
               allow_smaller_final_batch=False, 
               shared_name=None, 
               name=None)
# 第一个参数tensors：tensor序列或tensor字典，可以是含有单个样本的序列;
# 第二个参数batch_size: 生成的batch的大小;
# 第三个参数num_threads：执行tensor入队操作的线程数量，可以设置使用多个线程同时并行执行，提高运行效率，但也不是数量越多越好;
# 第四个参数capacity： 定义生成的tensor序列的最大容量;
# 第五个参数enqueue_many： 定义第一个传入参数tensors是多个tensor组成的序列，还是单个tensor;
# 第六个参数shapes： 可选参数，默认是推测出的传入的tensor的形状;
# 第七个参数dynamic_pad： 定义是否允许输入的tensors具有不同的形状，设置为True，会把输入的具有不同形状的tensor归一化到相同的形状;
# 第八个参数allow_smaller_final_batch： 设置为True，表示在tensor队列中剩下的tensor数量不够一个batch_size的情况下，允许最后一个batch的数量少于batch_size， 设置为False，则不管什么情况下，生成的batch都拥有batch_size个样本;
# 第九个参数shared_name： 可选参数，设置生成的tensor序列在不同的Session中的共享名称;
# 第十个参数name： 操作的名称;
 
    如果tf.train.batch的第一个参数 tensors 传入的是tenor列表或者字典，返回的是tensor列表或字典，如果传入的是只含有一个元素的列表，返回的是单个的tensor，而不是一个列表。
 
    与tf.train.batch函数相对的还有一个tf.train.shuffle_batch函数，两个函数作用一样，都是生成一定数量的tensor,组成训练一个batch需要的数据集，区别是tf.train.shuffle_batch会打乱样本顺序。
 
(3) TF循环产生批量数据batch 的完整例子
 
# -*-coding: utf-8 -*-
"""
    @Project: LSTM
    @File   : tf_create_batch_data.py
    @Author : panjq
    @E-mail : pan_jinquan@163.com
    @Date   : 2018-10-28 17:50:24
"""
import tensorflow as tf


def get_data_batch(inputs,batch_size,labels_nums,one_hot=False,shuffle=False,num_threads=1):
    '''
    :param inputs: 输入数据，可以是多个list
    :param batch_size:
    :param labels_nums:标签个数
    :param one_hot:是否将labels转为one_hot的形式
    :param shuffle:是否打乱顺序,一般train时shuffle=True,验证时shuffle=False
    :return:返回batch的images和labels
    '''
    # 生成队列
    inputs_que= tf.train.slice_input_producer(inputs, shuffle=shuffle)
    min_after_dequeue = 200
    capacity = min_after_dequeue + 3 * batch_size  # 保证capacity必须大于min_after_dequeue参数值
    if shuffle:
        out_batch = tf.train.shuffle_batch(inputs_que,
                                            batch_size=batch_size,
                                            capacity=capacity,
                                            min_after_dequeue=min_after_dequeue,
                                            num_threads=num_threads)
    else:
        out_batch = tf.train.batch(inputs_que,
                                    batch_size=batch_size,
                                    capacity=capacity,
                                    num_threads=num_threads)
    return out_batch

def get_batch_images(images,labels,batch_size,labels_nums,one_hot=False,shuffle=False):
    '''
    :param images:图像
    :param labels:标签
    :param batch_size:
    :param labels_nums:标签个数
    :param one_hot:是否将labels转为one_hot的形式
    :param shuffle:是否打乱顺序,一般train时shuffle=True,验证时shuffle=False
    :return:返回batch的images和labels
    '''
    images_que, labels_que= tf.train.slice_input_producer([images,labels], shuffle=shuffle)
    min_after_dequeue = 200
    capacity = min_after_dequeue + 3 * batch_size  # 保证capacity必须大于min_after_dequeue参数值
    if shuffle:
        images_batch, labels_batch = tf.train.shuffle_batch([images_que, labels_que],
                                                            batch_size=batch_size,
                                                            capacity=capacity,
                                                            min_after_dequeue=min_after_dequeue)
    else:
        images_batch, labels_batch = tf.train.batch([images_que, labels_que],
                                                    batch_size=batch_size,
                                                    capacity=capacity)
    if one_hot:
        labels_batch = tf.one_hot(labels_batch, labels_nums, 1, 0)
    return images_batch,labels_batch

def load_image_labels(finename):
    '''
    载图txt文件，文件中每行为一个图片信息，且以空格隔开：图像路径 标签1 标签1，如：test_image/1.jpg 0 2
    :param test_files:
    :return:
    '''
    images_list=[]
    labels_list=[]
    with open(finename) as f:
        lines = f.readlines()
        for line in lines:
            #rstrip：用来去除结尾字符、空白符(包括\n、\r、\t、' '，即：换行、回车、制表符、空格)
            content=line.rstrip().split(' ')
            name=content[0]
            labels=[]
            for value in content[1:]:
                labels.append(float(value))
            images_list.append(name)
            labels_list.append(labels)
    return images_list,labels_list

if __name__ == '__main__':
    filename='./training_data/train.txt'
    # 输入数据可以是list,也可以是np.array
    images_list, labels_list=load_image_labels(filename)
    # np.arange数组如：
    # images_list = np.reshape(np.arange(8*3), (8,3))
    # labels_list = np.reshape(np.arange(8*3), (8,3))

    iter = 5  # 迭代5次，每次输出一个batch个
    # batch_images, batch_labels = get_data_batch( inputs=[images_list, labels_list],batch_size=3,labels_nums=2,one_hot=False,shuffle=False,num_threads=1)
    # 或者
    batch_images, batch_labels = get_batch_images(images_list,labels_list,batch_size=3,labels_nums=2,one_hot=False,shuffle=False)
    with tf.Session() as sess:  # 开始一个会话
        sess.run(tf.global_variables_initializer())
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(coord=coord)
        for i in range(iter):
            # 在会话中取出images和labels
            images, labels = sess.run([batch_images, batch_labels] )
            print('**************************')
            print('batch_images:{}'.format(images ))
            print('batch_labels:{}'.format(labels))

        # 停止所有线程
        coord.request_stop()
        coord.join(threads)

 
    运行输出结果：
 
 
 
**************************
 batch_images:[b'1.jpg' b'2.jpg' b'3.jpg']
 batch_labels:[[ 1. 11.] [ 2. 12.][ 3. 13.]]
 **************************
 batch_images:[b'4.jpg' b'5.jpg' b'6.jpg']
 batch_labels:[[ 4. 14.] [ 5. 15.][ 6. 16.]]
 **************************
 batch_images:[b'7.jpg' b'8.jpg' b'1.jpg']
 batch_labels:[[ 7. 17.][ 8. 18.][ 1. 11.]]
 **************************
 batch_images:[b'2.jpg' b'3.jpg' b'4.jpg']
 batch_labels:[[ 2. 12.] [ 3. 13.][ 4. 14.]]
 **************************
 batch_images:[b'5.jpg' b'6.jpg' b'7.jpg']
 batch_labels:[[ 5. 15.][ 6. 16.][ 7. 17.]]
 
 
三、更加实用的方法：数据巨大的情况
 
    当数据量很大很大时，超过2T的数据时，我们不可能把所以数据都保存为一个文件，也不可能把数据都加载到内存。为了避免内存耗尽的情况，最简单的思路是：把数据分割成多个文件保存到硬盘（每个文件不超过2G），训练时，按batch大小逐个加载文件，再获取一个batch的训练数据。这种方法，也可以用TensorFlow TFRecord格式，利用队列方法读取文件，然后再产生一个batch数据，可以参考：《Tensorflow生成自己的图片数据集TFrecords》：https://blog.csdn.net/guyuealian/article/details/80857228
 
   但TensorFlow TFRecord格式存储的内容，有很多限制， 这里将实现一种类似于TensorFlow TFRecord的方法，但存储的内容没有限制，你可稍微修改保存不同的数据，基本思路是：
 
 
 
数据产生：利用numpy，把数据分割成多个*.npy文件保存到硬盘（每个文件不超过1G），当然你可以用其他Python工具保存其他文件格式，只要你能读取文件即可
获得训练数据：获取所有文件*.npy的列表，逐个读取文件的数据，并根据batch的大小，循环返回数据
 
     完整代码如下：
 
    这里将数据保存为data1.npy，data2.npy，data3.npy，其中*.npy文件的数据保存是二维矩阵Mat:第一列为样本的labels,剩余的列为样本的数据，
 
indexMat1:
[[0 0 5]
 [1 1 6]
 [2 2 7]
 [3 3 8]
 [4 4 9]]
indexMat2:
[[ 5 15 20]
 [ 6 16 21]
 [ 7 17 22]
 [ 8 18 23]
 [ 9 19 24]]
indexMat3:
[[10 30 35]
 [11 31 36]
 [12 32 37]
 [13 33 38]
 [14 34 39]]
 
# -*-coding: utf-8 -*-
"""
    @Project: nlp-learning-tutorials
    @File   : create_batch_data.py
    @Author : panjq
    @E-mail : pan_jinquan@163.com
    @Date   : 2018-11-08 09:29:19
"""

import math
import random
import os
import glob
import numpy as np
from sklearn import preprocessing


def get_data_batch(file_list,labels_nums,batch_size=None, shuffle=False,one_hot=False):
    '''
    加载*.npy文件的数据，循环产生批量数据batch，其中*.npy文件的数据保存是二维矩阵Mat:
    二维矩阵Mat:第一列为样本的labels,剩余的列为样本的数据，
    np.concatenate([label,data], axis=1)
    :param file_list: *.npy文件路径,type->list->[file0.npy,file1.npy,....]
    :param labels_nums: labels种类数
    :param batch_size: batch大小
    :param shuffle: 是否打乱数据,PS:只能打乱一个batch的数据，不同batch的数据不会干扰
    :param one_hot: 是否独热编码
    :return: 返回一个batch数据
    '''
    height = 0
    indexMat_labels = None
    i = 0
    while True:
        while height < batch_size:
            i = i%len(file_list)
            tempFile = file_list[i]
            tempMat_labels = np.load(tempFile)
            if indexMat_labels is None:
                indexMat_labels = tempMat_labels
            else:
                indexMat_labels = np.concatenate([indexMat_labels, tempMat_labels], 0)
            i=i+1
            height = indexMat_labels.shape[0]

        indices = list(range(height))
        batch_indices = np.asarray(indices[0:batch_size])  # 产生一个batch的index
        if shuffle:
            random.seed(100)
            random.shuffle(batch_indices)

        batch_indexMat_labels = indexMat_labels[batch_indices] # 使用下标查找，必须是ndarray类型类型
        indexMat_labels=np.delete(indexMat_labels,batch_indices,axis=0)
        height = indexMat_labels.shape[0]

        # 将数据分割成indexMat和labels
        batch_labels=batch_indexMat_labels[:,0]     # 第一列是labels
        batch_indexMat=batch_indexMat_labels[:,1:]  # 其余是indexMat

        # 是否进行独热编码
        if one_hot:
            batch_labels = batch_labels.reshape(len(batch_labels), 1)
            onehot_encoder = preprocessing.OneHotEncoder(sparse=False,categories=[range(labels_nums)])
            batch_labels = onehot_encoder.fit_transform(batch_labels)
        yield batch_indexMat,batch_labels


def get_next_batch(batch):
    return batch.__next__()

def get_file_list(file_dir,postfix):
    '''
    获得后缀名为postfix所有文件列表
    :param file_dir:
    :param postfix:
    :return:
    '''
    file_dir=os.path.join(file_dir,postfix)
    file_list=glob.glob(file_dir)
    return file_list


def create_test_data(out_dir):
    '''
    产生测试数据
    :return:
    '''
    data1 = np.arange(0, 10)
    data1 = np.transpose(data1.reshape([2, 5]))
    label1 = np.arange(0, 5)
    label1 = label1.reshape([5, 1])

    path1 = os.path.join(out_dir,'data1.npy')
    indexMat1 = np.concatenate([label1, data1], axis=1)  # 矩阵拼接，第一列为labels
    np.save(path1, indexMat1)

    data2 = np.arange(15, 25)
    data2 = np.transpose(data2.reshape([2, 5]))
    label2 = np.arange(5, 10)
    label2 = label2.reshape([5, 1])

    path2 = os.path.join(out_dir,'data2.npy')
    indexMat2 = np.concatenate([label2, data2], axis=1)
    np.save(path2, indexMat2)

    data3 = np.arange(30, 40)
    data3 = np.transpose(data3.reshape([2, 5]))
    label3 = np.arange(10, 15)
    label3 = label3.reshape([5, 1])

    path3 = os.path.join(out_dir,'data3.npy')
    indexMat3 = np.concatenate([label3, data3], axis=1)
    np.save(path3, indexMat3)

    print('indexMat1:\n{}'.format(indexMat1))
    print('indexMat2:\n{}'.format(indexMat2))
    print('indexMat3:\n{}'.format(indexMat3))


if __name__ == '__main__':
    out_dir='./output'
    create_test_data(out_dir)
    file_list=get_file_list(file_dir=out_dir, postfix='*.npy')
    iter = 3  # 迭代3次，每次输出一个batch个
    batch = get_data_batch(file_list, labels_nums=15,batch_size=8, shuffle=False,one_hot=False)
    for i in range(iter):
        print('**************************')
        batch_data, batch_label = get_next_batch(batch)
        print('batch_images:\n{}'.format(batch_data))
        print('batch_labels:\n{}'.format(batch_label))

 
运行结果： 
 
**************************
batch_images:
[[ 0  5]
 [ 1  6]
 [ 2  7]
 [ 3  8]
 [ 4  9]
 [15 20]
 [16 21]
 [17 22]]
batch_labels:
[0 1 2 3 4 5 6 7]
**************************
batch_images:
[[18 23]
 [19 24]
 [30 35]
 [31 36]
 [32 37]
 [33 38]
 [34 39]
 [ 0  5]]
batch_labels:
[ 8  9 10 11 12 13 14  0]
**************************
batch_images:
[[ 1  6]
 [ 2  7]
 [ 3  8]
 [ 4  9]
 [15 20]
 [16 21]
 [17 22]
 [18 23]]
batch_labels:
[1 2 3 4 5 6 7 8]
 
 

 
 
Spring Batch示例教程
 
 
 
欢迎使用Spring Batch示例。Spring Batch是一个用于执行批处理作业的弹簧框架模块。我们可以使用spring批处理来处理一系列作业。
 
目录[ 隐藏 ]
 
1 Spring Batch示例 
  
1.1 Spring Batch示例
1.2 Spring批处理示例目录结构
1.3 Spring Batch Maven依赖项
1.4 Spring批处理CSV输入文件
1.5 Spring批量作业配置
1.6 Spring批量模型类
1.7 Spring Batch FieldSetMapper
1.8弹簧批处理项目处理器
1.9 Spring配置文件
1.10 Spring批处理表
1.11弹簧批量测试程序
 
 
Spring Batch示例
 
在进行spring批处理示例程序之前，让我们先了解一下spring批处理术语。
 
工作可以包含'n'个步骤。每个步骤都包含Read-Process-Write任务，或者它可以具有单个操作，称为tasklet。
Read-Process-Write基本上是从数据库，CSV等源读取的，然后处理数据并将其写入数据库，CSV，XML等源。
Tasklet意味着执行单个任务或操作，如清理连接，在处理完成后释放资源。
Read-Process-Write和tasklet可以链接在一起以运行作业。
Spring Batch示例
 
让我们考虑一个实施弹簧批的实例。我们将考虑以下方案用于实现目的。
 
包含数据的CSV文件需要转换为XML以及数据和标记将在列名后面命名。
 
以下是用于弹簧批处理示例的重要工具和库。
 
Apache Maven 3.5.0 - 用于项目构建和依赖关系管理。
Eclipse Oxygen Release 4.7.0 - 用于创建spring批处理maven应用程序的IDE。
Java 1.8
Spring Core 4.3.12.RELEASE
Spring OXM 4.3.12.RELEASE
Spring JDBC 4.3.12.RELEASE
Spring Batch 3.0.8.RELEASE
MySQL Java Driver 5.1.25 - 基于MySQL的安装使用。这是Spring Batch元数据表所必需的。
Spring Batch示例目录结构
 
下图说明了Spring Batch示例项目中的所有组件。
 
 
 
 
Spring Batch Maven依赖项
 
下面是pom.xml文件的内容，其中包含spring batch示例项目的所有必需依赖项。
 

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>com.journaldev.spring</groupId>
	<artifactId>SpringBatchExample</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>

	<name>SpringBatchDemo</name>
	<url>http://maven.apache.org</url>

	<properties>
		<jdk.version>1.8</jdk.version>
		<spring.version>4.3.12.RELEASE</spring.version>
		<spring.batch.version>3.0.8.RELEASE</spring.batch.version>
		<mysql.driver.version>5.1.25</mysql.driver.version>
		<junit.version>4.11</junit.version>
	</properties>

	<dependencies>

		<!-- Spring Core -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-core</artifactId>
			<version>${spring.version}</version>
		</dependency>

		<!-- Spring jdbc, for database -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jdbc</artifactId>
			<version>${spring.version}</version>
		</dependency>

		<!-- Spring XML to/back object -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-oxm</artifactId>
			<version>${spring.version}</version>
		</dependency>

		<!-- MySQL database driver -->
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>${mysql.driver.version}</version>
		</dependency>

		<!-- Spring Batch dependencies -->
		<dependency>
			<groupId>org.springframework.batch</groupId>
			<artifactId>spring-batch-core</artifactId>
			<version>${spring.batch.version}</version>
		</dependency>
		<dependency>
			<groupId>org.springframework.batch</groupId>
			<artifactId>spring-batch-infrastructure</artifactId>
			<version>${spring.batch.version}</version>
		</dependency>

		<!-- Spring Batch unit test -->
		<dependency>
			<groupId>org.springframework.batch</groupId>
			<artifactId>spring-batch-test</artifactId>
			<version>${spring.batch.version}</version>
		</dependency>

		<!-- Junit -->
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>${junit.version}</version>
			<scope>test</scope>
		</dependency>

		<dependency>
			<groupId>com.thoughtworks.xstream</groupId>
			<artifactId>xstream</artifactId>
			<version>1.4.10</version>
		</dependency>

	</dependencies>
	<build>
		<finalName>spring-batch</finalName>
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-eclipse-plugin</artifactId>
				<version>2.9</version>
				<configuration>
					<downloadSources>true</downloadSources>
					<downloadJavadocs>false</downloadJavadocs>
				</configuration>
			</plugin>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>2.3.2</version>
				<configuration>
					<source>${jdk.version}</source>
					<target>${jdk.version}</target>
				</configuration>
			</plugin>
		</plugins>
	</build>
</project>
 
Spring批处理CSV输入文件
 
以下是用于弹簧批处理的示例CSV文件的内容。
 

1001,Tom,Moody, 29/7/2013
1002,John,Parker, 30/7/2013
1003,Henry,Williams, 31/7/2013
 
Spring批量作业配置
 
我们必须在配置文件中定义spring bean和spring批处理作业。下面是job-batch-demo.xml文件的内容，它是spring批处理项目中最重要的部分。
 

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:batch="http://www.springframework.org/schema/batch" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/batch
		http://www.springframework.org/schema/batch/spring-batch-3.0.xsd
		http://www.springframework.org/schema/beans 
		http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
	">

	<import resource="../config/context.xml" />
	<import resource="../config/database.xml" />

	<bean id="report" class="com.journaldev.spring.model.Report"
		scope="prototype" />
	<bean id="itemProcessor" class="com.journaldev.spring.CustomItemProcessor" />

	<batch:job id="DemoJobXMLWriter">
		<batch:step id="step1">
			<batch:tasklet>
				<batch:chunk reader="csvFileItemReader" writer="xmlItemWriter"
					processor="itemProcessor" commit-interval="10">
				</batch:chunk>
			</batch:tasklet>
		</batch:step>
	</batch:job>

	<bean id="csvFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader">

		<property name="resource" value="classpath:csv/input/report.csv" />

		<property name="lineMapper">
			<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
				<property name="lineTokenizer">
					<bean
						class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
						<property name="names" value="id,firstname,lastname,dob" />
					</bean>
				</property>
				<property name="fieldSetMapper">
					<bean class="com.journaldev.spring.ReportFieldSetMapper" />

					<!-- if no data type conversion, use BeanWrapperFieldSetMapper to map 
						by name <bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper"> 
						<property name="prototypeBeanName" value="report" /> </bean> -->
				</property>
			</bean>
		</property>

	</bean>

	<bean id="xmlItemWriter" class="org.springframework.batch.item.xml.StaxEventItemWriter">
		<property name="resource" value="file:xml/outputs/report.xml" />
		<property name="marshaller" ref="reportMarshaller" />
		<property name="rootTagName" value="report" />
	</bean>

	<bean id="reportMarshaller" class="org.springframework.oxm.jaxb.Jaxb2Marshaller">
		<property name="classesToBeBound">
			<list>
				<value>com.journaldev.spring.model.Report</value>
			</list>
		</property>
	</bean>

</beans>
 
我们正在使用FlatFileItemReader读取CSV文件，CustomItemProcessor处理数据并使用写入XML文件StaxEventItemWriter。
batch:job - 此标记定义我们要创建的作业。Id属性指定作业的ID。我们可以在一个xml文件中定义多个作业。
batch：step - 此标记用于定义spring批处理作业的不同步骤。
Spring Batch Framework提供了两种不同类型的处理方式，即“TaskletStep Oriented”和“Chunk Oriented”。在这个例子中使用的Chunk Oriented风格是指逐个读取数据并在事务边界内创建将被写出的“块”。
读者：用于读取数据的spring bean。我们csvFileItemReader在这个例子中使用了bean作为实例FlatFileItemReader。
处理器：这是用于处理数据的类。我们CustomItemProcessor在这个例子中使用过。
writer：bean用于将数据写入xml文件。
commit-interval：此属性定义处理完成后将提交的块的大小。基本上它意味着ItemReader将逐个读取数据，而ItemProcessor也将以相同的方式处理它，但ItemWriter只有在它等于commit-interval的大小时才会写入数据。
被用作该项目的一部分，是三个重要的接口ItemReader，ItemProcessor并ItemWriter从org.springframework.batch.item包。
Spring批量模型类
 
首先，我们将CSV文件读入java对象，然后使用JAXB将其写入xml文件。下面是我们的模型类，带有所需的JAXB 注释。
 
 
 

package com.journaldev.spring.model;

import java.util.Date;

import javax.xml.bind.annotation.XmlAttribute;
import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlRootElement;

@XmlRootElement(name = "record")
public class Report {

	private int id;
	private String firstName;
	private String lastName;
	private Date dob;

	@XmlAttribute(name = "id")
	public int getId() {
		return id;
	}

	public void setId(int id) {
		this.id = id;
	}

	@XmlElement(name = "firstname")
	public String getFirstName() {
		return firstName;
	}

	public void setFirstName(String firstName) {
		this.firstName = firstName;
	}

	@XmlElement(name = "lastname")
	public String getLastName() {
		return lastName;
	}

	public void setLastName(String lastName) {
		this.lastName = lastName;
	}

	@XmlElement(name = "dob")
	public Date getDob() {
		return dob;
	}

	public void setDob(Date dob) {
		this.dob = dob;
	}

	@Override
	public String toString() {
		return "Report [id=" + id + ", firstname=" + firstName + ", lastName=" + lastName + ", DateOfBirth=" + dob
				+ "]";
	}

}
 
请注意，模型类字段应与spring批处理映射器配置中定义的相同，即property name="names" value="id,firstname,lastname,dob"在我们的示例中。
 
Spring Batch FieldSetMapper
 
FieldSetMapper转换日期需要自定义。如果不需要数据类型转换，则只BeanWrapperFieldSetMapper应使用自动按名称映射值。
 
扩展的java类FieldSetMapper是ReportFieldSetMapper。
 

package com.journaldev.spring;

import java.text.ParseException;
import java.text.SimpleDateFormat;

import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.validation.BindException;

import com.journaldev.spring.model.Report;

public class ReportFieldSetMapper implements FieldSetMapper<Report> {

	private SimpleDateFormat dateFormat = new SimpleDateFormat("dd/MM/yyyy");

	public Report mapFieldSet(FieldSet fieldSet) throws BindException {

		Report report = new Report();
		report.setId(fieldSet.readInt(0));
		report.setFirstName(fieldSet.readString(1));
		report.setLastName(fieldSet.readString(2));

		// default format yyyy-MM-dd
		// fieldSet.readDate(4);
		String date = fieldSet.readString(3);
		try {
			report.setDob(dateFormat.parse(date));
		} catch (ParseException e) {
			e.printStackTrace();
		}

		return report;

	}

}
 
Spring Batch Item Processor
 
现在，在作业配置中定义，itemProcessor将在itemWriter之前触发。我们CustomItemProcessor.java为此创建了一个类。
 

package com.journaldev.spring;

import org.springframework.batch.item.ItemProcessor;

import com.journaldev.spring.model.Report;

public class CustomItemProcessor implements ItemProcessor<Report, Report> {

	public Report process(Report item) throws Exception {
		
		System.out.println("Processing..." + item);
		String fname = item.getFirstName();
		String lname = item.getLastName();
		
		item.setFirstName(fname.toUpperCase());
		item.setLastName(lname.toUpperCase());
		return item;
	}

}
 
我们可以在ItemProcessor实现中操作数据，因为您可以看到我将名字和姓氏值转换为大写。
 
弹簧配置文件
 
在我们的spring批量配置文件中，我们已经导入了两个额外的配置文件 - context.xml和database.xml。
 

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
		http://www.springframework.org/schema/beans 
		http://www.springframework.org/schema/beans/spring-beans-4.3.xsd">

	<!-- stored job-meta in memory -->
	<!--  
	<bean id="jobRepository"
		class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
		<property name="transactionManager" ref="transactionManager" />
	</bean>
 	 -->
 	 
 	 <!-- stored job-meta in database -->
	<bean id="jobRepository"
		class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
		<property name="dataSource" ref="dataSource" />
		<property name="transactionManager" ref="transactionManager" />
		<property name="databaseType" value="mysql" />
	</bean>
	
	<bean id="transactionManager"
		class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
	 
	<bean id="jobLauncher"
		class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
		<property name="jobRepository" ref="jobRepository" />
	</bean>

</beans>
 
jobRepository - JobRepository负责将每个Java对象存储到弹簧批处理的正确元数据表中。
transactionManager - 一旦commit-interval的大小和处理的数据相等，它负责提交事务。
jobLauncher - 这是春季批次的核心。此接口包含用于触发作业的run方法。

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/beans 
		http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
		http://www.springframework.org/schema/jdbc 
		http://www.springframework.org/schema/jdbc/spring-jdbc-4.3.xsd">

	<!-- connect to database -->
	<bean id="dataSource"
		class="org.springframework.jdbc.datasource.DriverManagerDataSource">
		<property name="driverClassName" value="com.mysql.jdbc.Driver" />
		<property name="url" value="jdbc:mysql://localhost:3306/Test" />
		<property name="username" value="test" />
		<property name="password" value="test123" />
	</bean>

	<bean id="transactionManager"
		class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />

	<!-- create job-meta tables automatically -->
	<!-- <jdbc:initialize-database data-source="dataSource"> <jdbc:script location="org/springframework/batch/core/schema-drop-mysql.sql" 
		/> <jdbc:script location="org/springframework/batch/core/schema-mysql.sql" 
		/> </jdbc:initialize-database> -->
</beans>
 
Spring Batch使用一些元数据表来存储批处理作业信息。我们可以从Spring批处理配置中创建它们，但建议通过执行SQL文件手动完成它们，如上面的注释代码所示。从安全的角度来看，最好不要给Spring批处理数据库用户提供DDL执行权限。
 
Spring批处理表
 
Spring Batch表与Java中表示它们的Domain对象非常匹配。例如 - JobInstance，JobExecution，JobParameters和StepExecution地图BATCH_JOB_INSTANCE，BATCH_JOB_EXECUTION，分别BATCH_JOB_EXECUTION_PARAMS和BATCH_STEP_EXECUTION。
 
ExecutionContext映射到BATCH_JOB_EXECUTION_CONTEXT和BATCH_STEP_EXECUTION_CONTEXT。
 
JobRepository负责将每个java对象保存并存储到正确的表中。
  
 
 

以下是每个元数据表的详细信息。
 
Batch_job_instance：BATCH_JOB_INSTANCE表包含与JobInstance相关的所有信息。
Batch_job_execution_params：BATCH_JOB_EXECUTION_PARAMS表包含与JobParameters对象相关的所有信息。
Batch_job_execution：BATCH_JOB_EXECUTION表保存与JobExecution对象相关的数据。每次运行作业时都会添加一个新行。
Batch_step_execution：BATCH_STEP_EXECUTION表包含与StepExecution对象相关的所有信息。
Batch_job_execution_context：BATCH_JOB_EXECUTION_CONTEXT表保存与Job的ExecutionContext相关的数据。每个JobExecution都只有一个Job ExecutionContext，它包含该特定作业执行所需的所有作业级数据。此数据通常表示在发生故障后必须检索的状态，以便JobInstance可以从失败的位置重新启动。
Batch_step_execution_context：BATCH_STEP_EXECUTION_CONTEXT表保存与Step的ExecutionContext相关的数据。每个StepExecution都只有一个ExecutionContext，它包含了特定步骤执行所需的所有数据。此数据通常表示在发生故障后必须检索的状态，以便JobInstance可以从失败的位置重新启动。
Batch_job_execution_seq：该表保存作业的数据执行顺序。
Batch_step_execution_seq：此表保存步骤执行序列的数据。
Batch_job_seq：此表保存作业序列的数据，以防我们有多个作业，我们将得到多行。
春季批量测试计划
 
我们的Spring Batch示例项目已准备就绪，最后一步是编写一个测试类来将其作为java程序执行。
 

package com.journaldev.spring;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class App {
	public static void main(String[] args) {

		String[] springConfig = { "spring/batch/jobs/job-batch-demo.xml" };

		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(springConfig);

		JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
		Job job = (Job) context.getBean("DemoJobXMLWriter");

		JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis())
				.toJobParameters();

		try {

			JobExecution execution = jobLauncher.run(job, jobParameters);
			System.out.println("Exit Status : " + execution.getStatus());

		} catch (Exception e) {
			e.printStackTrace();
		}

		System.out.println("Done");
		context.close();
	}
}
 
只需运行上面的程序，您将获得如下所示的输出xml。
 
 
 
Copy
 
<?xml version="1.0" encoding="UTF-8"?><report><record id="1001"><dob>2013-07-29T00:00:00+05:30</dob><firstname>TOM</firstname><lastname>MOODY</lastname></record><record id="1002"><dob>2013-07-30T00:00:00+05:30</dob><firstname>JOHN</firstname><lastname>PARKER</lastname></record><record id="1003"><dob>2013-07-31T00:00:00+05:30</dob><firstname>HENRY</firstname><lastname>WILLIAMS</lastname></record></report> 
 
这就是Spring Batch示例，您可以从下面的链接下载最终项目。
 
下载Spring Batch示例项目
 
参考：官方指南
 
 
 
转载来源：https://www.journaldev.com/17157/spring-batch-example

只做记录，原文：https://blog.csdn.net/wydbyxr/article/details/84855489
 尽可能地设大batchsize
 
在合理范围内，增大batch size的好处
 
1）内存利用率提高了，大矩阵乘法的并行化效率提高。
 2）跑完一次 epoch（全数据集）所需的迭代次数减少，对于相同数据量的处理速度进一步加快。
 3）在一定范围内，一般来说 Batch_Size 越大，其确定的下降方向越准，引起训练震荡越小。
 随着 Batch_Size 增大，处理相同数据量的速度越快。
 随着 Batch_Size 增大，达到相同精度所需要的 epoch 数量越来越多。" 由于最终收敛精度会陷入不同的局部极值，因此 Batch_Size 增大到某些时候，达到最终收敛精度上的最优。
 
盲目增大 Batch_Size 有何坏处
 
1）Batch_Size 增大到一定程度，其确定的下降方向已经基本不再变化。
 2）太大的batch size 容易陷入sharp minima，泛化性不好。
 3）Batch_Size 太小 收敛太慢，算法在 200 epoch 内不收敛。
 
不同的梯度下降算法下，不同Batch_Size设置的影响
 
样本量少的时候会带来很大的方差，而这个大方差恰好会导致我们在梯度下降到很差的局部最优点（只是微微凸下去的最优点）和鞍点的时候不稳定，一不小心就因为一个大噪声的到来导致炸出了局部最优点，从而有机会去寻找更优的最优点。
 因此，与之相反的，当样本量很多时，方差很小，对梯度的估计要准确和稳定的多，因此反而在差劲的局部最优点和鞍点时反而容易自信的呆着不走了，从而导致神经网络收敛到很差的点上，跟出了bug一样的差劲。
 总结一下，batch的size设置的不能太大也不能太小，因此实际工程中最常用的就是mini-batch，一般size设置为几十或者几百。
 使用更大的方差来试探最优点的质量。
 所以在小batch size时，逐渐减少learning rate的神经网络玩的就是退火算法。
 不同的batch size不仅仅会影响你的收敛速度，还经常影响你最终收敛时的准确率。
 工程上实验验证 batch size 越大，比如1000 对 100的效果要差。即发现batch 太多大， 训练一点效果没。
 
这之前我们的讨论是基于梯度下降的，而且默认是一阶的（即没有利用二阶导数信息，仅仅使用一阶导数去优化）。因此对于SGD（随机梯度下降）及其改良的一阶优化算法如Adagrad、Adam等是没问题的，但是对于强大的二阶优化算法如共轭梯度法、L-BFGS来说，如果估计不好一阶导数，那么对二阶导数的估计会有更大的误差，这对于这些算法来说是致命的。
 因此，对于二阶优化算法，减小batch换来的收敛速度提升远不如引入大量噪声导致的性能下降，因此在使用二阶优化算法时，往往要采用大batch哦。此时往往batch设置成几千甚至一两万才能发挥出最佳性能。
 
batch_size与2的倍数
 
一般而言，根据你的GPU显存，设置为最大，而且一般要求是8的倍数（比如32，128），这样，GPU内部的并行运算效率最高。
 那么怎么选择batch number呢？就像刚才说的，8的倍数，然后是稍微大一点（一般而言）。另外一个方法，就是选择一部分数据，跑几个batch看看你的loss是不是在变小,选择一个合适的就可以了。
 cpu有预取, gpu有合并访问, 不仅仅要求长度是2的次方, 内存地址也是有要求的。
 
另外，听说GPU对2的幂次的batch可以发挥更佳的性能，因此设置成16、32、64、128…时往往要比设置为整10、整100的倍数时表现更优。
 batch太大容易错事最有点，太小的话又会出现在局部最小点附近剃度下降特别慢，所以很多架构都有一个decay参数，在前期使用大的batch。
 
cpu是非常讨厌16，32，64…… 这样大小的数组的，gpu没有类似的问题（gpu 上最好都是 32 倍数，和 wrap 一致，可见cuda gpu的文档），但我还是要劝大家，超参的选取随意点。
 
batch size与BN
 
batch size大了则一个epoch update数量少的这个缺点，在bn面前似乎也没太多存在感了。
 不过bn的坏处就是不能用太小的batch size，要不然mean和variance就偏了。所以现在一般是显存能放多少就放多少。而且实际调起模型来，真的是数据分布和预处理更为重要，数据不行的话 玩再多花招也没用。
 如果用了batchnorm，batch size别太小（BN 大法虽然好，但是也存在一些局限和问题，诸如当 BatchSize 太小时效果不佳、对 RNN 等动态网络无法有效应用 BN 等）。
 
设置多大的batch_size(未整理，凑合着看吧)
 
Large Batch
 Large Batch在 keypoints上也是有用的，大概能提到0.4到0.3的点的AP。
 在利用 ImageNet 训练 AlexNet 模型时，其中每 GPU 的最优批量大小为 512。如果我们希望使用大量 GPU 并保证每 GPU 皆拥有理想的执行效率，则应当将批量大小设定为 16 x 512 = 8192。
 
Train的batch是多少？
 我们训练的时候根据输入图像的尺寸不同，batch_size在20到64之间。
 
SGD 算法的 batch size 并不能无限制地增大。
 SGD 采用较大的 batch size 时，如果还是使用同样的 epochs 数量进行运算，则准确度往往低于 batch size 较小的场景 ; 而且目前还不存在特定算法方案能够帮助我们高效利用较大的 batch size。
 
对于大批量训练场景，我们需要确保使用较大的 batch size 训练能够在相同 epochs 前提下获得与较小的 batch size 相近的测试准确度。这里我们之所以保持 epochs 数量不变，是因为从统计学角度来讲，一个 epoch 代表着算法与整体数据集接触一次 ; 而从计算角度来讲，固定 epochs 数量意味着保证浮点运算次数不变。
 
Batch_Size 既然 Full Batch Learning 并不适用大数据集，那么走向另一个极端怎么样？所谓另一个极端，就是每次只训练一个样本，即 Batch_Size = 1。这就是在线学习（Online Learning）。线性神经元在均方误差代价函数的错误面是一个抛物面，横截面是椭圆。对于多层神经元、非线性网络，在局部依然近似是抛物面。使用在线学习，每次修正方向以各自样本的梯度方向修正，横冲直撞各自为政，难以达到收敛。
 
batch_size设的大一些，收敛得快，也就是需要训练的次数少，准确率上升得也很稳定，但是实际使用起来精度不高。
 batch_size设的小一些，收敛得慢，而且可能准确率来回震荡，所以还要把基础学习速率降低一些；但是实际使用起来精度较高。一般我只尝试batch_size=64或者batch_size=1两种情况。
 Batch Size 的影响，目前可以实验证实的是：batch size 设置得较小训练出来的模型相对大 batch size 训练出的模型泛化能力更强，在测试集上的表现更好，而太大的 batch size 往往不太 Work，而且泛化能力较差。但是背后是什么原因造成的，目前还未有定论，持不同看法者各持己见。
 
Batch size设置以喂饱你的硬件为主要标准。只要显卡塞得下，首先挑大的。
 当感觉训练时噪音不够时，比如收敛碰到鞍点或者局部最小值时，调小batch size。（很少会碰到）
 当感觉训练时噪音太大时，调大batch size到喂饱硬件（因为2很少碰到，这也很少做），再不行就调小learning rate，也可以直接调小learning rate。
 综合起来用就是常见的带learning rate下降的sgd。开始时依赖batch带来的噪音快速下降，接下来使用较低的learning rate消除这些噪音寻求稳定收敛。一般而言只要batch不太大，样本里的噪音总是够用的。
 full batch是特殊的mini batch。
 batch size越大每次迭代的梯度越接近样本的真实梯度。
 小的batch size能有效避免样本中的冗余信息。
 大的batch能更充分利用GPU。
 batch的大小并没有固定模式，不需要非要2^n的size。
 综合以上几点进行权衡，选择你的batch。
 
就个人经验而论 batchsize越大越好1.震荡明显减少 2.收敛速度加快 3.同样学习率下大batch可以收敛到更好的水平。目前没发现什么大batch过早陷入局部最优的现象，在我这都是batch越大，精度越高，当然我还在继续实验观察
 
实验 这个做过几个实验，但是没有详细的做，主要是针对googlenet，alexnet以及vgg几个模型（实验结果就是batch为32的时候，alex开始收敛，但是googlenet不收敛；提高batch size，googlenet开始收敛）。就像lecunnote里说的一样，随机梯度下降不能像full batch那样明显的保证收敛。
 
可能因为我调参的能力比较有限，确实batchsize太小会出现网络收敛不稳定，最后结果比较差的情况，这个在ImageNet和其他数据库上都遇到过，而batchsize太大确实也会影响随机性的引入。目前一般调ImageNet的时候，大家都喜欢把显存占满，不过小一些的库，个人感觉还是应该大大小小都尝试一下。