天天看點
傳回

Spring Batch 批處理，騷氣還強大

作者：Java小老太

spring batch簡介

spring batch是spring提供的一個資料處理架構。企業域中的許多應用程式需要批量處理才能在關鍵任務環境中執行業務操作。這些業務營運包括：

  1. 無需使用者互動即可最有效地處理大量資訊的自動化，複雜處理。這些操作通常包括基于時間的事件（例如月末計算，通知或通信）。
  2. 在非常大的資料集中重複處理複雜業務規則的定期應用（例如，保險利益确定或費率調整）。
  3. 內建從内部和外部系統接收的資訊，這些資訊通常需要以事務方式格式化，驗證和處理到記錄系統中。批處理用于每天為企業處理數十億的交易。

Spring Batch是一個輕量級，全面的批處理架構，旨在開發對企業系統日常營運至關重要的強大批處理應用程式。Spring Batch建構了人們期望的Spring Framework特性（生産力，基于POJO的開發方法和一般易用性），同時使開發人員可以在必要時輕松通路和利用更進階的企業服務。Spring Batch不是一個schuedling的架構。

Spring Batch提供了可重用的功能，這些功能對于處理大量的資料至關重要，包括記錄/跟蹤，事務管理，作業處理統計，作業重新開機，跳過和資源管理。它還提供更進階的技術服務和功能，通過優化和分區技術實作極高容量和高性能的批處理作業。Spring Batch可用于兩種簡單的用例（例如将檔案讀入資料庫或運作存儲過程）以及複雜的大量用例（例如在資料庫之間移動大量資料，轉換它等等） 上）。大批量批處理作業可以高度可擴充的方式利用該架構來處理大量資訊。

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的方法可以定義為：

Spring Batch 批處理，騷氣還強大

是以，我麼可以通過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,下面是一個從資料庫當中截圖的執行個體：

Spring Batch 批處理，騷氣還強大

什麼是Step

每一個Step對象都封裝了批處理作業的一個獨立的階段。事實上，每一個Job本質上都是由一個或多個步驟組成。每一個step包含定義和控制實際批處理所需的所有資訊。任何特定的内容都由編寫Job的開發人員自行決定。一個step可以非常簡單也可以非常複雜。例如，一個step的功能是将檔案中的資料加載到資料庫中，那麼基于現在spring batch的支援則幾乎不需要寫代碼。更複雜的step可能具有複雜的業務邏輯，這些邏輯作為處理的一部分。與Job一樣，Step具有與JobExecution類似的StepExecution，如下圖所示：

Spring Batch 批處理，騷氣還強大

什麼是StepExecution

StepExecution表示一次執行Step, 每次運作一個Step時都會建立一個新的StepExecution，類似于JobExecution。但是，某個步驟可能由于其之前的步驟失敗而無法執行。且僅當Step實際啟動時才會建立StepExecution。

一次step執行的執行個體由StepExecution類的對象表示。每個StepExecution都包含對其相應步驟的引用以及JobExecution和事務相關的資料，例如送出和復原計數以及開始和結束時間。此外，每個步驟執行都包含一個ExecutionContext，其中包含開發人員需要在批處理運作中保留的任何資料，例如重新啟動所需的統計資訊或狀态資訊。下面是一個從資料庫當中截圖的執行個體：

Spring Batch 批處理，騷氣還強大

什麼是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的示意圖如下：

Spring Batch 批處理，騷氣還強大

它的意思就和圖示的一樣，由于我們一次batch的任務可能會有很多的資料讀寫操作，是以一條一條的處理并向資料庫送出的話效率不會很高，是以spring batch提供了chunk這個概念，我們可以設定一個chunk size，spring batch 将一條一條處理資料，但不送出到資料庫，隻有當處理的資料數量達到chunk size設定的值得時候，才一起去commit.

java的執行個體定義代碼如下：

Spring Batch 批處理，騷氣還強大

在上面這個step裡面，chunk size被設為了10，當ItemReader讀的資料數量達到10的時候，這一批次的資料就一起被傳到itemWriter，同時transaction被送出。

skip政策和失敗處理

一個batch的job的step，可能會處理非常大數量的資料，難免會遇到出錯的情況，出錯的情況雖出現的機率較小，但是我們不得不考慮這些情況，因為我們做資料遷移最重要的是要保證資料的最終一緻性。spring batch當然也考慮到了這種情況，并且為我們提供了相關的技術支援，請看如下bean的配置：

Spring Batch 批處理，騷氣還強大

我們需要留意這三個方法，分别是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. 特别是，需要尋找以下四個常見缺陷：
  1. 當資料可以被讀取一次并緩存或儲存在工作存儲中時，讀取每個事務的資料。
  2. 重新讀取先前在同一事務中讀取資料的事務的資料。
  3. 導緻不必要的表或索引掃描。
  4. 未在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也不再列印，等待一段時間之後，得到如下錯誤：

Spring Batch 批處理，騷氣還強大

紅字的資訊為：Resource exhaustion event：the JVM was unable to allocate memory from the heap.

翻譯過來的意思就是項目發出了一個資源耗盡的事件，告訴我們java虛拟機無法再為堆配置設定記憶體。

造成這個錯誤的原因是: 這個項目裡的batch job的reader是一次性拿回了資料庫裡的所有資料，并沒有進行分頁，當這個資料量太大時，就會導緻記憶體不夠用。解決的辦法有兩個:

  • 調整reader讀資料邏輯，按分頁讀取，但實作上會麻煩一些，且運作效率會下降
  • 增大service記憶體
spring batch spring pring ring
上一篇: 8Sep2022Day1911Ihavenotfailed.I'vejustfound10,000waysthatwon
下一篇: 從操作實踐解讀《資料出境安全評估申報指南》

繼續閱讀