Event Tracking: For statement #767

zhang1002 · 2021-07-08T11:19:01Z

Generate KyuubiStatementInfo. This object includes the following elements:

statement
statementId
appId
sessionId
exception: this statement's exception
physicalPlan
stateTime: contains each state and the time occurrence
sparkUser

Those data was packaged in KyuubiStatementInfo and we store those object in mem.
You can get some summary data from this object.

get each sql's physicalPlan and the starting time

1. Store the relationship between executionId and operationId 2. Store the relationship between operationId and statement 3. Store the relationship between executionId and physicalPlan 4. Store each state and its happen time for this statement: initialized, running, finished

Generate KStatement info. This object includes the following elements: 1. statement 2. statementId 3. appId 4. sessionId 5. executionId 6. physicalPlan 7. stateTime: contains each state and the time occurrence Those data was packaged in KStatement and we store those object in mem. You can get some summary data from this object.

yaooqinn · 2021-07-08T11:32:12Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

+    spark.sparkContext.applicationId,
+    session.getTypeInfo.identifier.toString,
+    OperationState.INITIALIZED.toString,
+    new Date().getTime)


System.currentTimeMillis?

2. Get each state by the function: setState 3. Get this statement's physicalPlan in ExecuteStatement 4. Add sparkUser item 5. Remove java code

Add newline in KyuubiStatementInfo

codecov-commenter · 2021-07-09T03:58:01Z

Codecov Report

Merging #767 (3c8d9af) into master (82fd58f) will decrease coverage by 0.14%.
The diff coverage is 43.47%.

@@             Coverage Diff              @@
##             master     #767      +/-   ##
============================================
- Coverage     79.27%   79.12%   -0.15%     
  Complexity       10       10              
============================================
  Files           130      132       +2     
  Lines          5090     5117      +27     
  Branches        657      658       +1     
============================================
+ Hits           4035     4049      +14     
- Misses          694      707      +13     
  Partials        361      361

