从 java 中读取 100 GB .csv 文件并插入 MS SQL server DB

Question

我正在尝试使用批量插入从 csv 文件中读取数据并插入数据库。但它会引发多个异常。

java.sql.BatchUpdateException: A statement attempted to return a result set in executeBatch().<br>
SEVERE: Servlet.service() for servlet oms threw exception
java.lang.OutOfMemoryError: Java heap space ……. <br>

下面给出了插入代码：-

 public long updateOpenOrdData(Connection conn, String[] paramStrObj)throws Exception {
            long updatedRow = 0;
            CallableStatement cstmt = null;
             // System.out.println("StrAtt length :"+ strArr.length);
            try{
                for(int i=0; i<paramStrObj.length; i++){
                    int count =1;
                    int index = 0;
                    String[] dataArr = paramStrObj[i].split(",(?=([^\"]*\"[^\"]*\")*[^\"]*$)", -1); 
                    if(!dataArr[0].equals("SLC_Code_Desc") && dataArr.length >= 24 ){
                        cstmt = conn.prepareCall(PROC_INSERT_OPEN_ORD_TEMP);
                        System.out.print(dataArr[index]+", ");
                        cstmt.setString(count++, dataArr[index++]);
                        System.out.print(dataArr[index]+", ");
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index++]);
                        cstmt.setString(count++, dataArr[index]);
                        cstmt.setString(count++, null);
                        cstmt.setString(count++, null);
                        cstmt.setString(count++, null);
                        cstmt.addBatch();
                        //cstmt.executeUpdate();
                        }
                    }
                int[] insertRow =cstmt.executeBatch();
                System.out.println("Inserted row are :: "+insertRow);
                }catch (Exception e) {
                //System.out.println("Wrong data at line "+i+" and column "+ --index);
                e.printStackTrace();
            }finally{
                new DBService().releaseResources(null, cstmt);
            }

            return updatedRow;
        }

我需要插入最大 100 GB 的数据。 在这段代码中，每次插入一行。

Answer 1

您正面临OutOfMemoryError - 如果您尝试将所有您的 INSERT 放入 一个批次，这并不奇怪。

设置一个合理的批量大小（本例中为 10000）并执行以下操作：

  cstmt.addBatch();
  if(++batchCounter % 10000 == 0) {
  cstmt.executeBatch();
  }
}
cstmt.executeBatch(); //one final time for the remaining rows

同样，您不应该使用从 CSV 读取的完整行数组来调用您的方法 - 所以也许您已经在该级别上实现了批量大小？

同样，一个 batched 语句不应该返回结果集 - 所以不是调用一个返回结果的过程，你 强烈建议将此作为“经典”INSERT 操作。

Answer 2

对于此类工作，您可以使用 mysql 工具：mysqlimport 您将获得更好的性能，并且无需处理 java 限制。

编辑： 您可以使用诸如 talend 或 sql server 集成服务之类的 ETL 来执行此操作并使用批量插入节点。

Answer 3

public SettingsDto uploadInChunck(Connection conn, SettingsDto oSettingsDto)throws Exception{
    SettingsDto objSettingsDto = new SettingsDto();
    List<SettingsDto> lstObj = null;
    FileChannel fc =null;
    File file = null;
    file = new File(oSettingsDto.getFormFile().getOriginalFilename());
    oSettingsDto.getFormFile().transferTo(file);
    RandomAccessFile raf = new RandomAccessFile(file, "r");
    fc = raf.getChannel();
    long startTime = System.currentTimeMillis();
        try{
            ByteBuffer buffer = ByteBuffer.allocate(512);
            StringBuilder sb = new StringBuilder();
            while(fc.read(buffer) > 0)
            {
                buffer.flip();
                for (int i = 0; i < buffer.limit(); i++)
                {
                    sb.append((char) buffer.get());
                }

                String[] strArr = sb.toString().split("\\n");

                if(null != oSettingsDto.getUploadFileTemplate() && oSettingsDto.getUploadFileTemplate().equals("Open Order Data")){
                    settingsDao.updateOpenOrdData(conn, strArr);
                } else if(null != oSettingsDto.getUploadFileTemplate() && oSettingsDto.getUploadFileTemplate().equals("Shipment Forecast Data")){
                    settingsDao.updateShipmentForcastData(conn, strArr);
                }else if(null != oSettingsDto.getUploadFileTemplate() && oSettingsDto.getUploadFileTemplate().equals("Inventory Target Data")){
                    settingsDao.updateInventoryTargetData(conn, strArr);
                }else if(null != oSettingsDto.getUploadFileTemplate() && oSettingsDto.getUploadFileTemplate().equals("Current Inventory Data")){
                    settingsDao.updateCurrentInventoryData(conn, strArr);
                }

                buffer.clear(); // do something with the data and clear/compact it.
                long endTime = System.currentTimeMillis();
                System.out.println("startTime :"+startTime);
                System.out.println("endTime :"+endTime);

            }
        }catch(Exception e){
        //e.printStackTrace();
        fc.close();
        raf.close();
        file.deleteOnExit();
    }


    return objSettingsDto;
}