记一次 mysql插入千万级别数据过程

╰+攻爆jí腚メ 2024-04-17 05:35 191阅读 0赞

  • mysql插入千万级别数据

    • 背景

      • 第一次插入数据

        • 效果
      • 第二种方式：通过批处理方式

    • 第三种方式

      • 生成csv文件
      • 加载文件到mysql中

    • 最终方式

        • 注意点
    • 参考链接

mysql插入千万级别数据

背景

项目中期，需要进行性能测试，大数据平台要测试千万级别的数据进行处理，测试性能问题，所以需要对MySQL中插入千万级别的数据

第一次插入数据

最直接的想法，就是通过jdbc方式，将数据插入到MySQL中，因为原来造测试数据，进行自测时，也是同样的做法

  1. private static final Logger LOG = LoggerFactory.getLogger(CreateLeagalPersonTest.class);
  2.     private static Random rdn = new Random();
  3.     private static Connection conn = null;
  4.     private static int correctDataCount = 20000000;
  5.     private static int errorDataCount = 10000000;
  8.     @Before
  9.     public void before() throws ClassNotFoundException, SQLException {
  10.         //1、连接数据库
  11.         Class.forName("com.mysql.jdbc.Driver");
  12.         String url = "jdbc:mysql://10.117.5.223:3306/bigdata";
  13.         String user = "root";
  14.         String password = "123456";
  15.         conn = DriverManager.getConnection(url, user, password);
  16.         if (conn == null) {
  17.             LOG.error("连接失败...");
  18.             System.exit(1);
  19.         }
  20.     }
  22.     //主入口
  23.     @Test
  24.     public void test() throws SQLException {
  25.         //2、编写带？的SQL
  26.         String sql = "INSERT INTO tb_test_leagal_person (ORG_CODE, ORG_NAME ,ORG_ADDRESS ,ORG_TYPE_CODE, REG_NO ," +
  27.                 "REG_DATE ,STATUS ,STATUS_NAME ,ECNOMIC_CODE, ECNOMIC_CODE_NAME ,INDUSTRY_CODE ,INDUSTRY_CODE_NAME ," +
  28.                 "REG_CAPITAL ,UNI_SCID ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
  30.         PreparedStatement pst = conn.prepareStatement(sql);// 对带？的sql进行预编译
  31.         for (int index = 0; index < correctDataCount; index++) {
  33.             // 4、把?用具体的值进行代替
  34.             pst.setString(1, createORGCODE(index, true)); //组织机构代码
  35.             pst.setString(2, createORGNAME(index, true)); //组织机构名称
  36.             pst.setString(3, creteORGADDRESS(index, true)); // 组织地址
  37.             pst.setString(4, createORGTYPECODE(index, true)); //机构类型
  38.             pst.setString(5, createREGNO(index, true)); //注册号
  39.             pst.setDate(6, creteDate(index, true)); //成立日期
  40.             pst.setString(7, createSTATUS(index, true)); //状态
  41.             pst.setString(8, createSTATUSNAME(index, true)); //状态名称
  42.             pst.setString(9, createECNOMICCODE(index, true)); //经济类型
  43.             pst.setString(10, createECNOMIC_CODE_NAME(index, true)); //经济类型名称
  44.             pst.setString(11, createStringType("行业类别", index, true)); //行业类别
  45.             pst.setString(12, createStringType("行业类别名称", index, true)); //行业类别名称
  46.             pst.setInt(13, createIntType(index, true));//注册资金
  47.             pst.setString(14, createStringType("统一信用代码", index, true)); //同一信用代码
  49.             // 5、执行sql
  50.             int len = pst.executeUpdate();
  51.             System.out.println(len > 0 ? "添加成功 " + index : "添加失败");
  52.         }
  53.         // 6、释放资源
  54.         pst.close();
  55.         conn.close();
  56.     }

效果

很直观的发现，效率比较慢，需要进行优化

第二种方式：通过批处理方式

