小编宇宙人*_*宇宙人的帖子

如果我有一个矩阵,矩阵的每个元素都是一个非负数.我想从左下角到右上角穿过矩阵.在每一步中,我只能向上或向右移动,每个被访问的元素将被设置为0; 之后,我从右上角走到左下角,每一步我都只能向下或向左移动.

我的问题是如何有效地找到具有最大总和的路径.

在我的spark应用程序中,有一个object ResourceFactory包含ActorSystem用于提供资源客户端的akka .所以当我运行这个spark应用程序时,每个工作节点都会创建一个ActorSystem.问题是当spark应用程序完成其工作并关闭时.在ActorSystem仍保持活着每一个工作节点上,并阻止整个应用程序终止,它只是挂在.

有没有办法注册一些监听器,SparkContext以便sc在关闭时,ActorSystem每个工作节点上的每个工作节点都会收到通知自己关闭？

更新:

以下是简化的骨架:

有一个ResourceFactory,它是一个object,它包含一个actor system.它还提供了一种fetchData方法.

object ResourceFactory{
  val actorSystem = ActorSystem("resource-akka-system")
  def fetchData(): SomeData = ...
}

然后,有一个user-defined RDD类,在它的compute方法中,它需要从中获取数据ResourceFactory.

class MyRDD extends RDD[SomeClass] {
  override def compute(...) {
    ...
    ResourceFactory.fetchData()
    ...
    someIterator
  }
}

因此,在每个节点上都会有一个ActorSystem名为"resource-akka-system"的MyRDD节点,分布在这些工作节点上的那些实例可以从"resource-akka-system"获取数据.

问题是,当SparkContext关闭时,不需要那些"resource-akka-system",但我不知道如何在ResourceFactory关闭时通知关闭"resource-akka-system" SparkContext.所以现在,"resouce-akka-system"在每个工作节点上都保持活动状态,并阻止整个程序退出.

UPDATE2:

通过一些实验,我发现在本地模式下程序挂起,但在yarn-cluster …

我正在使用Spark Streaming 2.1.我想定期刷新一些缓存表(由spark提供的DataSource,如镶木地板,MySQL或用户定义的数据源).

1. 如何刷新表？

   假设我有一些表加载

   spark.read.format("").load().createTempView("my_table")

   它也被缓存

   spark.sql("cache table my_table")

   是否足以使用以下代码刷新表,并且当下次加载表时,它将自动被缓存

   spark.sql("refresh table my_table")

   或者我必须手动执行此操作

   spark.table("my_table").unpersist spark.read.format("").load().createOrReplaceTempView("my_table") spark.sql("cache table my_table")

2. 同时刷新表是否安全？

   并发我的意思是使用ScheduledThreadPoolExecutor除主线程之外的刷新工作.

   当我在桌面上调用refresh时Spark会使用缓存表会发生什么？

如何同时使用 maven-flatten-plugin 和 maven-shade-plugin？

我用revision，sha1，changelist来管理一个多模块项目的版本。

为了部署可消耗的工件，我使用 maven-flatten-plugin 生成一个扁平的 pom，使 ${revision} 具有实际价值。

但是 maven-shade-plugin 在 ${revision} 不变的情况下产生了一个减少的 pom。

如何指定 maven-shade-plugin 使用扁平化的 pom 来减少 pom。

问题是:假设我们有一组Sets : Set(1,2,3) Set(1,2,3,4) Set(4,5,6) Set(1,2,3,4,6),我们需要删除所有子集,最后得到Result : Set(4,5,6) Set(1,2,3,4,6). (由于两个Set(1,2,3)和Set(1,2,3,4)是的子集Set(1,2,3,4,6),这两种都被删除.)

并且假设集合的元素有order,可以是Int,Char等.

是否有可能以map-reduce方式进行？

以map-reduce方式执行此操作的原因是,有时Group的组具有非常大的大小,这使得无法在单个机器的内存中执行此操作.所以我们希望以map-reduce方式实现它,它可能效率不高,但只是工作.

我的问题是:

1. 我不知道如何在map-reduce进程中为键值对定义一个键,以便正确地对Set进行分组.

2. 我不知道何时应该完成该过程,所有子集都已被删除.

EDIT:

1. 未来数据的规模将继续扩大.

2. 输入可以是一组或多行,每行包含一组集.目前val data = RDD[Set],我首先做的是输入data.collect(),这导致整组的集合.但我可以将输入的生成修改为a RDD[Array[Set]],这将为我提供多行,每行包含一组集合.

3. 可以通过修改程序的其他部分来对每组中的元素进行排序.

我正在读"Scala编程"一书.在本书中,它说"函数文字被编译成一个在运行时实例化时是一个函数值的类".并且它提到"函数值是对象,因此如果您愿意,可以将它们存储在变量中".

所以我尝试检查函数之间的相等性.但我失败了.

1. 如果函数是Scala中的对象,那么它应该像Scala中的其他对象一样.也许检查功能的平等是没有意义的,所以它是禁用的？

2. 并将函数编译成Scala中的对象？

以下代码有效：

def bbb(v: Double => Unit)(a: Double): Unit = v(a)
bbb{v: Double => v == 0 }(5)
bbb{v: Double =>  Array(v) }(5)

但是，如果我bbb按如下方式重载，那么除非我为第一个bbb调用手动分配类型签名，否则它将无法正常工作：

def bbb(v: Double => Unit)(a: Double): Unit = v(a)
def bbb(v: Double => Array[Double])(a: Double): Array[Double] = v(a)
bbb{v: Double => v == 0 }(5) // bbb{(v => v == 0):(Double => Unit)}(5)
bbb{v: Double =>  Array(v) }(5)

我想使用spala 2.11.x akka 2.3.x的spray.io,我在spray.io的Project Info页面中找到以下内容:

spray 1.3.1 is built against Scala 2.10.3 and Akka 2.3.0 as well as Scala 2.11.1 and Akka 2.3.2.

当我使用喷雾的客户,我遇到了一些问题,然后我发现下面的spray.io的文档页面,其中喷雾客户端依赖于阿卡2.10.x:

akka-actor 2.2.x (with ‘provided’ scope, i.e. you need to pull it in yourself)

提供的范围意味着什么？如何将它与scala 2.11.x akka 2.3.x中编写的程序的其他部分一起使用？

编辑

以下是文档页面中列出的最简单的用例:

import akka.actor.ActorSystem
import scala.concurrent.Future
object main {
  def main(args: Array[String]) {
    import spray.http._
    import spray.client.pipelining._
    implicit val system = ActorSystem()
    import system.dispatcher // execution context for futures
    val pipeline: HttpRequest => Future[HttpResponse] = sendReceive
    val …