当前位置: 首页>编程语言>正文

mysql的in太多怎么解决 mysql 大量in解决


mysql大数量批量插入方案 LOAD DATA LOCAL INFILE实现及相关问题解决

  • 业务场景
  • 过程
  • 代码
  • 性能对比
  • 问题及解决办法


业务场景

项目跟文件内容相关,上传文件将句子内容提取出来后,将每条句子信息插入到数据库,文件中提取出的句子数量较大,几k-几十w不等,为了将数据快速插入数据库,寻找方案。

过程

项目中使用mybatis做持久存储,首先尝试 批量sql的方式:

insert into table() values()()()()()()()...

由于数据量大,遇到两个问题:
1.values后面拼接数据,本地机器内存不够
2.插入效率仍然很慢

思考:使用多线程分批插入
结果:
1.文件大及数量多,线程数量太多
2.效率仍提升不明显

 

借鉴前人肩膀,但是仍然遇到许多问题,将问题一一解决后,对此过程中做一些记录,方便大家查阅。

原理:

正如原创博客作者所说:
MySQL使用 LOAD DATA LOCAL INFILE 从文件中导入数据比insert语句要快,MySQL文档上说要快20倍左右。
但是这个方法有个缺点,就是导入数据之前,必须要有文件,也就是说从文件中导入。这样就需要去写文件,以及文件删除等维护。某些情况下,比如数据源并发的话,还会出现写文件并发问题,很难处理。

而且如果先写文件,又增加了IO操作,所以为了避免写文件:

MySQL社区提供这样一个方法:setLocalInfileInputStream(),此方法位于com.mysql.jdbc.PreparedStatement 类中。通过使用 MySQL JDBC 的setLocalInfileInputStream 方法实现从Java InputStream中load data local infile 到MySQL数据库中。可以达到同样的效率,直接从内存(IO流中)中导入数据,而不需要写文件

代码

代码(springboot),根据自己业务改造表及字段参数:

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Date;

/**
 * @Description :大数量量插入数据库,快速插入方法
 * @Author : 
 * @Date : 2021/3/4
 **/
@Component
public class LoadDataInFileUtil {

    private static final Logger logger = LoggerFactory.getLogger(LoadDataInFileUtil.class);
    private Connection conn = null;
    @Value("${spring.shardingsphere.datasource.shardingdb.driver-class-name}")
    private String driverName;
    @Value("${spring.shardingsphere.datasource.shardingdb.url}")
    private String url;
    @Value("${spring.shardingsphere.datasource.shardingdb.username}")
    private String userName;
    @Value("${spring.shardingsphere.datasource.shardingdb.password}")
    private String password;
    @Value("${database-name}")
    private String database;


    /**
     * 获取原生jdbc连接
     *
     * @return
     * @throws ClassNotFoundException
     * @throws SQLException
     */
    private Connection getConnection() throws ClassNotFoundException, SQLException {
        Class.forName(driverName);
        conn = DriverManager.getConnection(url, userName, password);
        return conn;
    }

    /**
     * 将数据从输入流加载到MySQL。
     *
     * @param loadDataSql SQL语句。
     * @param dataStream  输入流。
     * @return int         成功插入的行数。
     */
    private int bulkLoadFromInputStream(String loadDataSql,
                                        InputStream dataStream) throws SQLException, ClassNotFoundException {
        if (null == dataStream) {
            logger.info("输入流为NULL,没有数据导入。");
            return 0;
        }
        //做测试发现连接池好像不支持,所以获取原生的jdbc连接
        conn = getConnection();
        PreparedStatement statement = null;
        int result = 0;
        try {
            statement = conn.prepareStatement(loadDataSql);
//            mysql8  使用下面的方式
        /*if (statement.isWrapperFor(com.mysql.cj.jdbc.JdbcStatement.class)) {
            com.mysql.cj.jdbc.ClientPreparedStatement mysqlStatement = statement.unwrap(com.mysql.cj.jdbc.ClientPreparedStatement.class);
            mysqlStatement.setLocalInfileInputStream(dataStream);
            mysqlStatement.executeUpdate();
        }*/
//        mysql5 使用下面的方式
            if (statement.isWrapperFor(com.mysql.jdbc.Statement.class)) {
                com.mysql.jdbc.PreparedStatement mysqlStatement = statement.unwrap(com.mysql.jdbc.PreparedStatement.class);
                mysqlStatement.setLocalInfileInputStream(dataStream);
                result = mysqlStatement.executeUpdate();
            }
        } finally {
            if (statement != null) {
                statement.close();
            }
        }
        return result;
    }

    /**
     * 组装 SQL 语句
     *
     * @param dataBaseName 数据库名。
     * @param tableName    表名。
     * @param columnName   要插入数据的列名。
     */
    public String assembleSql(String dataBaseName, String tableName, String[] columnName) {
        String insertColumnName = StringUtils.join(columnName, ",");
        //插入方案的核心就是下面这个语句
        String sql = "LOAD DATA LOCAL INFILE 'sql.csv' IGNORE INTO TABLE " + dataBaseName + "." + tableName + " (" + insertColumnName + ")";
        return sql;
    }

