问题标签 [apache-spark]

我刚开始使用 Spark 0.7.2 和 Scala 2.9.3 进行编程。我正在一台独立机器上测试机器学习算法，算法的最后一步需要计算两个矩阵之间的 MSE（均方误差），即 || A - M||^2 并且我们在两个矩阵之间进行元素减法。由于 A 的潜在大小非常大且非常稀疏，我们将矩阵存储为 (key, value) 对，其中键是坐标 (i,j)，值是 A 的对应元素的元组， M，即（A_ij，M_ij）。整个 ML 算法是梯度下降，因此对于每次迭代，我们都会计算 MSE 并针对某个阈值对其进行测试。但是，整个程序运行正常，没有计算每次迭代的 MSE。这是程序的样子：

该程序最多只能运行 45 次迭代，它会因以下 Spark 异常而崩溃：

另一个观察结果是，对于每次迭代，运行时间将增加约 5%。同样没有“reduce(_ + _)”，就没有 StackOverflowError。我试图将并行度增加到可能的物理线程总数，但这无济于事。

真的很感谢任何人都可以指出一些方向，我可以找出堆栈溢出错误的根本原因。

编辑：

1. A 的类型是 spark.RDD[((Double, Double), Array[Double])]

2. stackoverflow 异常，它从 " at sun.reflect.GeneratedMethodAccessor4.invoke(Unknown Source)" 重复 61 次：

   /li>

3. 主要迭代代码

一些实用功能包含在下一个列表元素中

一些使用的实用程序功能：

我正在尝试在我的本地机器上运行一个简单的 Spark-Streaming 示例。
我有一个将 As/Bs/Cs 写入套接字的线程：

我的主程序，我尝试计算 As/Bs/Cs 的数量如下所示（没有减少步骤）：

在这种情况下，一切正常（批处理的示例输出）：

但是，如果我在地图之后添加减少步骤，它就不再起作用了。此代码在 texts.print() 之后

在这种情况下，我只得到第一个“流”变量和“文本”变量的输出，而没有任何 reduce。在第一次批处理之后也没有任何反应。我还将 spark 日志级别设置为 DEBUG，但没有遇到任何异常或其他奇怪的事情。

这里会发生什么？为什么我会被锁定？

我正在尝试在 Windows 机器上设置 Spark 集群。
去这里的方法是使用独立模式，对吧？
不使用 Mesos 或 YARN 的具体缺点是什么？使用其中任何一种会有多痛苦？有人在这里有经验吗？

当我尝试运行 Spark shell 时，这就是我得到的

仅供参考，它说它无法绑定的 IP 地址是 hadoop 上主机的 IP（与我运行它的机器不同）。我正在使用正确版本的 Hadoop 和 Scala，但不确定我做错了什么。任何帮助都会有所帮助！:)

我有两个想要加入的 RDD，它们看起来像这样：

碰巧 的键值rdd1是唯一的，而 的元组键值rdd2也是唯一的。我想加入这两个数据集，以便得到以下 rdd：

实现这一目标的最有效方法是什么？以下是我想到的一些想法。

选项1：

选项 2：

选项 1 将收集所有数据以掌握，对吗？因此，如果 rdd1 很大（在我的情况下它相对较大，尽管比 rdd2 小一个数量级），这似乎不是一个好的选择。选项 2 做了一个丑陋的独特的笛卡尔积，这似乎也非常低效。我想到的另一种可能性（但尚未尝试）是执行选项 1 并广播地图，尽管以“智能”方式广播会更好，以便地图的键与的键rdd2。

有没有人遇到过这种情况？我很高兴有你的想法。

谢谢！

在 Spark 中，可以设置一些 hadoop 配置设置，例如

这行得通，复制因子设置为 1。假设是这种情况，我认为这种模式（将“spark.hadoop.”添加到常规 hadoop 配置属性中）也适用于 textinputformat.record.delimiter：

然而，火花似乎只是忽略了这个设置。我是否textinputformat.record.delimiter以正确的方式设置？有没有更简单的方法来设置textinputformat.record.delimiter. 我想避免自己写InputFormat，因为我真的只需要获取由两个换行符分隔的记录。

在官方 spark 文档中，有一个用于foreach直接在 RDD 上的调用中的累加器示例：

我实现了自己的累加器：

在本地环境中，这工作得很好。但是，如果我在具有多台机器的 spark 独立集群上运行此作业，工作人员会抛出一个

在增加累加器的行myCounter。

我的问题是：累加器只能用于直接应用于 RDD 而不是嵌套函数的“顶级”匿名函数吗？如果是，为什么我的调用在本地成功而在集群上失败？

编辑：增加异常的详细程度。

RDD可以更新吗？--- 即假设我们从“a.txt”文件创建了一个RDD。“a.txt”已更新。是否可以在不读取整个文件 a.txt 的情况下更新 RDD？

我试图通过 EC2 的 spark 脚本在 EC2 集群设置上使用 distcp 在 Hadoop 和 Amazon S3 之间进行复制

我得到的错误是

尝试在 spark 上运行示例时出现错误。任何人都可以让我知道我需要对我的 pom.xml 做哪些更改才能使用 spark 运行程序。