Azure Function上でCSVファイルのデータをDBに書き込む処理をカスタマイズしてみた

これまでこのブログで、Spring BatchのChunkモデルを用いて、Blob上のCSVファイルをDBのテーブルに書き込む処理を作成していたが、DBのテーブルに書き込む処理で、MyBatisBatchItemWriterクラスを使わず、カスタマイズすることもできる。

今回は、Blob上のCSVファイルをDBのテーブルに書き込む処理をカスタマイズしてみたので、そのサンプルプログラムを共有する。

前提条件

下記記事のサンプルプログラムを作成済であること。

Azure Function上でバッチモードとそれ以外のデータソースを混在させてみたAzure Function上で楽観ロックを実装する場合、DB接続する際のSqlSessionを生成する際に、Spring Batchの...

サンプルプログラムの作成

作成したサンプルプログラムの構成は以下の通り。

なお、上記の赤枠は、今回追加・変更したプログラムである。

カスタマイズしたDB書き込み処理は以下の通りで、writeメソッド内で各SQLを実行したり、更新件数を確認する処理を追加している。

package com.example.batch;

import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.ibatis.executor.BatchResult;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.InvalidDataAccessResourceUsageException;
import org.springframework.stereotype.Component;

import com.example.mybatis.batch.UserDataMapperBatch;
import com.example.mybatis.model.UserData;

@Component
public class DemoItemWriter implements ItemWriter<UserData> {
  
  /** USER_DATAテーブルにアクセスするためのMapper */
  @Autowired
  private UserDataMapperBatch userDataMapperBatch;
  
  /** SQLセッションファクトリ */
  @Autowired
  @Qualifier("sqlSessionFactoryBatch")
  private SqlSessionFactory sqlSessionFactoryBatch;
  
  /** SQLセッションテンプレート */
  private SqlSessionTemplate sqlSessionTemplate  = null;
  
  /**
   * 更新件数を取得するため、SQLセッションテンプレートを初期化
   */
  @PostConstruct
  public void init() {
    sqlSessionTemplate = new SqlSessionTemplate(
        sqlSessionFactoryBatch, ExecutorType.BATCH);
  }
  
  /**
   * 読み込んだファイルのデータを、DBに書き込む処理を定義する.
   */
  @Override
  public void write(List<? extends UserData> items) {
    for(UserData item : items) {
      // 指定したIDをもつUserDataオブジェクトを、DBからロックをかけて取得する
      Integer id = Integer.parseInt(item.getId());
      UserData userData = userDataMapperBatch.findByIdRowLock(id);
      
      // id=8の場合、バージョンを更新し、楽観ロックエラーにする
      /* if(id == 8) {
        userDataMapperBatch.updateVersion(userData);
      } */
      
      // DBにデータが無ければ追加、あれば更新する
      if(userData == null) {
        userDataMapperBatch.insert(item);
      } else {
        userDataMapperBatch.update(item);
      }
      
      // 更新件数を確認し、1件でなければエラーとする
      List<BatchResult> results = sqlSessionTemplate.flushStatements();
      if(results == null || results.size() != 1) {
        throw new InvalidDataAccessResourceUsageException(
            "楽観ロックエラーが発生しました。");
      }
    }
  }

}

package com.example.batch;

import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.ibatis.executor.BatchResult;

import org.apache.ibatis.session.ExecutorType;

import org.apache.ibatis.session.SqlSessionFactory;

import org.mybatis.spring.SqlSessionTemplate;

import org.springframework.batch.item.ItemWriter;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.dao.InvalidDataAccessResourceUsageException;

import org.springframework.stereotype.Component;

import com.example.mybatis.batch.UserDataMapperBatch;

import com.example.mybatis.model.UserData;

@Component