    /**
     * 通过 LOAD DATA LOCAL INFILE 大批量导入数据到 MySQL。
     *
     * 原理是使用 setLocalInfileInputStream 会忽略 sql.csv 文件名,不从文件读取,直接从输入流读取数据
     * @param sql     SQL语句。
     * @param builder 组装好的数据。
     */
    public int fastInsertData(String sql, StringBuilder builder) {
        int rows = 0;
        InputStream is = null;
        try {
            byte[] bytes = builder.toString().getBytes();
            is = new ByteArrayInputStream(bytes);
            //批量插入数据。
            long beginTime = System.currentTimeMillis();
            rows = bulkLoadFromInputStream(sql, is);
            long endTime = System.currentTimeMillis();
            logger.info(" LOAD DATA LOCAL INFILE :【插入" + rows + "行数据至MySql中,耗时" + (endTime - beginTime) + "ms。】");
        } catch (SQLException | ClassNotFoundException e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != is) {
                    is.close();
                }
                if (null != conn) {
                    conn.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return rows;
    }

    /**
     * 批量插入方法
     *
     * @return
     */
    public void insertData(String dataBaseName, String tableName, String[] columns, StringBuilder tableColumnValue, int size) {
    	
        //插入语句
        String sql = assembleSql(dataBaseName, tableName, columns);
        //待插入行数,为了判断是否全部插入
        int mainTableInsertRows = fastInsertData(sql, tableColumnValue);
        //插入数据未全部成功视为全部失败
        if (mainTableInsertRows != size)
            System.println.out("插入失败!");
    }
    
}

测试代码:

@RunWith(SpringRunner.class)
@SpringBootTest
public class LoadDataInFileUtilTest {
    @Autowired
    LoadDataInFileUtil loadDataInFileUtil;
 
 	@Test
    public void testInsert() {
        //表列值
        StringBuilder tableColumnValue = new StringBuilder();
        //数据库名
        String dataBaseName = "dataBaseName ";
        //待插入的表
        String tableName = "tableName";
        // 要插入数据的列名。(必须与插入的数据一一对应)
        String[] columns = new String[]{"id","name","age"};
        
 		int number = 500000;
        int id = 1;
        for (int i = 0; i < number; i++) {
            //顺序对应上面的列名
            builderAppend(tableColumnValue, id);
            builderAppend(tableColumnValue ,"测试名字");
            builderEnd(tableColumnValue, 10);
        }

        long startTime = System.currentTimeMillis();
        loadDataInFileUtil.insertData(dataBaseName , tableName , columns, tableColumnValue, number);
        long endTime = System.currentTimeMillis();
        System.out.println("插入数据,所用时间:" + (endTime - startTime));

    }
    
   /**
     * 往 StringBuilder 里追加数据。
     *
     * @param builder StringBuilder。
     * @param object  数据。
     */
    public void builderAppend(StringBuilder builder, Object object) {
        builder.append(object);
        builder.append("\t");
    }
 
    /**
     * 往 StringBuilder 里追加一条数据的最后一个字段。
     *
     * @param builder StringBuilder。
     * @param object  数据。
     */
    public void builderEnd(StringBuilder builder, Object object) {
        builder.append(object);
        builder.append("\n");
   }
}

执行顺序:
从 insertData() 方法进入,首先通过assembleSql() 方法组织导入数据的sql语句“LOAD DATA LOCAL INFILE ‘sql.csv’ IGNORE INTO TABLE tableName()”,然后通过fastInsertData() 方法传入sql语句及需要导入的数据转化为流,最后通过setLocalInfileInputStream(dataStream) 将数据插入数据库。

性能对比

自己做了一些测试来对比效率,对比的是拼装批量插入方式和此方式,对比结果:

mysql的in太多怎么解决 mysql 大量in解决,mysql的in太多怎么解决 mysql 大量in解决_数据,第1张

本机做的测试,若在服务器上,效率更佳。

此方案数据量越大,差距越明显。

问题及解决办法

1.mysql版本问题
一般我们使用 mysql5 或mysql8 版本的驱动包,对应的预编译 prepareStatement 对象为:

//            mysql8  使用下面的方式
        /*if (statement.isWrapperFor(com.mysql.cj.jdbc.JdbcStatement.class)) {
            com.mysql.cj.jdbc.ClientPreparedStatement mysqlStatement = statement.unwrap(com.mysql.cj.jdbc.ClientPreparedStatement.class);
            mysqlStatement.setLocalInfileInputStream(dataStream);
            mysqlStatement.executeUpdate();
        }*/
//        mysql5 使用下面的方式
           /* if (statement.isWrapperFor(com.mysql.jdbc.Statement.class)) {
                com.mysql.jdbc.PreparedStatement mysqlStatement = statement.unwrap(com.mysql.jdbc.PreparedStatement.class);
                mysqlStatement.setLocalInfileInputStream(dataStream);
                result = mysqlStatement.executeUpdate();
            }*/

代码中有注释说明

2.字段拼接问题
因为字段值是通过 “\t” 来区别字段和 “\n” 来区别一条数据的,所以在字段的值中如果含有 Tab

3.值得问题
因为拼接的字段值都是已字符串的形式识别的,所以不存在true/false 这种数据,如果model类默认将数据库的tinyint类型转化为boolean了,需要单独建立model,使用0/1插入,其他字段同理。

4.是否成功
一批数据,如有无法插入的会跳过,不会报错,但会返回插入的条数,只能根据返回结果判定是否全部插入,来处理数据是否

5.数据库连接
因为我自己测试的时候,从dataSource获取的连接,似乎无法执行,所以获取了原生的jdbc连接来操作,注意操作完成时正确的关闭,否则造成资源浪费。

6.字段为空
如果字段为空,需要转化为“”,否则插入数据库以 null形式存在、

7.权限
操作数据库的账户需要开启 Excute权限

8.无法引用 com.mysql.jdbc.ClientPreparedStatement问题
因为之前在项目中引入jar包时,mysql驱动是以运行时编译的状态引入的,所以修改pom文件的scope

<dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>${mysql.version}</version>
                <!--去掉运行时依赖,以便能使用com.mysql.jdbc.ClientPreparedStatement -->
<!--            <scope>runtime</scope>-->
            </dependency>

以上为整个过程的使用及问题梳理,欢迎补充!



https://www.xamrdz.com/lan/5gt1921827.html

相关文章: