我有一个小型Scala程序,可以在单个节点上运行.但是,我正在扩展它,因此它在多个节点上运行.这是我的第一次尝试.我只是想了解RDD如何在Spark中工作,所以这个问题是基于理论的,可能不是100%正确.
假设我创建了一个RDD:
val rdd = sc.textFile(file)
现在,一旦我这样做了,这是否意味着文件at file现在在节点之间进行分区(假设所有节点都可以访问文件路径)？
其次,我想计算RDD中的对象数量(足够简单),但是,我需要在需要应用于RDD中的对象的计算中使用该数字 - 伪代码示例:
rdd.map(x => x / rdd.size)
假设有100个对象rdd,并且说有10个节点,因此每个节点有10个对象的计数(假设这是RDD概念的工作方式),现在当我调用该方法时,每个节点将使用rdd.sizeas 执行计算10还是100？因为,总的来说,RDD是大小100但在每个节点上本地只是10.我是否需要在进行计算之前制作广播变量？这个问题与下面的问题有关.
最后,如果我转换到RDD,例如rdd.map(_.split("-")),然后我想要新size的RDD,我是否需要在RDD上执行操作,例如count(),所以所有信息都被发送回驱动程序节点？
我需要将RDD分成两部分:
满足条件的1部分; 另一部分没有.我可以filter在原始RDD上做两次但看起来效率低下.有没有办法可以做我想要的事情？我在API和文献中都找不到任何东西.
我在本地计算机上运行此代码:
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object SimpleApp {
def main(args: Array[String]) {
val logFile = "/Users/username/Spark/README.md"
val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2).cache()
val numAs = logData.filter(line => line.contains("a")).count()
val numBs = logData.filter(line => line.contains("b")).count()
println("Lines with a: %s, Lines with b: %s".format(numAs, numBs))
}
}
我想运行该程序,但在不同的文件上运行它 - 它目前只在README.md上运行.如何在运行Spark(或其他任何参数？)时传递另一个文件的文件路径.例如,我想contains("a")换个字母.
我让程序运行:
$ YOUR_SPARK_HOME/bin/spark-submit \
--class "SimpleApp" \
--master local[4] \
target/scala-2.10/simple-project_2.10-1.0.jar
谢谢!
我需要将变量传递给包含的Blade文件.我试过这两种方式; 然而,两者都没有成功.
将变量传递title给包含的文件:
@section('left')
@include('modal', ['title' => 'Hello'])
@stop
使用@yield并设置部分:
@section('left')
@include('modal')
@section('title')
Hello
@stop
@stop
我正在使用Laravel 4.2.我不知道我想做什么是可能的,但我想是的.
我运行了一个ALS推荐系统程序,该程序在Apache Spark网站上找到,该程序使用了Mllib.当使用评级为1-5的数据集时(我使用过MovieLens数据集),它会给出预测评级超过5的建议!我在小测试中发现的最高值是7.4.显然,我要么误解代码的意图,要么出错了.我研究过潜在因子推荐系统,并且认为Spark Mlib ALS实现基于此.
为什么它会使得收视率高于可能的收视率呢？这没有道理.
我误解了算法还是程序存在缺陷？
machine-learning collaborative-filtering apache-spark apache-spark-mllib
我有一个Scala程序,可以在一台计算机上正常工作.但是,我想让它在多个节点上工作.
程序的开头看起来像这样:
val filename = Source.fromFile("file://...")
val lines = filename.getLines
val linesArray = lines.map(x => x.split(" ").slice(0, 3))
val mapAsStrings = linesArray.toList.groupBy(_(0)).mapValues(x => x.map(_.tail))
val mappedUsers = mapAsStrings map {case (k,v) => k -> v.map(x => x(0) -> x(1).toInt).toMap}
当我尝试使用Spark运行程序时,我知道我需要一个SparkContext和SparkConf对象,并且它们用于创建RDD.
所以现在我有:
class myApp(filePath: String) {
private val conf = new SparkConf().setAppName("myApp")
private val sc = new SparkContext(conf)
private val inputData = sc.textFile(filePath)
inputData现在是一个RDD,它在之前的程序中是等价的filename(我假设).对于一个RDD方法是不同的.那么,相当于getLines什么？或者没有相应的？我很难想象RDD给我的工作内容,例如是 …
我的程序,我在不同的集群上运行了很多次突然停止.日志:
15/04/20 19:19:59 INFO scheduler.TaskSetManager: Finished task 12.0 in stage 15.0 (TID 374) in 61 ms on ip-XXX.compute.internal (16/24)
15/04/20 19:19:59 INFO storage.BlockManagerInfo: Added rdd_44_14 in memory on ip-XXX.compute.internal:37999 (size: 16.0 B, free: 260.6 MB)
Killed
"被杀"是什么意思,为什么会发生？没有其他错误.
如果我分区RDD为60,并且我在20台机器上共有20个核心,即20个单核机器实例,则任务数量为60(等于分区数量).为什么这对每个核心有一个分区并且有20个任务有益？
另外,我已经运行了一个实验,我将分区数设置为2,检查UI显示任何时候运行的2个任务; 然而,让我感到惊讶的是,它在完成任务时切换实例,例如node1和node2执行前2个任务,然后node6和node8执行下一组2个任务等.我认为通过将分区数设置为小于在集群中的核心(和实例)然后程序将只使用所需的最小实例数.谁能解释这种行为？
在使用 MVVM 模式时，当Model对象变得复杂时，即当它们包含非原始/非内置属性时，我会遇到一些麻烦。在我的特定实例中，我有一个ModelA包含ModelB对象集合的对象，这些对象本身包含一个ModelC对象集合：
class ModelA
{
public string Name { get; set; }
public OberservableCollection<ModelB> Bs { get; set; }
}
class ModelB
{
public string Make { get; set; }
public ObservableCollection<ModelC> Cs { get; set; }
}
class ModelC
{
public string Brand{ get; set; }
}
我有一个ModelAViewModel允许访问收集的ModelB Bs财产。在这种情况下，我没有创建ViewModelfor ModelB。我使用s为ModelB和ModelC集合（和单个实例）设置了样式DataTemplate：
<DataTemplate x:Key="modelATemplate">
<Grid Margin="5">
<Grid.RowDefinitions>
<RowDefinition /> …在Apache Spark上运行我的迭代程序时偶尔会收到消息:
INFO scheduler.DAGScheduler: Missing parents for Stage 4443: List(Stage 4441, Stage 4442)
我收集它意味着它需要计算父母RDD- 但我不是100%肯定.我不只是得到其中一个,我最终得到的是100个,如果不是一次数千个 - 它完全减慢了我的程序,另一个迭代没有完成10-15分钟(它们通常需要4-10秒) .
我cache主要RDD在每次迭代时使用StorageLevel.MEMORY_AND_DISK_SER.下一次迭代使用它RDD.因此,谱系RDD因此变得非常大,因此需要缓存.但是,如果我正在缓存(并且溢出到磁盘),父母怎么会丢失？
我一直有错误Task not Serializable.
我已经创建了一个小类,它扩展了Serializable - 当我需要序列化的值时,我认为这就是我的意思.
class SGD(filePath : String) extends Serializable {
val rdd = sc.textFile(filePath)
val mappedRDD = rdd.map(x => x.split(" ")
.slice(0,3))
.map(y => Rating(y(0).toInt, y(1).toInt, y(2).toDouble))
.cache
val RNG = new Random(1)
val factorsRDD = mappedRDD(x => (x.user, (x.product, x.rating)))
.groupByKey
.mapValues(listOfItemsAndRatings =>
Vector(Array.fill(2){RNG.nextDouble}))
}
最后一行总是会导致Task not Serializable错误.我不明白的是:Class是Serializable; 并且,Class Random也是根据API可序列化的.那么,我做错了什么？我一直无法得到这样的东西去工作; 因此,我想我的理解是错误的.我一直被告知Class必须是Serializable ......好吧它仍然不起作用!？