Impacted Files	Coverage Δ
.../engine/spark/monitor/KyuubiStatementMonitor.scala	`0.00% <0.00%> (ø)`
...ine/spark/monitor/entity/KyuubiStatementInfo.scala	`0.00% <0.00%> (ø)`
...uubi/engine/spark/operation/ExecuteStatement.scala	`86.41% <100.00%> (+1.91%)`	⬆️
...g/apache/kyuubi/engine/spark/KyuubiSparkUtil.scala	`92.30% <0.00%> (+1.39%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 82fd58f...3c8d9af. Read the comment docs.

ulysses-you · 2021-07-09T04:23:40Z

...kyuubi-spark-monitor/src/main/scala/org.apache.spark.kyuubi/entity/KyuubiStatementInfo.scala

+    var physicalPlan: String,
+    sparkUser: String,
+    var exception: String,
+    var stateTimeMap: Map[String, Long])


if this field can be modified, we can add into class.

case class { var xxx: Option[String]; }

and same with stateTimeMap, I'd like move it into class

ulysses-you · 2021-07-09T04:25:03Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

@@ -58,6 +61,12 @@ class ExecuteStatement(

  private val operationListener: SQLOperationListener = new SQLOperationListener(this, spark)

+  var kyuubiStatementInfo = KyuubiStatementInfo(


var -> private val

ulysses-you · 2021-07-09T04:26:09Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

+
+  override def setOperationException(opEx: KyuubiSQLException): Unit = {
+    KyuubiStatementMonitor.addOperationExceptionByOperationId(statementId, opEx.toString)
+    super.setOperationException(opEx)


please keep the same order.

super.setOperationException(opEx) KyuubiStatementMonitor.addOperationExceptionByOperationId(statementId, opEx.toString)

ulysses-you · 2021-07-09T04:26:48Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

+  var kyuubiStatementInfo = KyuubiStatementInfo(
+    statementId, statement, spark.sparkContext.applicationId,
+    session.getTypeInfo.identifier.toString, null,
+    spark.sparkContext.sparkUser, null, Map(state.toString->lastAccessTime))


Map(state.toString->lastAccessTime) -> Map(state.toString -> lastAccessTime)

ulysses-you · 2021-07-09T04:27:20Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

@@ -19,7 +19,10 @@ package org.apache.kyuubi.engine.spark.operation

 import java.util.concurrent.{RejectedExecutionException, ScheduledExecutorService, TimeUnit}

-import org.apache.spark.kyuubi.SQLOperationListener
+import scala.collection.mutable.Map


I think we don't need mutable Map here

ulysses-you · 2021-07-09T04:28:37Z

...als/kyuubi-spark-monitor/src/main/scala/org.apache.spark.kyuubi/KyuubiStatementMonitor.scala

+//  endStateList.add(OperationState.CANCELED.toString)
+//  endStateList.add(OperationState.CLOSED.toString)
+//  endStateList.add(OperationState.ERROR.toString)
+//  endStateList.add(OperationState.TIMEOUT.toString)


please remove unused code

ulysses-you · 2021-07-09T04:28:50Z

...als/kyuubi-spark-monitor/src/main/scala/org.apache.spark.kyuubi/KyuubiStatementMonitor.scala

+
+  def addEachStateTimeForOperationid(operationId: String, state: String, time: Long): Unit = {
+    operationStatementMap.get(operationId).stateTimeMap.put(state, time)
+    print(1231)


please remove unused code

ulysses-you · 2021-07-09T04:29:27Z

...als/kyuubi-spark-monitor/src/main/scala/org.apache.spark.kyuubi/KyuubiStatementMonitor.scala

+
+import org.apache.spark.kyuubi.entity.KyuubiStatementInfo
+
+object KyuubiStatementMonitor {


it's better to add some comment at class description.

2. Add some comment

yaooqinn · 2021-07-09T13:28:54Z

...kyuubi-spark-monitor/src/main/scala/org.apache.spark.kyuubi/entity/KyuubiStatementInfo.scala

+    statement: String,
+    appId: String,
+    sessionId: String,
+    var physicalPlan: String,


I think we can track queryExecution directly in memory

yaooqinn · 2021-07-12T04:00:44Z

externals/kyuubi-spark-monitor/src/main/scala/org/apache/kyuubi/KyuubiStatementMonitor.scala


 import java.util.concurrent.ConcurrentHashMap

-import org.apache.spark.kyuubi.entity.KyuubiStatementInfo
+import org.apache.kyuubi.entity.KyuubiStatementInfo


org.apache.kyuubi.engine.spark.monitor.KyuubiStatementInfo

2. code optimizing

yaooqinn · 2021-07-13T04:12:10Z

How about adding some unit tests?

yaooqinn · 2021-07-13T04:14:43Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+    statement: String,
+    appId: String,
+    sessionId: String,
+    var physicalPlan: QueryExecution,


var queryExecution: QueryExecutio

yaooqinn · 2021-07-13T04:15:25Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+    appId: String,
+    sessionId: String,
+    var physicalPlan: QueryExecution,
+    sparkUser: String,


remove this, we can manage this in session info

yaooqinn · 2021-07-13T04:16:12Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+    var physicalPlan: QueryExecution,
+    sparkUser: String,
+    var exception: KyuubiSQLException,
+    stateTimeMap: Map[OperationState, Long])


stateToTime

yaooqinn · 2021-07-13T04:18:09Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

+
+  override def setState(newState: OperationState): Unit = {
+    super.setState(newState)
+     kyuubiStatementInfo.stateTimeMap.put(newState, lastAccessTime)


yaooqinn · 2021-07-13T04:20:14Z

...rk-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

+    statementId, statement, spark.sparkContext.applicationId,
+    session.getTypeInfo.identifier.toString, null,
+    spark.sparkContext.sparkUser, null, Map(state -> lastAccessTime))
+  KyuubiStatementMonitor.addStatementDetailForOperationId(statementId, kyuubiStatementInfo)


could this lead to memory leak?

yaooqinn · 2021-07-13T05:58:22Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+    statementId: String,
+    statement: String,
+    appId: String,
+    sessionId: String,


use session handle?

2. Add some comment 3. Code optimization 4. Use queue to store kyuubiStatementInfo

2. Add PrivateMethodTester to get private method for unit test 3. Optimize some code

ulysses-you · 2021-07-14T09:15:17Z

externals/kyuubi-spark-monitor/src/test/resources/test-scheduler-pool.xml

+        <weight>1</weight>
+        <schedulingMode>FAIR</schedulingMode>
+    </pool>
+</allocations>


why we need this file ?

ulysses-you · 2021-07-14T09:16:07Z

...k-monitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/KyuubiStatementMonitor.scala

+import org.apache.kyuubi.engine.spark.monitor.entity.KyuubiStatementInfo
+
+// TODO: Thread Safe need to consider
+object KyuubiStatementMonitor extends Logging{


nit: Logging{ -> Logging {

ulysses-you · 2021-07-14T09:19:38Z

...k-monitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/KyuubiStatementMonitor.scala

+   */
+  // TODO: Lack size type threshold and time type threshold
+  def putStatementInfoIntoQueue(kyuubiStatementInfo: KyuubiStatementInfo): Unit = {
+    if (kyuubiStatementQueue.size() >= 7) {


I'd like to define a constant:

val STATEMEMT_CAPACITY = 10 val FORCE_DUMP_STATEMENT_THREDHOLD = 7 ...

It just for test. We will make those configurable soon.

ulysses-you · 2021-07-14T09:20:20Z

...k-monitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/KyuubiStatementMonitor.scala

+   * and dumpping them to a file by threshold.
+   */
+  // TODO: Need ensure those items have finished. If not, we should put them into this queue again.
+  def removeAndDumpStatementInfoFromQueue(): Unit = {


private[kyuubi]

ulysses-you · 2021-07-14T09:20:50Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+ */
+case class KyuubiStatementInfo(
+    statementId: String,
+    statement: String,


let's say: query: String

ulysses-you · 2021-07-14T09:21:33Z

...nitor/src/main/scala/org/apache/kyuubi/engine/spark/monitor/entity/KyuubiStatementInfo.scala

+    appId: String,
+    sessionId: HandleIdentifier,
+    var queryExecution: QueryExecution,
+    var exception: KyuubiSQLException,


can we move these var inside class ?

ulysses-you · 2021-07-14T09:25:24Z

...k-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/KyuubiStatementMonitorSuite.scala

+      var iterator = kyuubiStatementQueue.iterator()
+      while (iterator.hasNext) {
+        val kyuubiStatementInfo = iterator.next()
+        assert(kyuubiStatementInfo.statement !== null)


can we check the real value directly ? != null seems unreliable

ok, i will do it

ulysses-you · 2021-07-14T09:27:42Z

...k-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/KyuubiStatementMonitorSuite.scala

+    val getQueue = PrivateMethod[
+      ArrayBlockingQueue[KyuubiStatementInfo]](Symbol("kyuubiStatementQueue"))()
+    val kyuubiStatementQueue = KyuubiStatementMonitor.invokePrivate(getQueue)
+    kyuubiStatementQueue.clear()


is it better to add a new method reset in KyuubiStatementMonitor ?

This method is just for unit test.
In pro_env, we don't need this method.

ulysses-you · 2021-07-14T11:26:10Z

thanks, merging to master for v1.3.0

zhang1002 added 8 commits July 7, 2021 13:34

event tracking: for SQLExecutionStart

d167626

get each sql's physicalPlan and the starting time

Merge master branch and resolve conflict

b26345a

remove some unused conf

5a62f58

format change

6f0be54

Merge branch 'master' into branch-1.2_spark-monitor

5552261

merge master branch

f43b3c8

yaooqinn reviewed Jul 8, 2021

View reviewed changes

zhang1002 added 2 commits July 9, 2021 11:30

1. Remove the relationship between executionId and operationId

8bddb20

2. Get each state by the function: setState 3. Get this statement's physicalPlan in ExecuteStatement 4. Add sparkUser item 5. Remove java code

Format change:

f364d43

Add newline in KyuubiStatementInfo

ulysses-you reviewed Jul 9, 2021

View reviewed changes

1. Remove some unused code

6b87dff

2. Add some comment

zhang1002 closed this Jul 9, 2021

zhang1002 reopened this Jul 9, 2021

Add some comment

71d33b9

yaooqinn reviewed Jul 9, 2021

View reviewed changes

Change directory structure

fe1f7cc

yaooqinn reviewed Jul 12, 2021

View reviewed changes

1. change directory structure

3c8d9af

2. code optimizing

yaooqinn reviewed Jul 13, 2021

View reviewed changes

zhang1002 added 2 commits July 14, 2021 13:17

1. Add unit test

5ce5058

2. Add some comment 3. Code optimization 4. Use queue to store kyuubiStatementInfo

1. Remove KyuubiInfo

9ce9cf6

2. Add PrivateMethodTester to get private method for unit test 3. Optimize some code

yaooqinn previously approved these changes Jul 14, 2021

View reviewed changes

Add some logs

6608127

zhang1002 dismissed yaooqinn’s stale review via 6608127 July 14, 2021 07:45

Add log4j

9b9b2f7

zhang1002 closed this Jul 14, 2021

zhang1002 added 2 commits July 14, 2021 16:48

1. Modify unit test: clear kyuubiStatementInfoQueue first

74f5ebc

Modify some comment

885dd01

zhang1002 reopened this Jul 14, 2021

ulysses-you reviewed Jul 14, 2021

View reviewed changes

Code optimize

40acc5c

yaooqinn approved these changes Jul 14, 2021

View reviewed changes

ulysses-you merged commit ff52b20 into apache:master Jul 14, 2021

ulysses-you assigned zhang1002 Jul 14, 2021

ulysses-you added the kind:feature Feature request label Jul 14, 2021

ulysses-you added this to the v1.3.0 milestone Jul 14, 2021

pan3793 added the release highlight label Sep 3, 2021

zhang1002 deleted the branch-1.2_spark-monitor branch September 24, 2021 16:34

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Event Tracking: For statement #767

Event Tracking: For statement #767

zhang1002 commented Jul 8, 2021 •

edited

Loading

yaooqinn Jul 8, 2021

codecov-commenter commented Jul 9, 2021 •

edited

Loading

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

ulysses-you Jul 9, 2021

yaooqinn Jul 9, 2021

yaooqinn Jul 12, 2021

yaooqinn commented Jul 13, 2021 •

edited

Loading

yaooqinn Jul 13, 2021

yaooqinn Jul 13, 2021

yaooqinn Jul 13, 2021

yaooqinn Jul 13, 2021

yaooqinn Jul 13, 2021

yaooqinn Jul 13, 2021

ulysses-you Jul 14, 2021

ulysses-you Jul 14, 2021

ulysses-you Jul 14, 2021

zhang1002 Jul 14, 2021

ulysses-you Jul 14, 2021

ulysses-you Jul 14, 2021

ulysses-you Jul 14, 2021

zhang1002 Jul 14, 2021

ulysses-you Jul 14, 2021

zhang1002 Jul 14, 2021

ulysses-you Jul 14, 2021

zhang1002 Jul 14, 2021

ulysses-you commented Jul 14, 2021

		@@ -58,6 +61,12 @@ class ExecuteStatement(

		private val operationListener: SQLOperationListener = new SQLOperationListener(this, spark)

		var kyuubiStatementInfo = KyuubiStatementInfo(


		import org.apache.spark.kyuubi.entity.KyuubiStatementInfo

		object KyuubiStatementMonitor {

Event Tracking: For statement #767

Event Tracking: For statement #767

Conversation

zhang1002 commented Jul 8, 2021 • edited Loading

Choose a reason for hiding this comment

codecov-commenter commented Jul 9, 2021 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yaooqinn commented Jul 13, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ulysses-you commented Jul 14, 2021

zhang1002 commented Jul 8, 2021 •

edited

Loading

codecov-commenter commented Jul 9, 2021 •

edited

Loading

yaooqinn commented Jul 13, 2021 •

edited

Loading