public class DemoItemWriter implements ItemWriter<UserData> {

/** USER_DATAテーブルにアクセスするためのMapper */

@Autowired

private UserDataMapperBatch userDataMapperBatch;

/** SQLセッションファクトリ */

@Autowired

@Qualifier("sqlSessionFactoryBatch")

private SqlSessionFactory sqlSessionFactoryBatch;

/** SQLセッションテンプレート */

private SqlSessionTemplate sqlSessionTemplate = null;

/**

* 更新件数を取得するため、SQLセッションテンプレートを初期化

@PostConstruct

public void init() {

sqlSessionTemplate = new SqlSessionTemplate(

sqlSessionFactoryBatch, ExecutorType.BATCH);

}

/**

* 読み込んだファイルのデータを、DBに書き込む処理を定義する.

@Override

public void write(List<? extends UserData> items) {

for(UserData item : items) {

// 指定したIDをもつUserDataオブジェクトを、DBからロックをかけて取得する

Integer id = Integer.parseInt(item.getId());

UserData userData = userDataMapperBatch.findByIdRowLock(id);

// id=8の場合、バージョンを更新し、楽観ロックエラーにする

/* if(id == 8) {

userDataMapperBatch.updateVersion(userData);

} */

// DBにデータが無ければ追加、あれば更新する

if(userData == null) {

userDataMapperBatch.insert(item);

} else {

userDataMapperBatch.update(item);

}

// 更新件数を確認し、1件でなければエラーとする

List<BatchResult> results = sqlSessionTemplate.flushStatements();

if(results == null || results.size() != 1) {

throw new InvalidDataAccessResourceUsageException(

"楽観ロックエラーが発生しました。");

}

また、バッチ処理のMapperインタフェース・XMLファイルの内容は以下の通りで、ロックをかけないデータ取得・追加・更新処理を追加している。

package com.example.mybatis.batch;

import org.apache.ibatis.annotations.Mapper;
import com.example.mybatis.model.UserData;

@Mapper
public interface UserDataMapperBatch {

  /**
   * 引数のIDをキーにもつUserDataオブジェクトを行ロックをかけ取得する.
   * @param id ID
   * @return UserDataオブジェクト
   */
  UserData findByIdRowLock(Integer id);
  
  /**
   * 引数のUserDataオブジェクトのバージョンを更新する.
   * @param userData UserDataオブジェクト
   */
  void updateVersion(UserData userData);

  /**
   * 引数のIDをキーにもつUserDataオブジェクトを取得する.
   * @param id ID
   * @return UserDataオブジェクト
   */
  UserData findById(Integer id);
  
  /**
   * 引数のUserDataオブジェクトをDBに追加する.
   * @param userData UserDataオブジェクト
   */
  void insert(UserData userData);
  
  /**
   * DBにある引数のUserDataオブジェクトを更新する.
   * @param userData UserDataオブジェクト
   */
  void update(UserData userData);
}

package com.example.mybatis.batch;

import org.apache.ibatis.annotations.Mapper;

import com.example.mybatis.model.UserData;

@Mapper

public interface UserDataMapperBatch {

/**

* 引数のIDをキーにもつUserDataオブジェクトを行ロックをかけ取得する.

* @param id ID

* @return UserDataオブジェクト

UserData findByIdRowLock(Integer id);

/**

* 引数のUserDataオブジェクトのバージョンを更新する.

* @param userData UserDataオブジェクト

void updateVersion(UserData userData);

/**

* 引数のIDをキーにもつUserDataオブジェクトを取得する.

* @param id ID

* @return UserDataオブジェクト

UserData findById(Integer id);

/**

* 引数のUserDataオブジェクトをDBに追加する.

* @param userData UserDataオブジェクト

void insert(UserData userData);

/**

* DBにある引数のUserDataオブジェクトを更新する.

* @param userData UserDataオブジェクト

void update(UserData userData);

}

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
    PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
    "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.example.mybatis.batch.UserDataMapperBatch">
  <select id="findByIdRowLock" parameterType="java.lang.Integer" 
        resultType="com.example.mybatis.model.UserData">
    SELECT id, name, birth_year, birth_month, birth_day, sex, memo, version 
    FROM USER_DATA WITH(ROWLOCK, UPDLOCK, NOWAIT) <!-- Azure SQL Databaseで行ロックをかける -->
    WHERE id = #{id} 
  </select>
  <update id="updateVersion" parameterType="com.example.mybatis.model.UserData">
    UPDATE USER_DATA SET version = #{version} + 2 WHERE id = #{id}
  </update>
  <select id="findById" parameterType="java.lang.Integer" 
        resultType="com.example.mybatis.model.UserData">
    SELECT id, name, birth_year, birth_month, birth_day, sex, memo, version 
    FROM USER_DATA
    WHERE id = #{id} 
  </select>
  <insert id="insert" parameterType="com.example.mybatis.model.UserData">
    INSERT INTO USER_DATA ( id, name, birth_year, birth_month, birth_day
        , sex, memo, version ) 
    VALUES (#{id}, #{name}, #{birth_year}, #{birth_month}, #{birth_day}
        , #{sex}, #{memo}, #{version})
  </insert>
  <update id="update" parameterType="com.example.mybatis.model.UserData">
    UPDATE USER_DATA SET name = #{name}, birth_year = #{birth_year}
      , birth_month = #{birth_month}, birth_day = #{birth_day}, sex = #{sex}
      , memo = #{memo}, version = #{version} + 1
    WHERE id = #{id} AND version = #{version}
  </update>
</mapper>

<?xml version="1.0" encoding="UTF-8" ?>

<!DOCTYPE mapper

PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"

"http://mybatis.org/dtd/mybatis-3-mapper.dtd">

<select id="findByIdRowLock" parameterType="java.lang.Integer"

resultType="com.example.mybatis.model.UserData">

SELECT id, name, birth_year, birth_month, birth_day, sex, memo, version

FROM USER_DATA WITH(ROWLOCK, UPDLOCK, NOWAIT)

WHERE id = #{id}

</select>

UPDATE USER_DATA SET version = #{version} + 2 WHERE id = #{id}

</update>

<select id="findById" parameterType="java.lang.Integer"

resultType="com.example.mybatis.model.UserData">

SELECT id, name, birth_year, birth_month, birth_day, sex, memo, version

FROM USER_DATA

WHERE id = #{id}

</select>

INSERT INTO USER_DATA ( id, name, birth_year, birth_month, birth_day

, sex, memo, version )

VALUES (#{id}, #{name}, #{birth_year}, #{birth_month}, #{birth_day}

, #{sex}, #{memo}, #{version})

</insert>

UPDATE USER_DATA SET name = #{name}, birth_year = #{birth_year}

, birth_month = #{birth_month}, birth_day = #{birth_day}, sex = #{sex}

, memo = #{memo}, version = #{version} + 1

WHERE id = #{id} AND version = #{version}

</update>

</mapper>

さらに、Spring Batchの定義クラスは以下の通りで、DB書き込み処理でDemoItemWriterクラスを利用するよう修正している。

package com.example.batch;

import java.net.MalformedURLException;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.UrlResource;

import com.example.mybatis.model.UserData;
import com.example.service.DemoBlobService;

import lombok.RequiredArgsConstructor;

@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
public class DemoChunkConfig {

  /** ジョブ生成ファクトリ */
  public final JobBuilderFactory jobBuilderFactory;

