pyspark.SparkContext.setJobGroup¶
-
SparkContext。setJobGroup( groupId:str,描述:str,interruptOnCancel:bool=假 )→没有¶ -
分配一个组ID的所有工作开始由这个线程,直到组ID设置为不同的值或清除。
通常,一个执行单元在一个应用程序由多个引发操作或工作。应用程序员可以使用这种方法来组织一起工作,给一组描述。一旦设置,火花web UI将与这组助理这样的工作。
应用程序可以使用
SparkContext.cancelJobGroup ()在这组取消所有正在运行的工作。笔记
如果工作小组interruptOnCancel被设置为true,那么工作取消将导致Thread.interrupt()被称为工作的执行人线程。这是有用的帮助确保任务能够及时停止,但由于HDFS - 1208在默认情况下是关闭的,在HDFS可能回应Thread.interrupt()节点标记为死亡。
如果你运行并行工作,使用
pyspark.InheritableThread线程本地的产业。例子
> > >进口线程> > >从时间进口睡眠> > >从pyspark进口InheritableThread> > >结果=“未设置”> > >锁=线程。锁()> > >defmap_func(x):…睡眠(One hundred.)…提高RuntimeError(“任务应该被取消了”)> > >defstart_job(x):…全球结果…试一试:…sc。setJobGroup(“job_to_cancel”,“描述”)…结果=sc。并行化(范围(x))。地图(map_func)。收集()…除了异常作为e:…结果=“取消”…锁。释放()> > >defstop_job():…睡眠(5)…sc。cancelJobGroup(“job_to_cancel”)> > >抑制=锁。收购()> > >抑制=InheritableThread(目标=start_job,arg游戏=(10,))。开始()> > >抑制=InheritableThread(目标=stop_job)。开始()> > >抑制=锁。收购()> > >打印(结果)取消了