急中生智~利用Spark core完成"ETL"!

POOPE

背景介绍：　　今天接到老板分配的一个小任务：开发一个程序，实现从数据库中抽取数据并生成报表（这是我们数据库审计平台准备上线的一个功能）。既然是要生成报表，那么首先得有数据，于是便想到从该业务系统的测试环境抽取业务表的数据，然后装载至自己云主机上的Mysql中。
　　本来以为只要"select ...into outfile"和"load data infile..."两个命令就可以搞定的，可是还是出了意外。测试环境导出的
　　txt文件在云主机load时，报了"Row 1 doesn't contain data for all columns"这样的warning，表中的数据自然也是凌乱且不完整的。
　　仔细分析，感觉可能是两个方面出了问题：
　　1.由于测试环境的网段是隔离的，所以为了拿到"select ...into outfile"时生成的数据，我是打开CRT的日志，然后执行
　　"cat xxx.txt"，变相地将数据获取到了本地，然后上传至云主机的；
　　2.测试环境的Mysql和云主机上Mysql的小版本不一致。
　　这两个问题看似都没法解决，现在只有文本文件，怎么办？使用Spark不就得了！
　　之前也写过一篇使用Spark分析Mysql慢日志的博文，自己对Spark core的各种算子比较熟悉，所以决定试一试。
实战演练：　　表结构如下：

mysql> desc claims_case_loss_document;
+---------------+-------------+------+-----+---------+----------------+
| Field     | Type    | Null | Key | Default | Extra      |
+---------------+-------------+------+-----+---------+----------------+
| id      | int(11)   | NO   | PRI | NULL  | auto_increment |
| case_id     | varchar(22) | NO   |   | NULL  |        |
| case_times  | varchar(2)  | NO   |   | NULL  |        |
| document_list | text    | NO   |   | NULL  |        |
| create_time   | timestamp   | YES  |   | NULL  |        |
| update_time   | timestamp   | YES  |   | NULL  |        |
+---------------+-------------+------+-----+---------+----------------+
6 rows in set (0.00 sec)

　　文本结构如下：

1147  90100002700021437455  1     100100_收款方账户信息;001003_事故证明;001001_驾驶证;100000_收款方×××明;001002_索赔申请书   2017-11-16 12:08:08   2017-11-16 12:08:08

　　观察文本结构可知，每个字段间都有数个空格，而且两两字段间的空格数并不一致，所以得先使用Spark core将文本中字段提取出来，以便后续插入。
　　闲话少说，直接上程序！（以下程序均使用scala在eclipse ide for scala中编写和执行）

package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.collection.mutable.ArrayBuffer
import java.sql.DriverManager
object insert2Mysql {
  def main(args: Array[String]): Unit = {
  val t1=System.nanoTime()
  val conf = new SparkConf()
    .setAppName("insert2Mysql")
    .setMaster("local")
  val sc = new SparkContext(conf)
  //textFile方法只能读取字符集为utf-8的文件，否则中文会乱码。windows下，将文件另存为时，可以选择utf-8字符集
  //也可在代码中实施转换，但比较繁琐
  val lines = sc.textFile("D://Users//GAOZHONGZHENG186//Desktop//text001.txt", 1);
  val words = lines.map { line => line.split(" ") }
  val wordsNotNull = words.map{ word =>
             val wordArray_raw = new ArrayBuffer[String]()
             val wordArray = new ArrayBuffer[String]()
             for(i<-0 until word.length){
             if (word(i)!=""){
               wordArray_raw+=word(i)
             }
             }
             for(i<-0 until wordArray_raw.length-4){
             wordArray+=wordArray_raw(i)
             }
             wordArray+=wordArray_raw(4)+" "+wordArray_raw(5)
             wordArray+=wordArray_raw(6)+" "+wordArray_raw(7)
             wordArray
            }
  wordsNotNull.foreach { word =>
               Class.forName("com.mysql.cj.jdbc.Driver")
               val conn = DriverManager.getConnection("jdbc:mysql://10.25.80.7:3306/db1", "root", "123456")
               try {
                val statement = conn.createStatement()
                  val sql="insert into claims_case_loss_document values ("+
                      word(0)+","+
                      "'"+word(1)+"'"+","+
                      "'"+word(2)+"'"+","+
                      "'"+word(3)+"'"+","+
                      "'"+word(4)+"'"+","+
                      "'"+word(5)+"'"+")"
                  //执行插入
                  //println(sql)
                  statement.executeUpdate(sql)
                } catch{
                    case e:Exception =>e.printStackTrace
                     }
                  finally {
                      conn.close
                      }
             }
  val t2=System.nanoTime()
    //打印程序运行时间
  println((t2-t1)/1000000000 +"s")
  }
}

　　在插入的过程中，第一条记录总是会报错（后续语句插入正常），将eclipse中打印出的报错的insert语句手工粘贴至mysql执行时，仍报相同错误：

　　从图中可看出，1147的千位上的1确实发生了异常改变，而第二条语句中的1148是正常的，猜测可能是某个未知bug导致了第一条记录发生了异常改变。这个猜测在后续得到了证实：当把1147所在行从文本中删除后（此时1148所在行为第一条记录），1148所在行也报出同样的错误，而后续语句均可正常插入。
　　由于数据是作分析用的，所以丢失一条无伤大雅，而且这个bug实在诡异，这里就不再深究了。
　　细心的童鞋在看了代码后应该会问：数据插入的效率如何？实不相瞒，效率很差！5000条的数据足足用了近半个小时，即使是在这样的OLAP场景下，这样的效率也是不可容忍的！
　　仔细研究代码可发现，在对RDD调用foreach方法进行插入的时候，每一条记录都要创建一个连接，并且每一次insert都会在Mysql中触发一次commit操作（autocommit参数默认是打开的），这些都是很消耗资源的操作，插入效率自然很差。
　　发现这些问题后，针对代码进行了修改：

