问题标签 [pyspark]

我正在通过 iPython Notebook 在我的本地机器上使用 Apache Spark 的 PySpark 对一些代码进行原型设计。我编写了一些似乎可以正常工作的代码，但是当我对其进行简单更改时，它就会中断。

下面的第一个代码块有效。第二个块因给定错误而失败。非常感谢任何帮助。我怀疑这个错误与序列化 Python 对象有关。错误说它不能 Pickle TestClass。我找不到有关如何使我的班级可以泡菜的信息。该文档说“通常，如果您可以腌制该对象的每个属性，则可以腌制任何对象。不能腌制类，函数和方法-如果腌制一个对象，则不会腌制该对象的类，而只是一个标识什么的字符串它所属的类。这对大多数泡菜都有效（但请注意关于泡菜长期储存的讨论）。“。我不明白这一点，因为我尝试用日期时间类替换我的 TestClass 并且一切似乎都很好。

无论如何，代码：

上面的代码返回以下内容：

347098

['测试：你好'，'测试：你好'，'测试：你好']

...完毕

但是，当我删除标记为“(1) watch this row”的行（最后一个地图行）并重新运行时，我得到了以下错误。它很长，所以在发布输出之前，我将在这里结束我的问题。再次，我真的很感激这方面的帮助。

提前致谢！

输出： 347098

我对这个输出感到惊讶fold，我无法想象它在做什么。

我希望这something.fold(0, lambda a,b: a+1)将返回 中的元素数量something，因为折叠从每个元素开始0并添加1。

我来自 Scala，其中 fold 就像我所描述的那样工作。那么 fold 应该如何在 pyspark 中工作？谢谢你的想法。

我正在使用 spark 处理数据，它适用于一天的数据（40G），但在一周的数据上出现OOM失败：

不同ID的数量少于10k。每个 ID 都很小int。作业失败是因为太多的执行者因 OOM 而失败。当工作成功时（在小输入上），"myoutput"大约是 100k。

1. 我究竟做错了什么？
2. 我尝试替换saveAsTextFile为collect（因为我实际上想在保存之前在 python 中进行一些切片和切块），行为没有区别，同样的失败。这是可以预料的吗？
3. 我曾经有reduce(lambda x,y: x.union(y), [sqc.parquetFile(...)...])而不是sc.union-哪个更好？它有什么不同吗？

该集群有25个节点，其中825GB RAM 和224个内核。

调用是spark-submit --master yarn --num-executors 50 --executor-memory 5G.

单个 RDD 有大约 140 列并涵盖一小时的数据，因此一周是 168(=7*24) 个 RDD 的并集。

当我使用引用对象的闭包过滤 RDD 时，我得到一个 pickle 错误。

没有对象：

与对象：

我究竟做错了什么？

在 pySpark MLlib 中似乎无法保存和加载回归模型，例如 LogisticRegressionModel、SVMModel、NaiveBayesModel 和 DecisionTreeModel。通过 JavaSaveable 和 JavaLoader mixins 对推荐模型MatrixFactorizationModel进行加载和保存，但回归模型不是这样完成的。

有没有办法通过提供我自己的负载和保存程序来解决这个问题？如果是这样，我将如何处理？

这个功能是否会在未来的版本中出现，或者 pySpark MLlib 是否会被逐步淘汰？

我有用于读取文本文件并用作内存中已注册临时表的工作代码。我想使用脚本或模块导入加载一组这些表，然后以交互方式查询它们。如果将此代码放入脚本和函数中，我应该返回哪个对象？sc 上下文？桌子？HadoopRDD？

是否支持窗口函数（例如first, last, lag, lead）pyspark？

例如，如何按一列分组并按另一列排序，然后通过 SparkSQL 或数据框为每个组选择第一行（就像窗口函数一样）？

我发现pyspark.sql.functions类包含聚合函数first和last，但它们不能用于groupBy类。

我的问题是当我试图从 asql.Row中读取数据时String。我正在使用 pyspark，但我听说人们对 Scala API 也有这个问题。

pyspark.sql.Row 对象是一个非常顽固的生物。抛出以下异常：

所以我们所拥有的是其中一个字段被表示为一个字节数组。以下 python 打印结构不起作用

还

两者都会导致 ClassCastException。

所以..其他人是怎么做到的？我开始自己动手（不幸的是，不能在这里复制/粘贴..）但这有点重新发明轮子..或者我怀疑。

我是 Spark-DataFrame API 的初学者。

我使用此代码将 csv 选项卡分隔加载到 Spark Dataframe

假设我使用 Spark 从新文件创建 DataFrame，并使用内置方法 toPandas() 将其转换为 pandas，

• 它是否将 Pandas 对象存储到本地内存？
• Pandas 低级计算是否全部由 Spark 处理？
• 它是否公开了所有 pandas 数据框功能？（我想是的）
• 我可以将它转换为 Pandas 并完成它，而无需太多接触 DataFrame API 吗？

注意到 PySpark 的一些奇怪行为，将不胜感激任何见解。

假设我有一个由简单元素组成的 RDD

现在我有兴趣在一个简单的类中捕获该 RDD 的不同属性，例如使用从每个元素rdd.map(lambda s: getattr(s,'name'))中提取属性。name

所以这个类的对象

将设置它们并从 RDD 中name获取相应的值。values

但是，这遇到了一个我认为集中在self.name表达式中lambda的错误。第二节课效果很好

我添加的只是前面的调用n=self.name并传递n到lambda而不是self.name.

那么我们无法评估的问题是self.name什么lambda？我在纯 python 中创建了类似的情况（self.name在 alambda中）并且没有错误，所以我认为这是特定于 Spark 的。谢谢你的想法。