统计效果：
代码部分：
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{Logging, SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, _}
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.types.StringType import scala.collection.mutable.ArrayBuffer
/**
* 功能：对hive表的列信息进行统计。
* 统计结果包括：
* 1.包含每列的平均值，中位数，最小值最大值，方差，唯一值，缺失值，列类型。
* 2.列的直方图分布（字符串top10，数值列10个区间），四分位图分布（数值列）。
*
* 实现逻辑：
* 1.利用spark的describe函数获取到最大值，最小值，均值，方差等。
* 2。利用sql：获取唯一值及缺失值，sql样例如下：
* select count(distinct(id)) as unique_id , count(distinct(name)) as unique_name, sum(case when id is null then 1 else 0 end) as missing_id, sum(case when name is null then 1 else 0 end) as missing_name, sum(1) as totalrows from zpcrcf
* 结果：
* +---------+-----------+----------+------------+---------+
* |unique_id|unique_name|missing_id|missing_name|totalrows|
* +---------+-----------+----------+------------+---------+
* | 14| 12| 0| 0| 14|
* +---------+-----------+----------+------------+---------+
*
* 3.利用sql：获取四分位图，sql样例如下；
* select 'Quartile_id' as colName, ntil, max(id) as num from (select id, ntile(4) OVER (order by id)as ntil from zpcrcf) tt group by ntil
* 结果：
* +------------+----+---+
* | colName|ntil|num|
* +------------+----+---+
* | Quartile_id| 1| 3|
* | Quartile_id| 2| 7|
* | Quartile_id| 3| 14|
* | Quartile_id| 4|100|
*
* 4.数值型直方图10阶段分区间，通过最大值减最小值，获取各个区间内的分布。
* sql样例如下：
* select 'MathHistogram_age' as colName, partNum, count(1) as num from ( select age, (case when (age >= 29.0 and age <= 36.1) then 1 when (age > 36.1 and age <= 43.2) then 2 when (age > 43.2 and age <= 50.3) then 3 when (age > 50.3 and age <= 57.4) then 4 when (age > 57.4 and age <= 64.5) then 5 when (age > 64.5 and age <= 71.6) then 6 when (age > 71.6 and age <= 78.69999999999999) then 7 when (age > 78.69999999999999 and age <= 85.8) then 8 when (age > 85.8 and age <= 92.9) then 9 when (age > 92.9 and age <= 100.0) then 10 else 0 end ) as partNum from zpcrcf) temptableScala group by partNum
* 结果：
* +-----------------+-------+---+
* | colName|partNum|num|
* +-----------------+-------+---+
* |MathHistogram_age| 0| 1|
* |MathHistogram_age| 1| 3|
* |MathHistogram_age| 10| 10|
* | MathHistogram_id| 1| 10|
* | MathHistogram_id| 2| 3|
* | MathHistogram_id| 10| 1|
* +-----------------+-------+---+
*
*
*
* Created by zpc on 2016/4/26.
*/
object DataFrameVisiualize extends Logging { def runforstatistic(hiveContext: HiveContext, params: JSONObject) = {
val arr = params.getJSONArray("targetType")
var i = 0
while( arr != null && i < arr.size()){
val obj = arr.getJSONObject(i)
if("dataset".equalsIgnoreCase(obj.getString("targetType"))){
val tableNameKey = obj.getString("targetName")
val tableName = params.getString(tableNameKey)
val user = params.getString("user")
run(hiveContext, tableName, user)
}
i = i+1
}
} def run(hiveContext: HiveContext, tableName: String, user: String) = {
val pathParent = s"/user/$user/mlaas/tableStatistic/$tableName"
// val conf = new SparkConf().setAppName("DataFrameVisiualizeJob")
// val sc = new SparkContext(conf)
// val hiveContext = new HiveContext(sc)
// val sqlContext = new SQLContext(sc)
//0.获取DB的schema信息
val schemadf = hiveContext.sql("desc " + tableName)
//schema信息落地
val filePathSchema = pathParent + "/schemajson"
schemadf.write.mode(SaveMode.Overwrite).format("json").save(filePathSchema) //1.加载表到dataframe
val df = hiveContext.sql("select * from " + tableName)
//2.获取dataframe的describe信息，默认为获取到的都为数值型列
val dfdesc = df.describe()
// //3.描述信息落地
// val filePath = pathParent + "/describejson"
// des.write.mode(SaveMode.Overwrite).format("json").save(filePath)
// val dfdesc = sqlContext.read.format("json").load(filePath) //4.列信息区分为mathColArr 和 strColArr
val mathColArr = dfdesc.columns.filter(!_.equalsIgnoreCase("summary"))
val (colMin, colMax, colMean, colStddev, colMedian) = getDesfromDF(dfdesc, mathColArr)
val allColArr = df.columns // col type 存在vector类型,此处仅统计string和num类型的
val typeMap = df.dtypes.toMap
val strColArr = allColArr.filter(typeMap.get(_).get.equals(StringType.toString))
// val strColArr = allColArr.filter(!_.equalsIgnoreCase("summary")).diff(mathColArr) saveRecords(hiveContext, tableName, 100, pathParent + "/recordsjson")
val jsonobj = getAllStatistics(hiveContext, tableName, allColArr, strColArr, mathColArr, 10, colMin, colMax) jsonobj.put("colMin", colMin)
jsonobj.put("colMax", colMax)
jsonobj.put("colMean", colMean)
jsonobj.put("colStddev", colStddev)
jsonobj.put("colMedian", colMedian) val jsonStr = jsonobj.toString
val conf1 = new Configuration()
val fs = FileSystem.get(conf1)
val fileName = pathParent + "/jsonObj"
val path = new Path(fileName)
val hdfsOutStream = fs.create(path)
hdfsOutStream.write(jsonStr.getBytes("utf-8"))
hdfsOutStream.flush()
hdfsOutStream.close()
// fs.close(); } def saveRecords(hiveContext: HiveContext, tableName: String, num: Int, filePath: String) : Unit = {
hiveContext.sql(s"select * from $tableName limit $num").write.mode(SaveMode.Overwrite).format("json").save(filePath)
}
/**
* 根据allCols， mathColArr， strColArr 三个数组，返回带有所有统计信息（除去已经根据describe获取到的）的dataframes。
* 返回的dataframe结果进行遍历，填充各个属性的值。
*/
def getAllStatistics(hiveContext: HiveContext, tableName: String, allColArr: Array[String], strColArr: Array[String], mathColArr: Array[String], partNum: Int, colMin: java.util.HashMap[String, Double], colMax: java.util.HashMap[String, Double]) :
JSONObject = {
val jsonobj = new JSONObject()
val sb = new StringBuffer()
sb.append("select ")
allColArr.map{col => sb.append(s"count(distinct(`$col`)) as unique_$col ," +
s"sum(case when `$col` is null then 1 else 0 end) as missing_$col, ")}
sb.append(s"sum(1) as totalrows from $tableName")
val df = hiveContext.sql(sb.toString)
val colUnique = new java.util.HashMap[String, Long]//唯一值
val colMissing = new java.util.HashMap[String, Long]//缺失值
var totalrows = 0L
df.take(1).foreach(row => (totalrows = row.getAs[Long]("totalrows"), jsonobj.put("totalrows", totalrows) ,allColArr.foreach(col => (colUnique.put(col, row.getAs[Long]("unique_"+col)),colMissing.put(col, row.getAs[Long]("missing_"+col))) ) )) val dfArr = ArrayBuffer[DataFrame]()
val strHistogramSql = new StringBuffer()
strHistogramSql.append(s"""
SELECT tta.colName, tta.value, tta.num
FROM (
SELECT ta.colName, ta.value, ta.num, ROW_NUMBER() OVER (PARTITION BY ta.colName ORDER BY ta.num DESC) AS row
FROM (
""") var vergin = 0
for(col <- strColArr){
if(vergin == 1){
strHistogramSql.append(" UNION ALL ")
}
vergin = 1
strHistogramSql.append(s"""
SELECT 'StrHistogram_$col' AS colName, `$col` AS value, COUNT(1) AS num
FROM $tableName
GROUP BY `$col` """)
}
strHistogramSql.append(s"""
) ta
) tta
WHERE tta.row <= $partNum
""")
//整个表中，可能不存在字符串型的列。此时，sql是不完整的,添加到df中会报错。
if(strColArr != null && strColArr.size != 0 ){
val dfStrHistogram = hiveContext.sql(strHistogramSql.toString)
dfArr.append(dfStrHistogram)
} for(col <- mathColArr) {
val df1 = hiveContext.sql(s"select 'Quartile_$col' as colName, ntil, bigint(max(`$col`)) as num from (select `$col`, ntile(4) OVER (order by `$col`)as ntil from $tableName) tt group by ntil ")
log.info("col is :" + col + ", min is :" + colMin.get(col) + ", max is : " + colMax.get(col))
// when the column data contains null, the min and max may be null or be "Infinity".
if (colMin == null || colMin.get(col) == null || colMax.get(col) == null || colMax.get(col) == "Infinity" || colMin.get(col) == "Infinity") {
log.info("col is :" + col + ", min is :" + colMin.get(col) + ", max is : " + colMax.get(col))
} else {
//need toString first, then toDouble。 or：ClassCastException
val min = colMin.get(col).toString.toDouble
val max = colMax.get(col).toString.toDouble
val df2 = getHistogramMathDF(col, hiveContext, tableName, min, max, partNum)
dfArr.append(df1)
dfArr.append(df2)
}
}
//可能存在没有列可统计的情况， e.g. 表中的列都为double，但数据都是null.
//dfArr.reduce 和会报错：java.lang.UnsupportedOperationException: empty.reduceLeft
//总行数为0时，四分位，条形图也自然获取不到，且会出现NullPointerException。
if(dfArr.isEmpty || totalrows == 0L){
jsonobj.put("colUnique", colUnique)
jsonobj.put("colMissing", colMissing)
}else {
val dfAll = dfArr.reduce(_.unionAll(_))
val allRows = dfAll.collect()
val mathColMapQuartile = new java.util.HashMap[String, Array[java.util.HashMap[String, Long]]] //四分位
val mathColMapHistogram = new java.util.HashMap[String, Array[java.util.HashMap[String, Long]]] //条形图
val strColMapHistogram = new java.util.HashMap[String, Array[java.util.HashMap[String, Long]]] //条形图
val (mathColMapQuartile1, mathColMapHistogram1, strColMapHistogram1) = readRows(allRows)
for (col <- strColArr) {
strColMapHistogram.put(col, strColMapHistogram1.get(col).toArray[java.util.HashMap[String, Long]])
}
for (col <- mathColArr) {
mathColMapQuartile.put(col, mathColMapQuartile1.get(col).toArray[java.util.HashMap[String, Long]])
mathColMapHistogram.put(col, mathColMapHistogram1.get(col).toArray[java.util.HashMap[String, Long]])
}
jsonobj.put("mathColMapQuartile", mathColMapQuartile)
jsonobj.put("mathColMapHistogram", mathColMapHistogram)
jsonobj.put("strColMapHistogram", strColMapHistogram)
jsonobj.put("colUnique", colUnique)
jsonobj.put("colMissing", colMissing)
}
jsonobj
}
def readRows(rows: Array[Row]) : (java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]] , java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]], java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]])={
val mathColMapQuartile = new java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]] //四分位
val mathColMapHistogram = new java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]]//条形图
val strColMapHistogram = new java.util.HashMap[String, ArrayBuffer[java.util.HashMap[String,Long]]]//条形图
rows.foreach( row => {
val colName = row.getAs[String]("colName")
if (colName.startsWith("StrHistogram")) {
val value = row.getAs[String](1)
val num = row.getAs[Long](2)
val map = new java.util.HashMap[String, Long]()
val col = colName.substring(colName.indexOf('_') + 1)
map.put(value, num)
val mapValue = strColMapHistogram.get(col)
if (mapValue == null) {
val mapValueNew = ArrayBuffer[java.util.HashMap[String, Long]]()
mapValueNew.append(map)
strColMapHistogram.put(col, mapValueNew)
} else {
mapValue.append(map)
strColMapHistogram.put(col, mapValue)
}
} else if (colName.toString.startsWith("Quartile")) {
val value = row.get(1).toString
val num = row.getAs[Long](2)
val map = new java.util.HashMap[String, Long]()
val col = colName.substring(colName.indexOf('_') + 1)
map.put(value, num)
val mapValue = mathColMapQuartile.get(col)
if (mapValue == null) {
val mapValueNew = ArrayBuffer[java.util.HashMap[String, Long]]()
mapValueNew.append(map)
mathColMapQuartile.put(col, mapValueNew)
} else {
mapValue.append(map)
mathColMapQuartile.put(col, mapValue)
}
} else if (colName.toString.startsWith("MathHistogram")) {
val value =row.get(1).toString
val num = row.getAs[Long](2)
val map = new java.util.HashMap[String, Long]()
val col = colName.substring(colName.indexOf('_') + 1)
map.put(value, num)
val mapValue = mathColMapHistogram.get(col)
if (mapValue == null) {
val mapValueNew = ArrayBuffer[java.util.HashMap[String, Long]]()
mapValueNew.append(map)
mathColMapHistogram.put(col, mapValueNew)
} else {
mapValue.append(map)
mathColMapHistogram.put(col, mapValue)
}
}
})
(mathColMapQuartile, mathColMapHistogram, strColMapHistogram)
}
/** 数值型的列的条形分布获取方法*/
def getHistogramMathDF(col : String, hiveContext: HiveContext, tableName: String, min: Double, max: Double, partNum: Int) : DataFrame = {
val len = (max - min) / partNum
log.info(s"len is : $len")
val sb = new StringBuffer()
sb.append(s"select `$col`, (case ")
val firstRight = min + len
sb.append(s" when (`$col` >= $min and `$col` <= $firstRight) then 1 ")
for (i <- 2 until (partNum + 1)) {
val left = min + len * (i - 1)
val right = min + len * i
sb.append(s" when (`$col` > $left and `$col` <= $right) then $i ")
}
sb.append(s" else 0 end ) as partNum from $tableName")
sb.insert(0, s"select 'MathHistogram_$col' as colName, partNum, count(1) as num from ( ")
sb.append(") temptableScala group by partNum")
log.info("getHistogram is: " + sb.toString)
val df = hiveContext.sql(sb.toString)
df
}
def getDesfromDF(dfdesc : DataFrame, mathColArr: Array[String]):
(java.util.HashMap[String, Double], java.util.HashMap[String, Double], java.util.HashMap[String, Double], java.util.HashMap[String, Double], java.util.HashMap[String, Double])= {
val allRows = dfdesc.collect()
val colMin = new java.util.HashMap[String, Double]//最小值
val colMax = new java.util.HashMap[String, Double]//最大值
val colMean = new java.util.HashMap[String, Double]//平均值
val colStddev = new java.util.HashMap[String, Double]//标准差
val colMedian = new java.util.HashMap[String, Double]//中位值
allRows.foreach(row => {
val mapKey = row.getAs[String]("summary")
for(col <- mathColArr){
if("mean".equalsIgnoreCase(mapKey)){
colMean.put(col, row.getAs[Double](col))
}else if("stddev".equalsIgnoreCase(mapKey)){
colStddev.put(col, row.getAs[Double](col))
}else if("min".equalsIgnoreCase(mapKey)){
log.info("col is " + col +", min is : "+ row.getAs[Double](col))
colMin.put(col, row.getAs[Double](col))
}else if("max".equalsIgnoreCase(mapKey)){
log.info("col is " + col +", max is : "+ row.getAs[Double](col))
colMax.put(col, row.getAs[Double](col))
}else{
colMedian.put(col, row.getAs[Double](col))
}
}
})
(colMin, colMax, colMean, colStddev, colMedian)
}
}