Table of Contents

spring batch简介

Spring Batch架构介绍

Spring Batch核心概念介绍

什么是Job

什么是JobInstance

什么是JobParameters

 什么是JobExecution

什么是Step

什么是StepExecution

什么是ExecutionContext

 什么是JobRepository

什么是JobLauncher

什么是Item Reader

什么是Item Writer

什么是Item Processor

chunk 处理流程

skip策略和失败处理

批处理操作指南

批处理原则

如何默认不启动job

在读数据时内存不够

 

spring batch简介

spring batch是spring提供的一个数据处理框架。企业域中的许多应用程序需要批量处理才能在关键任务环境中执行业务操作。 这些业务运营包括：

无需用户交互即可最有效地处理大量信息的自动化，复杂处理。 这些操作通常包括基于时间的事件（例如月末计算，通知或通信）。
	
在非常大的数据集中重复处理复杂业务规则的定期应用（例如，保险利益确定或费率调整）。
	
集成从内部和外部系统接收的信息，这些信息通常需要以事务方式格式化，验证和处理到记录系统中。 批处理用于每天为企业处理数十亿的交易。
Spring Batch是一个轻量级，全面的批处理框架，旨在开发对企业系统日常运营至关重要的强大批处理应用程序。 Spring Batch构建了人们期望的Spring Framework特性（生产力，基于POJO的开发方法和一般易用性），同时使开发人员可以在必要时轻松访问和利用更高级的企业服务。 Spring Batch不是一个schuedling的框架。

Spring Batch提供了可重用的功能，这些功能对于处理大量的数据至关重要，包括记录/跟踪，事务管理，作业处理统计，作业重启，跳过和资源管理。 它还提供更高级的技术服务和功能，通过优化和分区技术实现极高容量和高性能的批处理作业。 Spring Batch可用于两种简单的用例（例如将文件读入数据库或运行存储过程）以及复杂的大量用例（例如在数据库之间移动大量数据，转换它等等） 上）。 大批量批处理作业可以高度可扩展的方式利用该框架来处理大量信息。

Spring Batch架构介绍

一个典型的批处理应用程序大致如下：

从数据库，文件或队列中读取大量记录。
	
以某种方式处理数据。
	
以修改之后的形式写回数据。
其对应的示意图如下：



 

spring batch的一个总体的架构如下：



在spring batch中一个job可以定义很多的步骤step，在每一个step里面可以定义其专属的ItemReader用于读取数据，ItemProcesseor用于处理数据，ItemWriter用于写数据，而每一个定义的job则都在JobRepository里面，我们可以通过JobLauncher来启动某一个job。

Spring Batch核心概念介绍

下面是一些概念是Spring batch框架中的核心概念。

什么是Job

Job和Step是spring batch执行批处理任务最为核心的两个概念。

其中Job是一个封装整个批处理过程的一个概念。Job在spring batch的体系当中只是一个最顶层的一个抽象概念，体现在代码当中则它只是一个最上层的接口，其代码如下:


/**
 * Batch domain object representing a job. Job is an explicit abstraction
 * representing the configuration of a job specified by a developer. It should
 * be noted that restart policy is applied to the job as a whole and not to a
 * step.
 */
public interface Job {

	String getName();


	boolean isRestartable();


	void execute(JobExecution execution);


	JobParametersIncrementer getJobParametersIncrementer();


	JobParametersValidator getJobParametersValidator();

}

在Job这个接口当中定义了五个方法，它的实现类主要有两种类型的job，一个是simplejob，另一个是flowjob。在spring batch当中，job是最顶层的抽象，除job之外我们还有JobInstance以及JobExecution这两个更加底层的抽象。

一个job是我们运行的基本单位，它内部由step组成。job本质上可以看成step的一个容器。一个job可以按照指定的逻辑顺序组合step，并提供了我们给所有step设置相同属性的方法，例如一些事件监听，跳过策略。

Spring Batch以SimpleJob类的形式提供了Job接口的默认简单实现，它在Job之上创建了一些标准功能。一个使用java config的例子代码如下：

@Bean
public Job footballJob() {
    return this.jobBuilderFactory.get("footballJob")
                     .start(playerLoad())
                     .next(gameLoad())
                     .next(playerSummarization())
                     .end()
                     .build();
}

这个配置的意思是：首先给这个job起了一个名字叫footballJob，接着指定了这个job的三个step，他们分别由方法，playerLoad,gameLoad, playerSummarization实现。

什么是JobInstance

我们在上文已经提到了JobInstance，他是Job的更加底层的一个抽象，他的定义如下：

public interface JobInstance {
	/**
	 * Get unique id for this JobInstance.
	 * @return instance id
	 */
	public long getInstanceId();
	/**
	 * Get job name.
	 * @return value of 'id' attribute from <job>
	 */
	public String getJobName();	
}


他的方法很简单，一个是返回Job的id，另一个是返回Job的名字。

JobInstance指的是job运行当中，作业执行过程当中的概念。Instance本就是实例的意思。

比如说现在有一个批处理的job，它的功能是在一天结束时执行行一次。我们假定这个批处理job的名字为'EndOfDay'。在这个情况下，那么每天就会有一个逻辑意义上的JobInstance, 而我们必须记录job的每次运行的情况。

什么是JobParameters

在上文当中我们提到了，同一个job每天运行一次的话，那么每天都有一个jobIntsance，但他们的job定义都是一样的，那么我们怎么来区别一个job的不同jobinstance了。 不妨先做个猜想，虽然jobinstance的job定义一样，但是他们有的东西就不一样，例如运行时间。