  /** ステップ生成ファクトリ */
  public final StepBuilderFactory stepBuilderFactory;

  /** データ加工処理 */
  private final DemoUpdateProcessor demoUpdateProcessor;
  
  /** データ加工前後処理 */
  private final DemoProcessorListener demoProcessorListener;
  
  /** データ書き込み前後処理 */
  private final DemoUpdateListener demoUpdateListener;
  
  /** データ書き込み処理 */
  private final DemoItemWriter demoItemWriter;
  
  /** ステップの前後処理 */
  private final DemoStepListener demoStepListener;
  
  /** BLOBへアクセスするサービス */
  @Autowired
  private DemoBlobService demoBlobService;

  /**
   * BlobStorageからファイルを読み込む.
   * @return 読み込みオブジェクト
   */
  @Bean
  public FlatFileItemReader<UserData> reader() {
    FlatFileItemReader<UserData> reader = new FlatFileItemReader<UserData>();
    try {
      // BlobStorageからファイル(user_data.csv)を読み込む際のURLをreaderに設定
      String url = demoBlobService.getBlobUrl("user_data.csv");
      reader.setResource(new UrlResource(url));
    } catch (MalformedURLException ex) {
      throw new RuntimeException(ex);
    }
    // ファイルを読み込む際の文字コード
    reader.setEncoding("UTF-8");
    // 1行目を読み飛ばす
    reader.setLinesToSkip(1);
    // ファイルから読み込んだデータをUserDataオブジェクトに格納
    reader.setLineMapper(new DefaultLineMapper<UserData>() {
      {
        setLineTokenizer(new DelimitedLineTokenizer() {
          {
            setNames(new String[] { "id", "name", "birth_year"
                , "birth_month", "birth_day", "sex", "memo" });
          }
        });
        setFieldSetMapper(new BeanWrapperFieldSetMapper<UserData>() {
          {
            setTargetType(UserData.class);
          }
        });
      }
    });
    return reader;
  }