上面的做法是通过一条一条的记录插入的，效率比较低，那就改为批处理的

  1. //批处理插入
  2.     @Test
  3.     public void testInsertBatchData()  {
  4.         //2、编写带？的SQL
  5.         String sql = "INSERT INTO tb_test_leagal_person (ORG_CODE, ORG_NAME ,ORG_ADDRESS ,ORG_TYPE_CODE, REG_NO ," +
  6.                 "REG_DATE ,STATUS ,STATUS_NAME ,ECNOMIC_CODE, ECNOMIC_CODE_NAME ,INDUSTRY_CODE ,INDUSTRY_CODE_NAME ," +
  7.                 "REG_CAPITAL ,UNI_SCID ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
  9.         PreparedStatement pst = null;// 对带？的sql进行预编译
  10.         try {
  11.             pst = conn.prepareStatement(sql);
  12.             for (int index = 4450000; index < correctDataCount; index++) {
  14.                 // 4、把?用具体的值进行代替
  15.                 pst.setString(1, createORGCODE(index, true)); //组织机构代码
  16.                 pst.setString(2, createORGNAME(index, true)); //组织机构名称
  17.                 pst.setString(3, creteORGADDRESS(index, true)); // 组织地址
  18.                 pst.setString(4, createORGTYPECODE(index, true)); //机构类型
  19.                 pst.setString(5, createREGNO(index, true)); //注册号
  20.                 pst.setDate(6, creteDate(index, true)); //成立日期
  21.                 pst.setString(7, createSTATUS(index, true)); //状态
  22.                 pst.setString(8, createSTATUSNAME(index, true)); //状态名称
  23.                 pst.setString(9, createECNOMICCODE(index, true)); //经济类型
  24.                 pst.setString(10, createECNOMIC_CODE_NAME(index, true)); //经济类型名称
  25.                 pst.setString(11, createStringType("行业类别", index, true)); //行业类别
  26.                 pst.setString(12, createStringType("行业类别名称", index, true)); //行业类别名称
  27.                 pst.setInt(13, createIntType(index, true));//注册资金
  28.                 pst.setString(14, createStringType("统一信用代码", index, true)); //同一信用代码
  30.                 pst.addBatch();
  31.                 if (index % 5000 == 0) {
  32.                     pst.executeBatch();
  33.                     System.out.println(("插入成功..." + index));
  34.                     pst.clearBatch();
  35.                 }
  36. //            // 5、执行sql
  37. //            int len = pst.addBatch();
  38. //            System.out.println(len > 0 ? "添加成功 " + index : "添加失败");
  39.             }
  40.             pst.executeBatch();
  41.         } catch (SQLException e) {
  42.             e.printStackTrace();
  43.         }finally {
  44.             // 6、释放资源
  45.             try {
  46.                 pst.close();
  47.                 conn.close();
  48.             } catch (SQLException e) {
  49.                 e.printStackTrace();
  50.             }
  51.         }
  52.     }

第三种方式

同事说使用mysql 命令可以快速加载文件到mysql中

生成csv文件

  1. //写入文件方式
  2.     @Test
  3.     public void testWrtieToCSV() {
  4.         String outputPath = "dir/tb_test_leagal_person_2.csv";
  5.         doCreateCVSData(0, correctDataCount, true, outputPath);
  7.         doCreateCVSData(correctDataCount, correctDataCount + errorDataCount, false, outputPath);
  9.     }
  11.     private List<String> doCreateCVSData(int start, int end, boolean flag, String outputPath) {
  12.         List<String> result = new ArrayList<>();
  13.         for (int index = start; index < end; index++) {
  14.             StringBuffer sbuffer = new StringBuffer();
  15.             sbuffer.append(index);
  16.             sbuffer.append(",");
  17.             sbuffer.append(createORGCODE(index, flag)); //组织机构代码
  18.             sbuffer.append(",");
  19.             sbuffer.append(createORGNAME(index, flag)); //组织机构名称
  20.             sbuffer.append(",");
  21.             sbuffer.append(creteORGADDRESS(index, flag)); // 组织地址
  22.             sbuffer.append(",");
  23.             sbuffer.append(createORGTYPECODE(index, flag)); //机构类型
  24.             sbuffer.append(",");
  25.             sbuffer.append(createREGNO(index, flag)); //注册号
  26.             sbuffer.append(",");
  27.             sbuffer.append(creteDate(index, flag)); //成立日期
  28.             sbuffer.append(",");
  29.             sbuffer.append(createSTATUS(index, flag)); //状态
  30.             sbuffer.append(",");
  31.             sbuffer.append(createSTATUSNAME(index, flag)); //状态名称
  32.             sbuffer.append(",");
  33.             sbuffer.append(createECNOMICCODE(index, flag)); //经济类型
  34.             sbuffer.append(",");
  35.             sbuffer.append(createECNOMIC_CODE_NAME(index, flag)); //经济类型名称
  36.             sbuffer.append(",");
  37.             sbuffer.append(createStringType("行业类别", index, flag)); //行业类别
  38.             sbuffer.append(",");
  39.             sbuffer.append(createStringType("行业类别名称", index, flag)); //行业类别名称
  40.             sbuffer.append(",");
  41.             sbuffer.append(createIntType(index, flag));//注册资金
  42.             sbuffer.append(",");
  43.             sbuffer.append(createStringType("统一信用代码", index, flag)); //同一信用代码
  44.             System.out.println(sbuffer.toString());
  45.             result.add(sbuffer.toString());
  47.             if (result.size() % 1000000 == 0) {
  48.                 for (String s : result) {
  49.                     try {
  50.                         writeLinesToFile1(outputPath, s);
  51.                     } catch (Exception e) {
  52.                         e.printStackTrace();
  53.                     }
  54.                 }
  55.                 result.clear();
  56.             }
  57.         }
  58.         return result;
  59.     }
  61.     public static boolean writeLinesToFile1(String fileName, String line) throws Exception {
  62.         FileWriter fw = new FileWriter(fileName, true);
  63.         fw.write(line + "\n");
  64.         fw.flush();
  65.         fw.close();
  66.         return true;
  67.     }

