[jira] [Commented] (SPARK-33409) Spark job can not be killed in BoradcastNestedLoopJoin

["Apache Spark (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-33409?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17229168#comment-17229168 ] 

Apache Spark commented on SPARK-33409:
--------------------------------------

User 'constzhou' has created a pull request for this issue:
https://github.com/apache/spark/pull/30317

> Spark job can not be killed in BoradcastNestedLoopJoin
> ------------------------------------------------------
>
>                 Key: SPARK-33409
>                 URL: https://issues.apache.org/jira/browse/SPARK-33409
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 3.0.1
>            Reporter: zhou xiang
>            Priority: Major
>
>  
> If we kill a spark job in spark web UI, the task content will be marked interrupted, as the code below shows:
> {code:java}
> /**
>  * Kills a task by setting the interrupted flag to true. This relies on the upper level Spark
>  * code and user code to properly handle the flag. This function should be idempotent so it can
>  * be called multiple times.
>  * If interruptThread is true, we will also call Thread.interrupt() on the Task's executor thread.
>  */
> def kill(interruptThread: Boolean, reason: String): Unit = {
>   require(reason != null)
>   _reasonIfKilled = reason
>   if (context != null) {
>     context.markInterrupted(reason)
>   }
>   if (interruptThread && taskThread != null) {
>     taskThread.interrupt()
>   }
> }{code}
>  
> And spark will check the interrupt flag during the loop to stop it. Like this:
> {code:java}
>  /**
>  * :: DeveloperApi ::
>  * An iterator that wraps around an existing iterator to provide task killing functionality.
>  * It works by checking the interrupted flag in [[TaskContext]].
>  */
> @DeveloperApi
> class InterruptibleIterator[+T](val context: TaskContext, val delegate: Iterator[T])
>   extends Iterator[T] {
>   def hasNext: Boolean = {
>     // TODO(aarondav/rxin): Check Thread.interrupted instead of context.interrupted if interrupt
>     // is allowed. The assumption is that Thread.interrupted does not have a memory fence in read
>     // (just a volatile field in C), while context.interrupted is a volatile in the JVM, which
>     // introduces an expensive read fence.
>     context.killTaskIfInterrupted()
>     delegate.hasNext
>   }
>   def next(): T = delegate.next()
> }{code}
> In my case, there is a "not in" in my spark sql,  which leads to the "BoradcastNestedLoopJoin"
> The related code as below:
> {code:java}
> private def leftExistenceJoin(
>     relation: Broadcast[Array[InternalRow]],
>     exists: Boolean): RDD[InternalRow] = {
>   assert(buildSide == BuildRight)
>   streamed.execute().mapPartitionsInternal { streamedIter =>
>     val buildRows = relation.value
>     val joinedRow = new JoinedRow
>     if (condition.isDefined) {
>       streamedIter.filter(l =>
>         buildRows.exists(r => boundCondition(joinedRow(l, r))) == exists
>       )
>     } else if (buildRows.nonEmpty == exists) {
>       streamedIter
>     } else {
>       Iterator.empty
>     }
>   }
> }{code}
> The "streamedIter" and "buildRows" both have millions of records,  the executor get stuck in the join loop, I found something wrong in my sql and try to kill the job, but the executor thread is not interrupted. I have to restart the executor to stop it.
> I think we should also do this check: " context.killTaskIfInterrupted() "  in BoradcastNestedLoopJoin to support real cancel.
>  
>  
> {code:java}
>  {code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org