  /**
   * Spring Batchのジョブ内で指定する処理単位(ステップ)を定義する.
   * @param reader 読み込みオブジェクト
   * @return Spring Batchのジョブ内で指定する処理単位
   */
  @Bean
  public Step step(ItemReader<UserData> reader) {
    // 生成するステップ内で読み込み/加工/加工後/書き込み/書き込み後/ステップ実行前後の処理の
    // 一連の流れを指定する
    // その際のチャンクサイズ(=何件読み込む毎にコミットするか)を3に指定している
    return stepBuilderFactory.get("step")
        .<UserData, UserData>chunk(3)
        .reader(reader)
        .processor(demoUpdateProcessor)
        .listener(demoProcessorListener)
        .writer(demoItemWriter)
        .listener(demoUpdateListener)
        .listener(demoStepListener)
        .build();
  }

  /**
   * Spring Batchのジョブを定義する.
   * @param jobListener 実行前後の処理(リスナ)
   * @param step        実行するステップ
   * @return Spring Batchのジョブ
   */
  @Bean
  public Job updateUserData(DemoJobListener jobListener, Step step) {
    // 生成するジョブ内で、実行前後の処理(リスナ)と処理単位(ステップ)を指定する
    return jobBuilderFactory
        .get("updateUserData")
        .incrementer(new RunIdIncrementer())
        .listener(jobListener)
        .flow(step)
        .end()
        .build();
  }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

package com.example.batch;

import java.net.MalformedURLException;

import org.springframework.batch.core.Job;

import org.springframework.batch.core.Step;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;

import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;

import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;

import org.springframework.batch.core.launch.support.RunIdIncrementer;

import org.springframework.batch.item.ItemReader;

import org.springframework.batch.item.file.FlatFileItemReader;

import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;

import org.springframework.batch.item.file.mapping.DefaultLineMapper;

import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.core.io.UrlResource;

import com.example.mybatis.model.UserData;

import com.example.service.DemoBlobService;

import lombok.RequiredArgsConstructor;

@Configuration

@EnableBatchProcessing

@RequiredArgsConstructor

public class DemoChunkConfig {

/** ジョブ生成ファクトリ */

public final JobBuilderFactory jobBuilderFactory;

/** ステップ生成ファクトリ */

public final StepBuilderFactory stepBuilderFactory;

/** データ加工処理 */

private final DemoUpdateProcessor demoUpdateProcessor;

/** データ加工前後処理 */

private final DemoProcessorListener demoProcessorListener;

/** データ書き込み前後処理 */

private final DemoUpdateListener demoUpdateListener;

/** データ書き込み処理 */

private final DemoItemWriter demoItemWriter;

/** ステップの前後処理 */

private final DemoStepListener demoStepListener;

/** BLOBへアクセスするサービス */

@Autowired

private DemoBlobService demoBlobService;

/**

* BlobStorageからファイルを読み込む.

* @return 読み込みオブジェクト

@Bean

public FlatFileItemReader<UserData> reader() {

FlatFileItemReader<UserData> reader = new FlatFileItemReader<UserData>();

try {

// BlobStorageからファイル(user_data.csv)を読み込む際のURLをreaderに設定

String url = demoBlobService.getBlobUrl("user_data.csv");

reader.setResource(new UrlResource(url));

} catch (MalformedURLException ex) {

throw new RuntimeException(ex);

}

// ファイルを読み込む際の文字コード

reader.setEncoding("UTF-8");

// 1行目を読み飛ばす

reader.setLinesToSkip(1);

// ファイルから読み込んだデータをUserDataオブジェクトに格納

reader.setLineMapper(new DefaultLineMapper<UserData>() {

{

setLineTokenizer(new DelimitedLineTokenizer() {

{

setNames(new String[] { "id", "name", "birth_year"

, "birth_month", "birth_day", "sex", "memo" });

}

});

setFieldSetMapper(new BeanWrapperFieldSetMapper<UserData>() {

{

setTargetType(UserData.class);

}

});

}

});

return reader;

}

/**

* Spring Batchのジョブ内で指定する処理単位(ステップ)を定義する.

* @param reader 読み込みオブジェクト

* @return Spring Batchのジョブ内で指定する処理単位

@Bean

public Step step(ItemReader<UserData> reader) {

// 生成するステップ内で読み込み/加工/加工後/書き込み/書き込み後/ステップ実行前後の処理の

// 一連の流れを指定する

// その際のチャンクサイズ(=何件読み込む毎にコミットするか)を3に指定している

return stepBuilderFactory.get("step")

.<UserData, UserData>chunk(3)

.reader(reader)

.processor(demoUpdateProcessor)

.listener(demoProcessorListener)

.writer(demoItemWriter)

.listener(demoUpdateListener)

.listener(demoStepListener)

.build();

}

/**

* Spring Batchのジョブを定義する.

* @param jobListener 実行前後の処理(リスナ)

* @param step 実行するステップ

* @return Spring Batchのジョブ

@Bean

public Job updateUserData(DemoJobListener jobListener, Step step) {

// 生成するジョブ内で、実行前後の処理(リスナ)と処理単位(ステップ)を指定する

return jobBuilderFactory

.get("updateUserData")

.incrementer(new RunIdIncrementer())

.listener(jobListener)

.flow(step)

.end()

.build();

}

また、データ加工前後の処理を定義したクラスは以下の通りで、バージョンを取得する処理でロックをかけないようにしている。

package com.example.batch;

import org.springframework.batch.core.ItemProcessListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.example.mybatis.batch.UserDataMapperBatch;
import com.example.mybatis.model.UserData;

@Component
public class DemoProcessorListener implements ItemProcessListener<UserData, UserData> {

  @Autowired
  private UserDataMapperBatch userDataMapperBatch;
  
  /**
   * データ加工前の処理を定義する.
   */
  @Override
  public void beforeProcess(UserData item) {
    // 何もしない  
  }

