Merge branch 'ESPARK-112' into 'spark_2.1'

[ESPARK-112] 收集sql以及表依赖关系

实现思路就是通过event的方式来收集sql和依赖  
1、在sql执行的前收集所有sql  
2、依赖是在optimizer的阶段把依赖都解析出来，传到event里面，然后通过外置的listener去收集  
listener的地址是：https://git.elenet.me/dt-arch/spark-query-log
使用方式：
```shell
spark.extraListeners=com.eleme.tools.sql.SparkDependencyCollect,com.eleme.tools.sql.SparkSQLQueryLog
spark.sql.querylog.url=
spark.sql.querylog.user=
spark.sql.querylog.password=
#spark.listener.printError=false
#spark.collect.sqlDependency
```
resolve apache#112 

See merge request !94

Author: cenyuhai

core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala

@@ -20,6 +20,7 @@ package org.apache.spark.scheduler
 import java.util.Properties
 import javax.annotation.Nullable
 
+import scala.collection.mutable.HashSet
 import scala.collection.Map
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo
@@ -39,6 +40,17 @@ trait SparkListenerEvent {
   protected[spark] def logEvent: Boolean = true
 }
 
+@DeveloperApi
+case class SQLEvent(sql: String) extends SparkListenerEvent {
+  override protected[spark] def logEvent: Boolean = false
+}
+
+@DeveloperApi
+case class DependencyEvent(readTables: HashSet[String], writeTables: HashSet[String])
+    extends SparkListenerEvent {
+  override protected[spark] def logEvent: Boolean = false
+}
+
 @DeveloperApi
 case class TimeSeriesMetricEvent(executorId: String, name: String, stat: StatCounter)
   extends SparkListenerEvent

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

@@ -18,15 +18,14 @@
 package org.apache.spark.sql.execution
 
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{SaveMode, Strategy}
+import org.apache.spark.sql.{execution, SaveMode, Strategy}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.planning._
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical.{BroadcastHint, EventTimeWatermark, LogicalPlan}
 import org.apache.spark.sql.catalyst.plans.physical._
-import org.apache.spark.sql.execution
 import org.apache.spark.sql.execution.columnar.{InMemoryRelation, InMemoryTableScanExec}
 import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources._

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveOptimizerRules.scala

@@ -19,13 +19,16 @@ package org.apache.spark.sql.hive
 
 import java.io.IOException
 
+import scala.collection.mutable
 import scala.util.control.Breaks.{break, breakable}
 
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hive.common.StatsSetupConst
 
+import org.apache.spark.scheduler.DependencyEvent
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeSet, PredicateHelper, Rand}
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.command.DDLUtils
@@ -128,3 +131,24 @@ case class MergeSmallFiles(sparkSession: SparkSession) extends Rule[LogicalPlan]
     }
   }
 }
+
+case class DependencyCollect(sparkSession: SparkSession) extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = {
+    if (sparkSession.sparkContext.conf.getBoolean("spark.collectDependencies", true)) {
+      val readTables = mutable.HashSet[String]()
+      val writeTables = mutable.HashSet[String]()
+      plan transformDown {
+        case i @ InsertIntoTable(table: MetastoreRelation, _, _, _, _) =>
+          writeTables += s"${table.databaseName}.${table.tableName}"
+          i
+        case p @ PhysicalOperation(_, _, table: MetastoreRelation) =>
+          readTables += s"${table.databaseName}.${table.tableName}"
+          p
+      }
+      if (readTables.size > 0 || writeTables.size > 0) {
+        sparkSession.sparkContext.listenerBus.post(DependencyEvent(readTables, writeTables))
+      }
+    }
+    plan
+  }
+}

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.hive
 
+import org.apache.spark.scheduler.SQLEvent
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.analysis.Analyzer
 import org.apache.spark.sql.catalyst.optimizer.Optimizer
@@ -54,6 +55,9 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
   }
 
   override protected[sql] def auth(command: String): Unit = {
+    if (sparkSession.sparkContext.conf.getBoolean("spark.hive.sql.collect", true)) {
+      sparkSession.sparkContext.listenerBus.post(SQLEvent(command))
+    }
     if (!sparkSession.sparkContext.conf.getBoolean("spark.hive.auth.enable", true)) {
       return
     }
@@ -81,7 +85,9 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
     override def batches: Seq[Batch] = super.batches :+
       Batch("Determine stats of partitionedTable", Once,
         DeterminePartitionedTableStats(sparkSession)) :+
-      Batch("Merge small files when insert into hive tables", Once, MergeSmallFiles(sparkSession))
+      Batch("Merge small files when insert into hive tables", Once,
+        MergeSmallFiles(sparkSession)) :+
+      Batch("Collect read and write tables", Once, DependencyCollect(sparkSession))
   }
 
   /**