spring batch中提供的用来标识一个jobinstance的东西是：JobParameters。 JobParameters对象包含一组用于启动批处理作业的参数，它可以在运行期间用于识别或甚至用作参考数据。我们假设的运行时间，就可以作为一个JobParameters。

例如, 我们前面的'EndOfDay'的job现在已经有了两个实例，一个产生于1月1日，另一个产生于1月2日，那么我们就可以定义两个JobParameter对象：一个的参数是01-01, 另一个的参数是01-02。 因此，识别一个JobInstance的方法可以定义为：



因此，我么可以通过Jobparameter来操作正确的JobInstance

 什么是JobExecution

JobExecution指的是单次尝试运行一个我们定义好的Job的代码层面的概念。 job的一次执行可能以失败也可能成功。只有当执行成功完成时，给定的与执行相对应的JobInstance才也被视为完成。

还是以前面描述的EndOfDay的job作为示例，假设第一次运行01-01-2019的JobInstance结果是失败。 那么此时如果使用与第一次运行相同的Jobparameter参数（即01-01-2019）作业参数再次运行，那么就会创建一个对应于之前jobInstance的一个新的JobExecution实例,JobInstance仍然只有一个。

JobExecution的接口定义如下：

public interface JobExecution {
	/**
	 * Get unique id for this JobExecution.
	 * @return execution id
	 */
	public long getExecutionId();
	/**
	 * Get job name.
	 * @return value of 'id' attribute from <job>
	 */
	public String getJobName(); 
	/**
	 * Get batch status of this execution.
	 * @return batch status value.
	 */
	public BatchStatus getBatchStatus();
	/**
	 * Get time execution entered STARTED status. 
	 * @return date (time)
	 */
	public Date getStartTime();
	/**
	 * Get time execution entered end status: COMPLETED, STOPPED, FAILED 
	 * @return date (time)
	 */
	public Date getEndTime();
	/**
	 * Get execution exit status.
	 * @return exit status.
	 */
	public String getExitStatus();
	/**
	 * Get time execution was created.
	 * @return date (time)
	 */
	public Date getCreateTime();
	/**
	 * Get time execution was last updated updated.
	 * @return date (time)
	 */
	public Date getLastUpdatedTime();
	/**
	 * Get job parameters for this execution.
	 * @return job parameters  
	 */
	public Properties getJobParameters();
	
}


每一个方法的注释已经解释的很清楚，这里不再多做解释。只提一下BatchStatus，JobExecution当中提供了一个方法getBatchStatus用于获取一个job某一次特地执行的一个状态。BatchStatus是一个代表job状态的枚举类，其定义如下：

public enum BatchStatus {STARTING, STARTED, STOPPING, 
			STOPPED, FAILED, COMPLETED, ABANDONED }

这些属性对于一个job的执行来说是非常关键的信息，并且spring batch会将他们持久到数据库当中. 在使用Spring batch的过程当中spring batch会自动创建一些表用于存储一些job相关的信息，用于存储JobExecution的表为batch_job_execution,下面是一个从数据库当中截图的实例：



什么是Step

每一个Step对象都封装了批处理作业的一个独立的阶段。 事实上，每一个Job本质上都是由一个或多个步骤组成。 每一个step包含定义和控制实际批处理所需的所有信息。 任何特定的内容都由编写Job的开发人员自行决定。 一个step可以非常简单也可以非常复杂。 例如，一个step的功能是将文件中的数据加载到数据库中，那么基于现在spring batch的支持则几乎不需要写代码。 更复杂的step可能具有复杂的业务逻辑，这些逻辑作为处理的一部分。 与Job一样，Step具有与JobExecution类似的StepExecution，如下图所示：



 

什么是StepExecution

StepExecution表示一次执行Step, 每次运行一个Step时都会创建一个新的StepExecution，类似于JobExecution。 但是，某个步骤可能由于其之前的步骤失败而无法执行。 且仅当Step实际启动时才会创建StepExecution。

一次step执行的实例由StepExecution类的对象表示。 每个StepExecution都包含对其相应步骤的引用以及JobExecution和事务相关的数据，例如提交和回滚计数以及开始和结束时间。 此外，每个步骤执行都包含一个ExecutionContext，其中包含开发人员需要在批处理运行中保留的任何数据，例如重新启动所需的统计信息或状态信息。下面是一个从数据库当中截图的实例：



什么是ExecutionContext

ExecutionContext即每一个StepExecution 的执行环境。它包含一系列的键值对。我们可以用如下代码获取ExecutionContext

ExecutionContext ecStep = stepExecution.getExecutionContext();
ExecutionContext ecJob = jobExecution.getExecutionContext();

 什么是JobRepository

JobRepository是一个用于将上述job，step等概念进行持久化的一个类。 它同时给Job和Step以及下文会提到的JobLauncher实现提供CRUD操作。 首次启动Job时，将从repository中获取JobExecution，并且在执行批处理的过程中，StepExecution和JobExecution将被存储到repository当中。

@EnableBatchProcessing注解可以为JobRepository提供自动配置。

什么是JobLauncher

JobLauncher这个接口的功能非常简单，它是用于启动指定了JobParameters的Job，为什么这里要强调指定了JobParameter，原因其实我们在前面已经提到了，jobparameter和job一起才能组成一次job的执行。下面是代码实例：

