private def singleDataSaveMysql(sql: String) ={
val dataFrame2: DataFrame=ss.sql(sql)
val resultRDD=df2rdd(dataFrame2)
val value: RDD[Map[String, Map[String, Map[String, String]]]]= resultRDD.map(diagLis =>{var diagLisMap: Map[String, Map[String, Map[String, String]]] =Map()
val diag: String=diagLis._1
val lisText: String=diagLis._2if (!diagLisMap.contains(diag)) {//用空行分割字符串
val itemResults: Array[String] = lisText.split("(\n|\r\n)\\s+")var itemSpecial: Map[String, Map[String, String]] =Map()for (i
val split1: Array[String]= itemResults(i).split("\n")var item: String = ""
var special: Map[String, String] =Map()if (split1.length > 1) for (j
item= split1(j).replaceAll("【下沙】", "")
}else{
val splits: Array[String]= split1(j).split("\t")if (splits.length > 2) {
val spell: String= splits(0).split(":")(0)
val betw: String= splits(1)
special+= (spell ->betw)
}
}
itemSpecial+= (item ->special)
}
}
diagLisMap+= (diag ->itemSpecial)
}
diagLisMap
})
val driver= "com.mysql.jdbc.Driver"val url= "jdbc:mysql://192.168.21.2351:3306/diagbot-app?useSSL=false"val username= "root"val password= "diagbot@20180822kwz"Class.forName(driver)//遍历测试
value.foreachPartition(l =>{
@transientvar connectionMqcrm =DriverManager.getConnection(url, username, password)
l.foreach(m =>{for (k
val diag: String=k._1
val iteamSpecal: Map[String, Map[String, String]]=k._2for (l
val iteam: String= l._1.toString().replace(":", "") //大项
val specails: Map[String, String] =l._2for (spe
val it: String=spe._1
val bet: String=spe._2
println(diag+ "\t" + iteam + "\t" + it + "\t" +bet)
val sql= "insert into doc_diag_lises(diag,iteam,it,bet) values ('" + diag + "','" + iteam + "','" + it + "','" + bet + "')"val statement: Statement=connectionMqcrm.createStatement()
statement.executeUpdate(sql)
}
}
}
})
connectionMqcrm.close()
})
}
转播