  /**
   * データ加工後の処理を定義する.
   */
  @Override
  @Transactional
  public void afterProcess(UserData item, UserData result) {
    // 処理結果がnullの場合は、処理を終了する
    if(result == null) {
      return;
    }
    
    // 既に登録済データのVERSIONを取得する
    Integer id = Integer.parseInt(result.getId());
    UserData ud = userDataMapperBatch.findById(id);
    
    // バージョンの値を結果に設定する
    if(ud != null) {
      result.setVersion(ud.getVersion());
    } else {
      result.setVersion(0);
    }
  }

  /**
   * データ編集エラー後の処理を定義する.
   */
  @Override
  public void onProcessError(UserData item, Exception e) {
    // 何もしない
  }

}

package com.example.batch;

import org.springframework.batch.core.ItemProcessListener;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import org.springframework.transaction.annotation.Transactional;

import com.example.mybatis.batch.UserDataMapperBatch;

import com.example.mybatis.model.UserData;

@Component

public class DemoProcessorListener implements ItemProcessListener<UserData, UserData> {

@Autowired

private UserDataMapperBatch userDataMapperBatch;

/**

* データ加工前の処理を定義する.

@Override

public void beforeProcess(UserData item) {

// 何もしない

}

/**

* データ加工後の処理を定義する.

@Override

@Transactional

public void afterProcess(UserData item, UserData result) {

// 処理結果がnullの場合は、処理を終了する

if(result == null) {

return;

}

// 既に登録済データのVERSIONを取得する

Integer id = Integer.parseInt(result.getId());

UserData ud = userDataMapperBatch.findById(id);

// バージョンの値を結果に設定する

if(ud != null) {

result.setVersion(ud.getVersion());

} else {

result.setVersion(0);

}

/**

* データ編集エラー後の処理を定義する.

@Override

public void onProcessError(UserData item, Exception e) {

// 何もしない

}

その他のソースコード内容は、以下のサイトを参照のこと。
https://github.com/purin-it/azure/tree/master/azure-functions-customize-db-write/demoAzureFunc

サンプルプログラムの実行結果

サンプルプログラムの実行結果は、以下の記事の「サンプルプログラムの実行結果」と同じ内容となる。

Azure Function上でCSVファイルのデータをDBに取り込むプログラムで楽観ロックを使ってみたこれまでこのブログで、Spring BatchのChunkモデルを用いて、Blob上のCSVファイルをDBのテーブルに書き込む処理を作成...

要点まとめ

Spring BatchのChunkモデルを用いて、Blob上のCSVファイルをDBのテーブルに書き込む処理を、ItemWriterインタフェースを継承してカスタマイズすることができる。

前提条件

サンプルプログラムの作成

サンプルプログラムの実行結果

要点まとめ

TimerTriggerによって動作するAzure Functionsの関数を手動実行してみた

TimerTriggerによって動作するAzure Function上でCSVファイルの内容をDBに書き込んでみた

CSVファイルの内容をDBに書き込む前にファイルの文字コードを判定してみた

Spring BatchのCSVファイルを出力する処理でBlobContainerClientクラスを利用してみた

Spring BatchのChunkモデルを利用したバッチで中間ファイルを作らずCSVファイルを出力してみた

Azure Functionsのプログラムを2023年9月30日に動作できるよう修正してみた