Spark读取分析在ES中存储的SQL

用户通过elasticsearch-sql对存储在elasticsearch中的数据进行查询，假设事先会把查询语句保存在elasticsearch中，那么如何对这些sql语句中涉及到的表进行统计？

Spark读取Elasticsearch

import org.elasticsearch.spark._
val esOptions = Map("es.nodes"->"localhost", "es.port"->"9200","es.mapping.date.rich"->"false")
val esRDD = spark.sparkContext.esRDD("collectorapimetricslog2-2020.12/logs", esOptions)
esRDD.take(20).foreach(println)
val esJsonRDD = esRDD.map(x=>{
  import org.json4s._
  import org.json4s.JsonDSL._
  import org.json4s.jackson.JsonMethods._
  import org.json4s.jackson.Serialization
  import org.json4s.DefaultFormats
  implicit val json4sFormats = DefaultFormats
  val origM = x._2
  Serialization.writePretty(origM)
})
val esDF = spark.read.json(esRDD)

用RDD方式把query语句从es中读取出来，转换为json串之后，再转换为DataFrame。

那为什么不直接采用Elasticsearch-Hadoop中提供的Dataframe接口方式， 原因在于使用DataFrame方式直接读取，会有多种格式不匹配或出错的问题出现，elasticsearch-hadoop在兼容性方面，还有许多细节考虑不周。

JSqlParser

使用JSqlParser把query语句中涉及到的表找出来

第一步， 加载jsqlparser库

bin/spark-shell --packages "com.github.jsqlparser:jsqlparser:3.1"

第二步， 分析使用的代码，先去除识别上错误，然后parse

import net.sf.jsqlparser.util.TablesNamesFinder._
import net.sf.jsqlparser.util.TablesNamesFinder
import net.sf.jsqlparser.parser.CCJSqlParserUtil
import net.sf.jsqlparser.statement.select._
val stmt = CCJSqlParserUtil.parse("select * from tabl1 a join tab2 b on a.id=b.id")
val sel = stmt.asInstanceOf[Select]
val tblFinder = new TablesNamesFinder()
tblFinder.getTableList(sel)

val esQueryContentDF = esDF.filter("engine=='es'").select("queryContent")
val parsedQueryDF = esQueryContentDF.map(r => {
    import net.sf.jsqlparser.util.TablesNamesFinder._
    import net.sf.jsqlparser.util.TablesNamesFinder
    import net.sf.jsqlparser.parser.CCJSqlParserUtil
    import net.sf.jsqlparser.statement.select._
    import spark.implicits._
    import scala.collection.JavaConverters._
    var targetTable:String = "exception"
    val originalQuery = r.getString(0)
    try {
        val sQuery = r.getString(0)
        val dateHistoPattern = "date_histogram(?:.*[)])".r
        val sQuery2 = dateHistoPattern.replaceAllIn(sQuery,"date_histogram()")
        val qPattern = raw"(w+-[d.]+)".r
        val queryStr = qPattern.replaceAllIn(sQuery2,"`$1`")
        val stmt = CCJSqlParserUtil.parse(queryStr)
        val sel = stmt.asInstanceOf[Select]
        val tblNamesFinder = new TablesNamesFinder()
        val tblLst = tblNamesFinder.getTableList(sel)
        targetTable = tblLst.asScala.mkString(",")
    }catch {
        case ex: Exception => {
            targetTable = "exception: " + originalQuery
        }
    }
    targetTable
})

parsedQueryDF.filter(" value not like 'exception%'").createOrReplaceTempView("parsed_query")
spark.sql("select split(replace(value,'`',''),'-')[0] from parsed_query").distinct.collect.foreach(println)