public interface JobLauncher {

public JobExecution run(Job job, JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException,
                   JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

上面run方法实现的功能是根据传入的job以及jobparamaters从JobRepository获取一个JobExecution并执行Job。

什么是Item Reader

ItemReader是一个读数据的抽象，它的功能是为每一个Step提供数据输入。 当ItemReader以及读完所有数据时，它会返回null来告诉后续操作数据已经读完。Spring Batch为ItemReader提供了非常多的有用的实现类，比如JdbcPagingItemReader，JdbcCursorItemReader等等。

ItemReader支持的读入的数据源也是非常丰富的，包括各种类型的数据库，文件，数据流，等等。几乎涵盖了我们的所有场景。

下面是一个JdbcPagingItemReader的例子代码：

@Bean
public JdbcPagingItemReader itemReader(DataSource dataSource, PagingQueryProvider queryProvider) {
        Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("status", "NEW");

        return new JdbcPagingItemReaderBuilder<CustomerCredit>()
                                           .name("creditReader")
                                           .dataSource(dataSource)
                                           .queryProvider(queryProvider)
                                           .parameterValues(parameterValues)
                                           .rowMapper(customerCreditMapper())
                                           .pageSize(1000)
                                           .build();
}

@Bean
public SqlPagingQueryProviderFactoryBean queryProvider() {
        SqlPagingQueryProviderFactoryBean provider = new SqlPagingQueryProviderFactoryBean();

        provider.setSelectClause("select id, name, credit");
        provider.setFromClause("from customer");
        provider.setWhereClause("where status=:status");
        provider.setSortKey("id");

        return provider;
}

JdbcPagingItemReader必须指定一个PagingQueryProvider，负责提供SQL查询语句来按分页返回数据。

下面是一个JdbcCursorItemReader的例子代码:

 private JdbcCursorItemReader<Map<String, Object>> buildItemReader(final DataSource dataSource, String tableName,
            String tenant) {

        JdbcCursorItemReader<Map<String, Object>> itemReader = new JdbcCursorItemReader<>();
        itemReader.setDataSource(dataSource);
        itemReader.setSql("sql here");
        itemReader.setRowMapper(new RowMapper());
        return itemReader;
    }

 

什么是Item Writer

既然ItemReader是读数据的一个抽象，那么ItemWriter自然就是一个写数据的抽象，它是为每一个step提供数据写出的功能。写的单位是可以配置的，我们可以一次写一条数据，也可以一次写一个chunk的数据，关于chunk下文会有专门的介绍。ItemWriter对于读入的数据是不能做任何操作的。

Spring Batch为ItemWriter也提供了非常多的有用的实现类，当然我们也可以去实现自己的writer功能。

什么是Item Processor

ItemProcessor对项目的业务逻辑处理的一个抽象, 当ItemReader读取到一条记录之后，ItemWriter还未写入这条记录之前，I我们可以借助temProcessor提供一个处理业务逻辑的功能，并对数据进行相应操作。如果我们在ItemProcessor发现一条数据不应该被写入，可以通过返回null来表示。ItemProcessor和ItemReader以及ItemWriter可以非常好的结合在一起工作，他们之间的数据传输也非常方便。我们直接使用即可。

 

chunk 处理流程

spring batch提供了让我们按照chunk处理数据的能力，一个chunk的示意图如下：



它的意思就和图示的一样，由于我们一次batch的任务可能会有很多的数据读写操作，因此一条一条的处理并向数据库提交的话效率不会很高，因此spring batch提供了chunk这个概念，我们可以设定一个chunk size，spring batch 将一条一条处理数据，但不提交到数据库，只有当处理的数据数量达到chunk size设定的值得时候，才一起去commit.

java的实例定义代码如下：



在上面这个step里面，chunk size被设为了10，当ItemReader读的数据数量达到10的时候，这一批次的数据就一起被传到itemWriter，同时transaction被提交。

skip策略和失败处理

一个batch的job的step，可能会处理非常大数量的数据，难免会遇到出错的情况，出错的情况虽出现的概率较小，但是我们不得不考虑这些情况，因为我们做数据迁移最重要的是要保证数据的最终一致性。spring batch当然也考虑到了这种情况，并且为我们提供了相关的技术支持，请看如下bean的配置：



我们需要留意这三个方法，分别是skipLimit(),skip(),noSkip(),

skipLimit方法的意思是我们可以设定一个我们允许的这个step可以跳过的异常数量，假如我们设定为10，则当这个step运行时，只要出现的异常数目不超过10，整个step都不会fail。注意，若不设定skipLimit，则其默认值是0.

skip方法我们可以指定我们可以跳过的异常，因为有些异常的出现，我们是可以忽略的。

noSkip方法的意思则是指出现这个异常我们不想跳过，也就是从skip的所以exception当中排除这个exception，从上面的例子来说，也就是跳过所有除FileNotFoundException的exception。那么对于这个step来说，FileNotFoundException就是一个fatal的exception，抛出这个exception的时候step就会直接fail

批处理操作指南

本部分是一些使用spring batch时的值得注意的点

批处理原则

在构建批处理解决方案时，应考虑以下关键原则和注意事项。

批处理体系结构通常会影响体系结构
	
尽可能简化并避免在单批应用程序中构建复杂的逻辑结构
	
保持数据的处理和存储在物理上靠得很近（换句话说，将数据保存在处理过程中）。
	
最大限度地减少系统资源的使用，尤其是I / O. 在internal memory中执行尽可能多的操作。
	
查看应用程序I / O（分析SQL语句）以确保避免不必要的物理I / O. 特别是，需要寻找以下四个常见缺陷：
	

		
当数据可以被读取一次并缓存或保存在工作存储中时，读取每个事务的数据。
		
		

		
重新读取先前在同一事务中读取数据的事务的数据。
		
		

		
导致不必要的表或索引扫描。
		
		

		
未在SQL语句的WHERE子句中指定键值。
		
	
	

	
在批处理运行中不要做两次一样的事情。 例如，如果需要数据汇总以用于报告目的，则应该（如果可能）在最初处理数据时递增存储的总计，因此您的报告应用程序不必重新处理相同的数据。
	
	

	
在批处理应用程序开始时分配足够的内存，以避免在此过程中进行耗时的重新分配。
	
	

	
总是假设数据完整性最差。 插入适当的检查和记录验证以维护数据完整性。
	
	

	
尽可能实施校验和以进行内部验证。 例如，对于一个文件里的数据应该有一个数据条数纪录，告诉文件中的记录总数以及关键字段的汇总。
	
	

	
在具有真实数据量的类似生产环境中尽早计划和执行压力测试。
	
	

	
在大批量系统中，数据备份可能具有挑战性，特别是如果系统以24-7在线的情况运行。 数据库备份通常在在线设计中得到很好的处理，但文件备份应该被视为同样重要。 如果系统依赖于文件，则文件备份过程不仅应该到位并记录在案，还应定期进行测试。
	
如何默认不启动job

在使用java config使用spring batch的job时，如果不做任何配置，项目在启动时就会默认去跑我们定义好的批处理job。那么如何让项目在启动时不自动去跑job呢？

spring batch的job会在项目启动时自动run，如果我们不想让他在启动时run的话，可以在application.properties中添加如下属性：

spring.batch.job.enabled=false


在读数据时内存不够

在使用spring batch做数据迁移时，发现在job启动后，执行到一定时间点时就卡在一个地方不动了，且log也不再打印，等待一段时间之后，得到如下错误：



红字的信息为：Resource exhaustion event：the JVM was unable to allocate memory from the heap.

翻译过来的意思就是项目发出了一个资源耗尽的事件，告诉我们java虚拟机无法再为堆分配内存。

造成这个错误的原因是: 这个项目里的batch job的reader是一次性拿回了数据库里的所有数据，并没有进行分页，当这个数据量太大时，就会导致内存不够用。解决的办法有两个:

调整reader读数据逻辑，按分页读取，但实现上会麻烦一些，且运行效率会下降
	
增大service内存
更多信息请参考博客：spring batch使用reader读数据的内存容量问题

 

 

/* 版权声明：可以任意转载，转载时请标明文章原始出处和作者信息 .*/

 

                                                     author: 张俊林

 

 

                                     （想更系统地学习深度学习知识？请参考：深度学习枕边书）

Batch Normalization作为最近一年来DL的重要成果，已经广泛被证明其有效性和重要性。目前几乎已经成为DL的标配了，任何有志于学习DL的同学们朋友们雷迪斯俺的詹特曼们都应该好好学一学BN。BN倒过来看就是NB，因为这个技术确实很NB，虽然有些细节处理还解释不清其理论原因，但是实践证明好用才是真的好，别忘了DL从Hinton对深层网络做Pre-Train开始就是一个经验领先于理论分析的偏经验的一门学问。

 

如何理解BatchNorm?请参考论文：Batch Normalization: Accelerating Deep Network Training by Reducing Internal Covariate Shift。因为对于部分基础不是太好的同学们朋友们雷迪斯俺的詹特曼们可能阅读理解有一定障碍，所以本文是为了更容易理解BN而做的一番导读。由于本人水平也很有限，假设导游导错了了路，那么…….您就认倒霉好了，毕竟这是免费的导游您说是不，期望别太高，“任何对其它人或者事物报以极高期望的人是这个世界上最不幸福的人”，这是出自我的非名人的名言，所以“降低期望是通向幸福之路”，这也是我的名言。

 

机器学习领域有个很重要的假设：IID独立同分布假设，就是假设训练数据和测试数据是满足相同分布的，这是通过训练数据获得的模型能够在测试集获得好的效果的一个基本保障。而BatchNorm是干啥的呢？BatchNorm就是在深度神经网络训练过程中使得每一层神经网络的输入保持相同分布的。OK，BN讲完了，再见。

 

嗯，这么讲步子迈得有点大，我们放慢脚步，把学习率调小一点，一步一步接近理解BN的最优解。

 

为什么深度神经网络随着网络深度加深，训练起来越困难，收敛越来越慢？这是个在DL领域很接近本质的好问题。很多论文都是解决这个问题的，比如ReLU激活函数，再比如Residual Network，BN本质上也是解释并从某个不同的角度来解决这个问题的。

 

|“Internal Covariate Shift”问题

 

从论文名字可以看出，BN是用来解决“InternalCovariate Shift”问题的，那么首先得理解什么是“Internal Covariate Shift”？

 

论文首先说明Mini-Batch SGD相对于One Example SGD的两个优势：梯度更新方向更准确；并行计算速度快；（本文作者：为啥要说这些？因为BatchNorm是基于Mini-Batch SGD的，所以先夸下Mini-Batch SGD，当然也是大实话）；

 

然后吐槽下SGD训练的缺点：超参数调起来很麻烦。（本文作者：作者隐含意思是用我大BN就能解决很多SGD的缺点：用了大BN，妈妈再也不用担心我的调参能力啦）

 

接着引入covariate shift的概念：如果ML系统实例集合<X,Y>中的输入值X的分布老是变，这不符合IID假设啊，那您怎么让我稳定的学规律啊，这不得引入迁移学习才能搞定吗，我们的ML系统还得去学习怎么迎合这种分布变化啊。

 

对于深度学习这种包含很多隐层的网络结构，在训练过程中，因为各层参数老在变，所以每个隐层都会面临covariate shift的问题，也就是在训练过程中，隐层的输入分布老是变来变去，这就是所谓的“Internal Covariate Shift”，Internal指的是深层网络的隐层，是发生在网络内部的事情，而不是covariate shift问题只发生在输入层。

 

然后提出了BatchNorm的基本思想：能不能让每个隐层节点的激活输入分布固定下来呢？这样就避免了“Internal Covariate Shift”问题了。

 

BN不是凭空拍脑袋拍出来的好点子，它是有启发来源的：之前的研究表明如果在图像处理中对输入图像进行白化（Whiten）操作的话——所谓白化，就是对输入数据分布变换到0均值，单位方差的正态分布——那么神经网络会较快收敛，那么BN作者就开始推论了：图像是深度神经网络的输入层，做白化能加快收敛，那么其实对于深度网络来说，其中某个隐层的神经元是下一层的输入，意思是其实深度神经网络的每一个隐层都是输入层，不过是相对下一层来说而已，那么能不能对每个隐层都做白化呢？这就是启发BN产生的原初想法，而BN也确实就是这么做的，可以理解为对深层神经网络每个隐层神经元的激活值做简化版本的白化操作。

 

|BatchNorm的本质思想

 

BN的基本思想其实相当直观：因为深层神经网络在做非线性变换前的激活输入值（就是那个x=WU+B，U是输入）随着网络深度加深或者在训练过程中，其分布逐渐发生偏移或者变动，之所以训练收敛慢，一般是整体分布逐渐往非线性函数的取值区间的上下限两端靠近（对于Sigmoid函数来说，意味着激活输入值WU+B是大的负值或正值），所以这导致后向传播时低层神经网络的梯度消失，这是训练深层神经网络收敛越来越慢的本质原因，而BN就是通过一定的规范化手段，把每层神经网络任意神经元这个输入值的分布强行拉回到均值为0方差为1的标准正太分布而不是萝莉分布（哦，是正态分布），其实就是把越来越偏的分布强制拉回比较标准的分布，这样使得激活输入值落在非线性函数对输入比较敏感的区域，这样输入的小变化就会导致损失函数较大的变化，意思是这样让梯度变大，避免梯度消失问题产生，而且梯度变大意味着学习收敛速度快，能大大加快训练速度。

 

THAT’S IT。其实一句话就是：对于每个隐层神经元，把逐渐向非线性函数映射后向取值区间极限饱和区靠拢的输入分布强制拉回到均值为0方差为1的比较标准的正态分布，使得非线性变换函数的输入值落入对输入比较敏感的区域，以此避免梯度消失问题。因为梯度一直都能保持比较大的状态，所以很明显对神经网络的参数调整效率比较高，就是变动大，就是说向损失函数最优值迈动的步子大，也就是说收敛地快。NB说到底就是这么个机制，方法很简单，道理很深刻。

 

上面说得还是显得抽象，下面更形象地表达下这种调整到底代表什么含义。



                    

  图1. 几个正态分布

 

假设某个隐层神经元原先的激活输入x取值符合正态分布，正态分布均值是-2，方差是0.5，对应上图中最左端的浅蓝色曲线，通过BN后转换为均值为0，方差是1的正态分布（对应上图中的深蓝色图形），意味着什么，意味着输入x的取值正态分布整体右移2（均值的变化），图形曲线更平缓了（方差增大的变化）。这个图的意思是，BN其实就是把每个隐层神经元的激活输入分布从偏离均值为0方差为1的正态分布通过平移均值压缩或者扩大曲线尖锐程度，调整为均值为0方差为1的正态分布。

 

那么把激活输入x调整到这个正态分布有什么用？

 

首先我们看下均值为0，方差为1的标准正态分布代表什么含义：



 

 图2.均值为0方差为1的标准正态分布图

 

这意味着在一个标准差范围内，也就是说64%的概率x其值落在[-1,1]的范围内，在两个标准差范围内，也就是说95%的概率x其值落在了[-2,2]的范围内。那么这又意味着什么？我们知道，激活值x=WU+B,U是真正的输入，x是某个神经元的激活值，假设非线性函数是sigmoid，那么看下sigmoid(x)其图形：



 

图3. Sigmoid(x)

 

及sigmoid(x)的导数为：G’=f(x)*(1-f(x))，因为f(x)=sigmoid(x)在0到1之间，所以G’在0到0.25之间，其对应的图如下：




 

图4. Sigmoid(x)导数图

 

假设没有经过BN调整前x的原先正态分布均值是-6，方差是1，那么意味着95%的值落在了[-8,-4]之间，那么对应的Sigmoid（x）函数的值明显接近于0，这是典型的梯度饱和区，在这个区域里梯度变化很慢，为什么是梯度饱和区？请看下sigmoid(x)如果取值接近0或者接近于1的时候对应导数函数取值，接近于0，意味着梯度变化很小甚至消失。而假设经过BN后，均值是0，方差是1，那么意味着95%的x值落在了[-2,2]区间内，很明显这一段是sigmoid(x)函数接近于线性变换的区域，意味着x的小变化会导致非线性函数值较大的变化，也即是梯度变化较大，对应导数函数图中明显大于0的区域，就是梯度非饱和区。

 

从上面几个图应该看出来BN在干什么了吧？其实就是把隐层神经元激活输入x=WU+B从变化不拘一格的正态分布通过BN操作拉回到了均值为0，方差为1的正态分布，即原始正态分布中心左移或者右移到以0为均值，拉伸或者缩减形态形成以1为方差的图形。什么意思？就是说经过BN后，目前大部分Activation的值落入非线性函数的线性区内，其对应的导数远离导数饱和区，这样来加速训练收敛过程。

 

但是很明显，看到这里，稍微了解神经网络的读者一般会提出一个疑问：如果都通过BN，那么不就跟把非线性函数替换成线性函数效果相同了？这意味着什么？我们知道，如果是多层的线性函数变换其实这个深层是没有意义的，因为多层线性网络跟一层线性网络是等价的。这意味着网络的表达能力下降了，这也意味着深度的意义就没有了。所以BN为了保证非线性的获得，对变换后的满足均值为0方差为1的x又进行了scale加上shift操作(y=scale*x+shift)，每个神经元增加了两个参数scale和shift参数，这两个参数是通过训练学习到的，意思是通过scale和shift把这个值从标准正态分布左移或者由移一点并长胖一点或者变瘦一点，每个实例挪动的程度不一样，这样等价于非线性函数的值从正中心周围的线性区往非线性区动了动。核心思想应该是想找到一个线性和非线性的较好平衡点，既能享受非线性的较强表达能力的好处，又避免太靠非线性区两头使得网络收敛速度太慢。当然，这是我的理解，论文作者并未明确这样说。但是很明显这里的scale和shift操作是会有争议的，因为按照论文作者论文里写的理想状态，就会又通过scale和shift操作把变换后的x调整回未变换的状态，那不是饶了一圈又绕回去原始的“Internal Covariate Shift”问题里去了吗，感觉论文作者并未能够清楚地解释scale和shift操作的理论原因。

 

|训练阶段如何做BatchNorm

 

上面是对BN的抽象分析和解释，具体在Mini-Batch SGD下做BN怎么做？其实论文里面这块写得很清楚也容易理解。为了保证这篇文章完整性，这里简单说明下。

 

假设对于一个深层神经网络来说，其中两层结构如下：



 

  图5. DNN其中两层

 

要对每个隐层神经元的激活值做BN，可以想象成每个隐层又加上了一层BN操作层，它位于X=WU+B激活值获得之后，非线性函数变换之前，其图示如下：



 

  图6. BN操作

 

对于Mini-Batch SGD来说，一次训练过程里面包含m个训练实例，其具体BN操作就是对于隐层内每个神经元的激活值来说，进行如下变换：



 

要注意，这里t层某个神经元的x(k)不是指原始输入，就是说不是t-1层每个神经元的输出，而是t层这个神经元的激活x=WU+B，这里的U才是t-1层神经元的输出。

 

变换的意思是：某个神经元对应的原始的激活x通过减去mini-Batch内m个实例获得的m个激活x求得的均值E(x)并除以求得的方差Var(x)来进行转换。

 

上文说过经过这个变换后某个神经元的激活x形成了均值为0，方差为1的正态分布，目的是把值往后续要进行的非线性变换的线性区拉动，增大导数值，增强反向传播信息流动性，加快训练收敛速度。但是这样会导致网络表达能力下降，为了防止这一点，每个神经元增加两个调节参数（scale和shift），这两个参数是通过训练来学习到的，用来对变换后的激活反变换，使得网络表达能力增强，即对变换后的激活进行如下的scale和shift操作，这其实是变换的反操作：



BN其具体操作流程，如论文中描述的一样：



 

 

过程非常清楚，就是上述公式的流程化描述，这里不解释了，直接应该能看懂。

 

|BatchNorm的推理过程

 

BN在训练的时候可以根据Mini-Batch里的若干训练实例进行激活数值调整，但是在推理（inference）的过程中，很明显输入就只有一个实例，看不到Mini-Batch其它实例，那么这时候怎么对输入做BN呢？因为很明显一个实例是没法算实例集合求出的均值和方差的。这可如何是好？这可如何是好？这可如何是好？

 

既然没有从Mini-Batch数据里可以得到的统计量，那就想其它办法来获得这个统计量，就是均值和方差。可以用从所有训练实例中获得的统计量来代替Mini-Batch里面m个训练实例获得的均值和方差统计量，因为本来就打算用全局的统计量，只是因为计算量等太大所以才会用Mini-Batch这种简化方式的，那么在推理的时候直接用全局统计量即可。

 

决定了获得统计量的数据范围，那么接下来的问题是如何获得均值和方差的问题。很简单，因为每次做Mini-Batch训练时，都会有那个Mini-Batch里m个训练实例获得的均值和方差，现在要全局统计量，只要把每个Mini-Batch的均值和方差统计量记住，然后对这些均值和方差求其对应的数学期望即可得出全局统计量，即：



有了均值和方差，每个隐层神经元也已经有对应训练好的Scaling参数和Shift参数，就可以在推导的时候对每个神经元的激活数据计算NB进行变换了，在推理过程中进行NB采取如下方式：



这个公式其实和训练时



 

是等价的，通过简单的合并计算推导就可以得出这个结论。那么为啥要写成这个变换形式呢？我猜作者这么写的意思是：在实际运行的时候，按照这种变体形式可以减少计算量，为啥呢？因为对于每个隐层节点来说：



 

 

都是固定值，这样这两个值可以事先算好存起来，在推理的时候直接用就行了，这样比原始的公式每一步骤都现算少了除法的运算过程，乍一看也没少多少计算量，但是如果隐层节点个数多的话节省的计算量就比较多了。

 

|BatchNorm的好处

 

BatchNorm为什么NB呢，关键还是效果好。不仅仅极大提升了训练速度，收敛过程大大加快，还能增加分类效果，一种解释是这是类似于Dropout的一种防止过拟合的正则化表达方式，所以不用Dropout也能达到相当的效果。另外调参过程也简单多了，对于初始化要求没那么高，而且可以使用大的学习率等。总而言之，经过这么简单的变换，带来的好处多得很，这也是为何现在BN这么快流行起来的原因。

 

扫一扫关注微信号：“布洛卡区” ，深度学习在自然语言处理等智能应用的技术研讨与科普公众号。

 

 

 

 

 

 

 

f.train.batch与tf.train.shuffle_batch的作用都是从队列中读取数据．

tf.train.batch

tf.train.batch() 按顺序读取队列中的数据

队列中的数据始终是一个有序的队列．队头一直按顺序补充，队尾一直按顺序出队．

参数：

tensors：排列的张量或词典．
	
batch_size：从队列中提取新的批量大小．
	
num_threads：线程数量．若批次是不确定　num_threads > 1．
	
capacity：队列中元素的最大数量．
	
enqueue_many：tensors中的张量是否都是一个例子．
	
shapes:每个示例的形状．(可选项）
	
dynamic_pad：在输入形状中允许可变尺寸．
	
allow_smaller_final_batch：为True时，若队列中没有足够的项目，则允许最终批次更小．(可选项）
	
shared_name：如果设置，则队列将在多个会话中以给定名称共享．(可选项）
	
name：操作的名称．(可选项）
若enqueue_many为False，则认为tensors代表一个示例．输入张量形状为[x, y, z]时，则输出张量形状为[batch_size, x, y, z]．

若enqueue_many为True，则认为tensors代表一批示例，其中第一个维度为示例的索引，并且所有成员tensors在第一维中应具有相同大小．若输入张量形状为[*, x, y, z]，则输出张量的形状为[batch_size, x, y, z]．

tf.train.batch()示例

#!/usr/bin/python
# coding:utf-8
import tensorflow as tf
import numpy as np

images = np.random.random([5,2])
label = np.asarray(range(0, 5))
images = tf.cast(images, tf.float32)
label = tf.cast(label, tf.int32)
# 切片
input_queue = tf.train.slice_input_producer([images, label], shuffle=False)
# 按顺序读取队列中的数据
image_batch, label_batch = tf.train.batch(input_queue, batch_size=10, num_threads=1, capacity=64)

with tf.Session() as sess:
    # 线程的协调器
    coord = tf.train.Coordinator()
    # 开始在图表中收集队列运行器
    threads = tf.train.start_queue_runners(sess, coord)
    image_batch_v, label_batch_v = sess.run([image_batch, label_batch])
    for j in range(5):
        print(image_batch_v[j]),
        print(label_batch_v[j])
    # 请求线程结束
    coord.request_stop()
    # 等待线程终止
    coord.join(threads)

按顺序读取队列中的数据，输出：

[ 0.05013787  0.53446019] 0
[ 0.91189879  0.69153142] 1
[ 0.39966023  0.86109054] 2
[ 0.85078746  0.05766034] 3
[ 0.71261722  0.60514599] 4

tf.train.shuffle_batch

tf.train.shuffle_batch() 将队列中数据打乱后再读取出来．

函数是先将队列中数据打乱，然后再从队列里读取出来，因此队列中剩下的数据也是乱序的．

tensors：排列的张量或词典．
	
batch_size：从队列中提取新的批量大小．
	
capacity：队列中元素的最大数量．
	
min_after_dequeue：出队后队列中元素的最小数量，用于确保元素的混合级别．
	
num_threads：线程数量．
	
seed：队列内随机乱序的种子值．
	
enqueue_many：tensors中的张量是否都是一个例子．
	
shapes：每个示例的形状．(可选项）
	
allow_smaller_final_batch：为True时，若队列中没有足够的项目，则允许最终批次更小．(可选项）
	
shared_name：如果设置，则队列将在多个会话中以给定名称共享．(可选项）
	
name：操作的名称．(可选项）
其他与tf.train.batch()类似．

tf.train.shuffle_batch示例

#!/usr/bin/python
# coding:utf-8
import tensorflow as tf
import numpy as np

images = np.random.random([5,2])
label = np.asarray(range(0, 5))
images = tf.cast(images, tf.float32)
label = tf.cast(label, tf.int32)
input_queue = tf.train.slice_input_producer([images, label], shuffle=False)
# 将队列中数据打乱后再读取出来
image_batch, label_batch = tf.train.shuffle_batch(input_queue, batch_size=10, num_threads=1, capacity=64, min_after_dequeue=1)

with tf.Session() as sess:
    # 线程的协调器
    coord = tf.train.Coordinator()
    # 开始在图表中收集队列运行器
    threads = tf.train.start_queue_runners(sess, coord)
    image_batch_v, label_batch_v = sess.run([image_batch, label_batch])
    for j in range(5):
        # print(image_batch_v.shape, label_batch_v[j])
        print(image_batch_v[j]),
        print(label_batch_v[j])
    # 请求线程结束
    coord.request_stop()
    # 等待线程终止
    coord.join(threads)

将队列中数据打乱后再读取出来，输出：

[ 0.08383977  0.75228119] 1
[ 0.03610427  0.53876138] 0
[ 0.33962703  0.47629601] 3
[ 0.21824744  0.84182823] 4
[ 0.8376292   0.52254623] 2

深度学习中经常看到epoch、 iteration和batchsize，下面按自己的理解说说这三个的区别：
（1）batchsize：批大小。在深度学习中，一般采用SGD训练，即每次训练在训练集中取batchsize个样本训练；

（2）iteration：1个iteration等于使用batchsize个样本训练一次；

（3）epoch：1个epoch等于使用训练集中的全部样本训练一次；
举个例子，训练集有1000个样本，batchsize=10，那么：

训练完整个样本集需要：

100次iteration，1次epoch。
1.当数据量足够大的时候可以适当的减小batch_size,由于数据量太大，内存不够。但盲目减少会导致无法收敛，batch_size=1时为在线学习，也是标准的SGD，这样学习，如果数据量不大，noise数据存在时，模型容易被noise带偏，如果数据量足够大，noise的影响会被“冲淡”，对模型几乎不影响。2.batch的选择，首先决定的是下降方向，如果数据集比较小，则完全可以采用全数据集的形式。这样做的好处有两点，
 1）全数据集的方向能够更好的代表样本总体，确定其极值所在。

2）由于不同权重的梯度值差别巨大，因此选取一个全局的学习率很困难。
3.增大batchsize的好处有三点：
     1）内存的利用率提高了，大矩阵乘法的并行化效率提高。

     2）跑完一次epoch(全数据集)所需迭代次数减少，对于相同的数据量的处理速度进一步加快。

     3）一定范围内，batchsize越大，其确定的下降方向就越准，引起训练震荡越小。

4.盲目增大的坏处有三点：
1）当数据集太大时，内存撑不住。
     2）跑完一次epocffe-master/tools/extra/parse_log.sh  caffe-master/tools/extra/extract_seconds.py和h(全数据集)所需迭代次数减少了，但要想达到相同的      精度，时间开销太大，参数的修正更加缓慢。

     3）batchsize增大到一定的程度，其确定的下降方向已经基本不再变化。

总结：
     1）batch数太小，而类别又比较多的时候，真的可能会导致loss函数震荡而不收敛，尤其是在你的网络比较复杂的时候。

     2）随着batchsize增大，处理相同的数据量的速度越快。

     3）随着batchsize增大，达到相同精度所需要的epoch数量越来越多。

     4）由于上述两种因素的矛盾， Batch_Size 增大到某个时候，达到时间上的最优。

     5）由于最终收敛精度会陷入不同的局部极值，因此 Batch_Size 增大到某些时候，达到最终收敛精度上的最优。

     6）过大的batchsize的结果是网络很容易收敛到一些不好的局部最优点。同样太小的batch也存在一些问题，比如训练速度很慢，训练不容易收敛等。

     7）具体的batch size的选取和训练集的样本数目相关。

还有一个代码上的细节：就是代码实现上选取一个batch的时候似乎是按着数据库的图片顺序选取输入图片的，所以在生成数据库的时候切记要shuffle一下图片顺序。caffe中ImageDataLayer有shuffle参数，生成lmdb时也有shuffle参数不必手动。
显存占用不是和batch size简单成正比
增大batch size能减缓梯度震荡，需要更少的迭代优化次数，收敛的更快，但是每次迭代耗时更长。

https://zhuanlan.zhihu.com/p/31558973
要想收敛到同一个最优点，使用整个样本集时，虽然迭代次数少，但是每次迭代的时间长，耗费的总时间是大于使用少量样本多次迭代的情况的。

实际上，工程上在使用GPU训练时，跑一个样本花的时间与跑几十个样本甚至几百个样本的时间是一样的！当然得益于GPU里面超多的核，超强的并行计算能力啦。因此，在工程实际中，从收敛速度的角度来说，小批量的样本集是最优的，也就是我们所说的mini-batch。这时的batch size往往从几十到几百不等，但一般不会超过几千。
样本量少的时候会带来很大的方差，而这个大方差恰好会导致我们在梯度下降到很差的局部最优点（只是微微凸下去的最优点）和鞍点的时候不稳定，一不小心就因为一个大噪声的到来导致炸出了局部最优点。

与之相反的，当样本量很多时，方差很小，对梯度的估计要准确和稳定的多，因此反而在差劲的局部最优点和鞍点时反而容易自信的呆着不走了，从而导致神经网络收敛到很差的点上，跟出了bug一样的差劲。
batch的size设置的不能太大也不能太小，因此实际工程中最常用的就是mini-batch，一般size设置为几十或者几百。

对于二阶优化算法，减小batch换来的收敛速度提升远不如引入大量噪声导致的性能下降，因此在使用二阶优化算法时，往往要采用大batch哦。此时往往batch设置成几千甚至一两万才能发挥出最佳性能。

GPU对2的幂次的batch可以发挥更佳的性能，因此设置成16、32、64、128…时往往要比设置为整10、整100的倍数时表现更优