package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import java.sql.DriverManager
import scala.collection.mutable.ArrayBuffer
object insert2Mysql {
  def main(args: Array[String]): Unit = {
  val t1=System.nanoTime()
  val conf = new SparkConf()
    .setAppName("insert2Mysql")
    .setMaster("local")
  val sc = new SparkContext(conf)
  //textFile方法只能读取字符集为utf-8的文件，否则中文会乱码。windows下，将文件另存为时，可以选择utf-8字符集
  //也可在代码中实施转换，但比较繁琐
  val lines = sc.textFile("D://Users//GAOZHONGZHENG186//Desktop//text01.txt", 1);
  val words = lines.map { line => line.split(" ") }
  val wordsNotNull = words.map{ word =>
             val wordArray_raw = new ArrayBuffer[String]()
             val wordArray = new ArrayBuffer[String]()
             for(i<-0 until word.length){
             if (word(i)!=""){
               wordArray_raw+=word(i)
             }
             }
             for(i<-0 until wordArray_raw.length-4){
             wordArray+=wordArray_raw(i)
             }
             wordArray+=wordArray_raw(4)+" "+wordArray_raw(5)
             wordArray+=wordArray_raw(6)+" "+wordArray_raw(7)
             wordArray
            }
  
  val sqlRDD=wordsNotNull.map{ word =>  
                  val sql="insert into claims_case_loss_document values ("+
                         word(0)+","+
                         "'"+word(1)+"'"+","+
                         "'"+word(2)+"'"+","+
                         "'"+word(3)+"'"+","+
                         "'"+word(4)+"'"+","+
                         "'"+word(5)+"'"+")"
                    sql
                  }
  
  val sqlArray=sqlRDD.toArray()
    
  //加载驱动
  Class.forName("com.mysql.cj.jdbc.Driver") 
  val conn = DriverManager.getConnection("jdbc:mysql://10.25.80.7:3306/db1", "root", "123456")                 
  try {
    conn.setAutoCommit(false)
    val statement = conn.createStatement() 
    //这里有bug，处理出来的第一行格式都会报ERROR 1054 (42S22): Unknown column '1147' in 'field list'
    //为了避免程序跳出循环，所以循环从1开始，即从第2条开始插入
    for(i<-1 until sqlArray.length){
       //执行插入
      println(sqlArray(i))
      statement.executeUpdate(sqlArray(i))
      }
    conn.commit()
    } 
  catch{
      case e:Exception =>e.printStackTrace
      }   
  finally{
      conn.close
      }
  
  val t2=System.nanoTime()
  println((t2-t1)/1000000000 +"s")
  }
}

　　修改后的代码规避了上述缺陷，在同样插入5000条数据的情况下，只用了221s！效率大大提升！
　　到Mysql验证数据：

mysql> select count(*) from claims_case_loss_document;
+----------+
| count(*) |
+----------+
|   4999 |  --插入时跳过了第一条，所以为4999条
+----------+
1 row in set (0.00 sec)
mysql> select * from claims_case_loss_document limit 1\G
*************************** 1. row ***************************
       id: 1148
    case_id: 90100002700021437450
   case_times: 1
document_list: 100100_收款方账户信息;001003_事故证明;001001_驾驶证;100000_收款方×××明;001002_索赔申请书
  create_time: 2017-11-16 12:08:08
  update_time: 2017-11-16 12:08:08
1 row in set (0.00 sec)

　　至此，问题圆满解决！整个过程和数据仓库领域的ETL很接近，抽取-转换-装载，三个环节都有涉及，只是没有使用
　　kettle之类的工具罢了。
总结：　　在大数据时代，DBA应该积极做出改变，掌握一定开发技能，以便更好地适应时代变化，切不可固守自己的一亩三分地！
　　最后，给我们上海分组自研的数据库审计平台打个广告 ^.^
　　数据库审计平台是我们分组历时两年打造的产品，可用于Mysql、Oracle、Postgres等多种数据库，具备以下核心工能：
　　1.审计违规sql，前端一键生成报告
　　2.对相同功能点的sql可实现自动归类，方便后续统一整改
　　3.内嵌Percona toolkit，前端一键调用
　　4.一键抓取低效sql，并自动给出优化建议
还有很多很酷的功能就不一一介绍了，总之，谁用谁说好！感兴趣的DBA童鞋可以留言，可免费试用哦！

[MySQL] 急中生智~利用Spark core完成"ETL"!

相关帖子

动态在线扩容root根分区大小的方法详解

MySQL视图和索引专篇精讲

Java8的Lambda和排序

Spring Security登陆流程讲解

如何利用word制做思维导图？用word怎么制作思维导图？

机器学习服务文本翻译能力升级，中文直译模型让译文表达更地道！

利用CSS禁止复制网页内容

怎么利用配置文件来管理多个 Node.js 进程

uni-app 利用canvas画简单海报并保存图片

利用html+js触发弹窗事件

POOPE LV4