加载文件到mysql中

  1. #mysql加载csv文件
  2. LOAD DATA LOCAL INFILE 'C:\\Users\\wangyg\\Desktop\\tb_test_leagal_person_2.csv' INTO TABLE bigdata.tb_test_leagal_person_2 FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n';

缺陷：

将数据写入csv文件时，同样会有大量的IO，导致所花费的时间很长，所以从某种角度来讲，时间还是一样长,不能解决问题

最终方式

最终。在网上查询资料，使用这种方式，2千万+数据 不到半小时就完成

  1. @Test
  2.     public void testInsert() throws SQLException, ClassNotFoundException {
  3.         initConn();
  4.         insert(0, 2000000, true);
  5. //        insert(correctDataCount, correctDataCount+errorDataCount, false);
  6.     }
  8.     private void initConn() throws ClassNotFoundException, SQLException {
  9.         //1、连接数据库
  10.         Class.forName("com.mysql.jdbc.Driver");
  11.         String url = "jdbc:mysql://10.117.5.223:3306/bigdata";
  12.         String user = "root";
  13.         String password = "123456";
  14.         conn = DriverManager.getConnection(url, user, password);
  15.         if (conn == null) {
  16.             LOG.error("连接失败...");
  17.             System.exit(1);
  18.         }
  19.     }
  21.     private static int singleCount = 100000;
  23.     //批量插入
  24.     public static void insert(int startCount, int endCount, boolean flag) {
  25.         // 开时时间
  26.         Long begin = new Date().getTime();
  27.         // sql前缀
  28.         String prefix = " INSERT INTO tb_test_leagal_person_3 (ORG_CODE, ORG_NAME ,ORG_ADDRESS ,ORG_TYPE_CODE, REG_NO ," +
  29.                 "STATUS ,STATUS_NAME ,ECNOMIC_CODE, ECNOMIC_CODE_NAME ,INDUSTRY_CODE ,INDUSTRY_CODE_NAME ," +
  30.                 "REG_CAPITAL ,UNI_SCID ) VALUES";
  31.         try {
  32.             // 保存sql后缀
  33.             StringBuffer suffix = new StringBuffer();
  34.             // 设置事务为非自动提交
  35.             conn.setAutoCommit(false);
  36.             // Statement st = conn.createStatement();
  37.             // 比起st，pst会更好些
  38.             PreparedStatement pst = conn.prepareStatement("");
  39.             // 外层循环，总提交事务次数
  40.             for (int i = 0; i <= endCount / singleCount; i++) {
  41.                 // 第次提交步长
  42.                 for (int j = 1; j <= singleCount; j++) {
  43.                     // 构建sql后缀
  44. //                    suffix.append("(" + j * i + ", SYSDATE(), " + i * j
  45. //                            * Math.random() + ",'" + randomStr(6) + "'),");
  46.                     int index = i * singleCount + j;
  47.                     //字符串类型添加'引号
  48.                     suffix.append("(");
  50.                     suffix.append("'");
  51.                     suffix.append(createORGCODE(index, flag));
  52.                     suffix.append("'");
  54.                     suffix.append(",");
  55.                     suffix.append("'");
  56.                     suffix.append(createORGNAME(index, flag));
  57.                     suffix.append("'");
  58.                     suffix.append(",");
  59.                     suffix.append("'");
  60.                     suffix.append(creteORGADDRESS(index, flag));
  61.                     suffix.append("'");
  62.                     suffix.append(",");
  63.                     suffix.append("'");
  64.                     suffix.append(createORGTYPECODE(index, flag)); //机构类型
  65.                     suffix.append("'");
  66.                     suffix.append(",");
  67.                     suffix.append("'");
  68.                     suffix.append(createREGNO(index, flag)); //注册号
  69.                     suffix.append("'");
  70. //                    suffix.append(",");
  71. //                    suffix.append(creteDate(index, flag)); //成立日期
  72.                     suffix.append(",");
  73.                     suffix.append("'");
  74.                     suffix.append(createSTATUS(index, flag)); //状态
  75.                     suffix.append("'");
  76.                     suffix.append(",");
  77.                     suffix.append("'");
  78.                     suffix.append(createSTATUSNAME(index, flag)); //状态名称
  79.                     suffix.append("'");
  80.                     suffix.append(",");
  81.                     suffix.append("'");
  82.                     suffix.append(createECNOMICCODE(index, flag)); //经济类型
  83.                     suffix.append("'");
  84.                     suffix.append(",");
  85.                     suffix.append("'");
  86.                     suffix.append(createECNOMIC_CODE_NAME(index, flag)); //经济类型名称
  87.                     suffix.append("'");
  88.                     suffix.append(",");
  89.                     suffix.append("'");
  90.                     suffix.append(createStringType("行业类别", index, flag)); //行业类别
  91.                     suffix.append("'");
  92.                     suffix.append(",");
  93.                     suffix.append("'");
  94.                     suffix.append(createStringType("行业类别名称", index, flag)); //行业类别名称
  95.                     suffix.append("'");
  96.                     suffix.append(",");
  97.                     suffix.append(createIntType(index, flag));//注册资金
  98.                     suffix.append(",");
  99.                     suffix.append("'");
  100.                     suffix.append(createStringType("统一信用代码", index, flag)); //同一信用代码
  101.                     suffix.append("'");
  102.                     suffix.append("),");
  104.                 }
  105.                 // 构建完整sql
  106.                 String sql = prefix + suffix.substring(0, suffix.length() - 1);
  107.                 pst.addBatch(sql);
  109.                 // 添加执行sql
  110.                 // 执行操作
  111.                 pst.executeBatch();
  112.                 // 提交事务
  113.                 conn.commit();
  114.                 System.out.println("提交成功..." + i * singleCount);
  115.                 // 清空上一次添加的数据
  116.                 suffix = new StringBuffer();
  117.             }
  118.             // 头等连接
  119.             pst.close();
  120.             conn.close();
  121.         } catch (SQLException e) {
  122.             e.printStackTrace();
  123.         }
  124.         // 结束时间
  125.         Long end = new Date().getTime();
  126.         // 耗时
  127.         System.out.println("cast : " + (end - begin) / 1000 + " ms");
  128.     }

