JPA SQL Server 批量插入

Question

问题陈述

我正在尝试提高 JPA 中插入过程的性能。目前将大约 350,000 条记录插入我的数据库需要 4 分钟。为了提高性能，我想使用批量插入。我已经制作了一个大纲，显示了我开始使用的代码。我为尝试提高性能以及修复内存问题所做的修改。这些修改的结果。我的一些其他尝试未在修改中显示。请让我知道如何改进我的代码以允许使用 sql server 和 hibernate 进行大量插入。如果需要，我可以提供有关整个插入过程的更多信息。

起始代码

要启用此功能，请在 application.yml 中输入：

jpa:
  properties:
    hibernate:
      jdbc:
        batch_size: 1000
        batch_versioned_data: true
      order_inserts: true

我的 Loader 类中的代码如下所示：

try(Stream<String> lines = Files.lines("/path/to/file")) {
    Iterators.partition(lines.iterator(), 1000).forEachRemaining(batchList -> {
        List<CustomEntity> mappedEntities = list.stream().map(mapLineToEntity).collect(Collectors.toList());

        //Insert batch
        repository.saveAll(mappendEntities);
        repository.flush();
    })
}

代码修改

但这导致了内存问题，提示使用 EntityManager 的自定义 sql 实现来刷新和清除持久化的实体。为此，我创建了一个 CustomEntityServiceCustom.java 接口及其实现。以下是我对修改后的 Loader 类的两次尝试：

public interface CustomEntityServiceCustom {
  void batchInsertProcess(List<CustomEntity> customEntities, int start); //try 1 
  void batchInsertProcess(List<CustomEntity> customEntities, AtomicInteger start); //try 2
}

public class Custom CustomEntityServiceCustomImpl implements CustomEntityServiceCustom {
  @PersistenceContext
  private EntityManager em;

  //Try 1
  @Override
  @Transactional
  void batchInsertProcess(List<CustomEntity> customEntities, int start) {
    for(CustomEntity ent : customEntities) {
      em.persist(ent)
    }
    em.flush();
    em.clear();
  }

  //Try 2
  @Override
  @Transactional
  void batchInsertProcess(List<CustomEntity> customEntities, AtomicInteger start) {
    final int numRecsPerInsert = 25;
    Iterators.partition(customEntities.iterator(), numRecsPerInsert).forEachRemaining(batchList -> {
      /*
      code not included but createInsert will create an insert statement like the following:
      INSERT INTO table (col1, col2) VALUES (rec1val1, rec1val2), (rec2val, rec2val2)
      for 25 records at a time and then update the AtomicInteger 
      */
      String multiLineInsert = createInsert(batchList, start.get());
      start.addAndGet(numRecsPerInsert);
      em.createNativeQuery(multiLineInsert).executeUpdate();
    })
    em.flush();
    em.clear();
  }
} 

//updated loader

try(Stream<String> lines = Files.lines("/path/to/file")) {
    AtomicInteger start = new AtomicInteger(1);
    Iterators.partition(lines.iterator(), 1000).forEachRemaining(batchList -> {
        List<CustomEntity> mappedEntities = list.stream().map(mapLineToEntity).collect(Collectors.toList());
    repository.batchInsertProcess(mappedEntities, start);
    })
}

尝试 1 和 2 都通过不使用来利用不使用自动生成的 ID：

@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)

当前指标

进行所有这些更改后，初始代码的性能并没有显着提升。
Try 1 在 35 秒内实现了约 50,000 条记录的插入，而 Try 2 在 1 分钟内执行相同的插入。我不明白这一点，因为 Try 1 一次插入 1 条记录，例如： INSERT INTO 表 (COL1, COL2) 值 (VAL1, VAL2);而 Try 2 在 1 条语句中插入 25 条记录。我还按照推荐的here 尝试了每次插入 4 条记录，但这仍然是 52 秒，这比不使用批量插入的 35 秒要大得多。

其他注意事项

我试图允许休眠处理This 之后的每个语句的多个记录，但我没有看到 SQL Server 重写BatchedStatements 的选项。我试过添加 useBulkCopyForBatchInsert=true;到详细的连接字符串here，但我不确定我是否必须修改我的代码才能看到此更改的好处？

我也不确定在执行更新（）之后是否需要刷新 EntityManager，因为在日志中我收到一条消息，指出执行 0 次刷新花费了 0 纳秒，执行 2074 次部分刷新花费了 6596474 纳秒。这可能是另一个瓶颈，但我不确定幕后究竟发生了什么。

Answer 1

如果您使用这样的多值子句运行插入语句，您可能会一直进行硬解析。您应该改用参数，以便将语句缓存在服务器端。此外，您应该重用 Query 对象并重新绑定值。这就是 Hibernate 在使用批量插入时在幕后为您所做的。此外，它将使用 JDBC Batch API，该 API 可以在协议级别上做一些技巧，以进一步提高性能。

总结一下，我认为多值子句不会比通过 JDBC Batch API 进行的批量插入更好。如果这真的表现得更好，我会说这是 JDBC 驱动程序中的一个错误，应该修复。即使修复只是在幕后使用多值子句语句。

无论如何，如果您想尝试一下，您应该按以下方式构建它：

  @Override
  @Transactional
  void batchInsertProcess(List<CustomEntity> customEntities, AtomicInteger start) {
    final int numRecsPerInsert = 25;
    // creates "insert into ... values (?, ?, ?), (?, ?, ?), ..."
    String multiLineInsert = createInsert(numRecsPerInsert);
    Query query = em.createNativeQuery(multiLineInsert);
    Iterators.partition(customEntities.iterator(), numRecsPerInsert).forEachRemaining(batchList -> {
      bindValues(query, batchList, start.get(), numRecsPerInsert);
      start.addAndGet(numRecsPerInsert);
      query.executeUpdate();
    });
  }