注意点

  1. 创建表时，指定MYISAM

    CREATE TABLE bigdata.tb_test_leagal_person_2 (

    1. id INT AUTO_INCREMENT,
    2. ORG_CODE VARCHAR(9),
    3. ORG_NAME VARCHAR(140),
    4. ORG_ADDRESS VARCHAR(300),
    5. ORG_TYPE_CODE VARCHAR(100),
    6. REG_NO VARCHAR(100),
    7. `STATUS` VARCHAR(100),
    8. STATUS_NAME VARCHAR(100),
    9. ECNOMIC_CODE VARCHAR(100),
    10. ECNOMIC_CODE_NAME VARCHAR(100),
    11. INDUSTRY_CODE VARCHAR(100),
    12. INDUSTRY_CODE_NAME VARCHAR(100),
    13. REG_CAPITAL  INT(10),
    14. UNI_SCID VARCHAR(100),
    15. PRIMARY KEY (id)

    ) ENGINE MYISAM, CHARSET utf8,
    AUTO_INCREMENT 0;

  2. SQL拼接多个value值

  3. 设置事务为非自动提交
  4. SET GLOBAL max_allowed_packet=100 *1024*1024*20;

参考链接

https://blog.csdn.net/u010180815/article/details/79166502

https://blog.csdn.net/hewei314599782/article/details/80774162

发表评论

表情：
评论列表 （有 0 条评论，191人围观）

还没有评论，来说两句吧...

相关阅读