问题标签 [sparkr]

我有一个 DataFrame x。它包含 ID 和日期。所以

给出 ID = 1 2 ... 和 Date = "2012-05-05" "2015-01-01" ...

我想创建一个新的数据集，其中删除所有日期高于“2013-01-01”+ P 的 ID。P 是天，例如 P = 3 天。

在 sparkR 我这样做

但 sparkR 不接受这一点。

请原谅我的简单问题，但我对 Spark/Hadoop 比较陌生。

我正在尝试将一堆小的 CSV 文件加载到 Apache Spark 中。它们目前存储在 S3 中，但如果这样可以简化操作，我可以在本地下载它们。我的目标是尽可能有效地做到这一点。让一些单线程主机下载和解析一堆 CSV 文件，而我的几十个 Spark 工作人员却无所事事，这似乎是一种耻辱。我希望有一种惯用的方式来分发这项工作。

CSV 文件排列在一个目录结构中，如下所示：

我有两年的数据，每天都有目录，每个里面都有几百个 CSV。所有这些 CSV 都应该具有相同的架构，但当然有可能一个 CSV 有问题，如果有几个有问题的文件，我不希望整个工作崩溃。只要在某处发生的日志中通知我，就可以跳过这些文件。

似乎我想到的每个 Spark 项目都是这种形式，我不知道如何解决。（例如，尝试读入一堆制表符分隔的天气数据，或者读入一堆日志文件来查看这些数据。）

我试过的

我已经尝试过 SparkR 和 Scala 库。我真的不在乎我需要使用哪种语言。我对使用正确的成语/工具更感兴趣。

纯斯卡拉

我最初的想法是枚举parallelize所有year/mm-dd组合的列表，这样我就可以让我的 Spark 工作人员每天都独立处理（下载并解析所有 CSV 文件，然后将它们堆叠在一起（unionAll()）以减少它们）。不幸的是，使用spark-csv库下载和解析 CSV 文件只能在“父”/主作业中完成，而不是从每个子作业中完成，因为Spark 不允许作业嵌套。因此，只要我想使用 Spark 库进行导入/解析，这将不起作用。

混合语言

当然，您可以使用该语言的原生 CSV 解析来读取每个文件，然后将它们“上传”到 Spark。在 R 中，这是一些包的组合，用于将文件从 S3 中取出，然后是 a read.csv，最后以 acreateDataFrame()将数据放入 Spark。不幸的是，这真的很慢，而且似乎倒退了我希望 Spark 的工作方式。如果我的所有数据在进入 Spark 之前都通过 R 管道传输，为什么还要使用 Spark？

Hive/Sqoop/Phoenix/Pig/Flume/Flume Ng/s3distcp

我已经开始研究这些量身定制的工具，很快就不知所措。我的理解是，许多/所有这些工具都可用于将我的 CSV 文件从 S3 获取到 HDFS。

当然，从 HDFS 读取我的 CSV 文件会比 S3 更快，这样可以解决部分问题。但是我仍然有数以万计的 CSV 需要解析，并且我不知道在 Spark 中执行此操作的分布式方式。

我在 sparkR 中有 100 个 DataFrame。

我想将它们收集在一个新的 DataFrame 中。一种方法是这样

现在 'all' 是一个 DataFrame 并包含 100 个其他 DataFrame。这需要一些时间在 SparkR 中运行，我想知道是否有更好的方法来做到这一点？

假设我们有一个包含 ID 的数据集“数据”，并且对于每个 ID，我们有许多“分数”。例如，对于 ID=1，我们的分数为 25,41,23,...,33。

对于用户 1，我想计算该用户分数的总和。在 sparkR 我这样做

现在“ss”给了我 ID=1 的“分数”的总和。我想创建一个数据集，在其中获得所有 ID 的分数总和。

例如，在 RI 中会这样做

但这在 sparkR 中不起作用。它说“向量分配中的类型/长度无效（S4/0）”。有什么好的方法吗？

我有一个 DataFrame X。在 sparkR 中，我可以通过键入找到第一行

现在我想找到最后一个。'last' 是一个聚合函数。

但是当我在 sparkR 中键入 last(X) 时，我收到此消息 Error in (function (classes, fdef, mtable) : unable to find an inherit method for function 'last' for signature '"DataFrame"' 这怎么可能以及如何我可以解决这个问题吗？

此外，我可以执行 head(X) 来获取第一行，但是如何获取 X 中的最后一行？

我在 sparkR 中有一个 DataFrame X。X 包含 ID = 1 2 3 1 2 3 9 ... 的列以及每个条目的分数： score = 1241 233 20100 ....

因此，要找到 ID 的所有分数

然后我们得到 ID 1 的所有分数，我们可以将它们相加。

我想知道 X 中 ID=1 的数量，所以我使用 SparkR 中的“计数”函数

但这需要很长时间来计算。有一个更好的方法吗？

假设我们已经安排或排序了 X 所以 ID = 1 1 1 2 3 3 3 4 ..... 那么也许有更好的选择来避免做 count(s) 。

在 RI 中创建了两个数据集，我通过以下方式将它们保存为 csv 文件

我现在想在 SparkR 中打开这些 csv 文件。所以我输入

事实证明，在 SparkR 中成功加载了一个数据集“liste”，但是，由于某些奇怪的原因，无法加载“data”。

'liste' 只是 R 中的数字向量，而 'data' 是我在 R 中加载并删除了 data.frame 的某些部分的 data.frame。SparkR 给了我这个错误信息：

错误：returnStatus == 0 不是 TRUE

IR 和 sparkR 我想改变'max-ppsize'。

https://cran.r-project.org/doc/manuals/r-release/R-intro.pdf 第 93/105 页有对该命令的说明。

当我在 Ubuntu 上打开 R 或 sparkR 程序时，我该如何输入它？我在其他地方找不到明确的描述。

我已经查看了 如何在 R 中设置最大 ppsize 的答案？

我可以在 R 中成功地做到这一点，但在 SparkR 中我不能做到。

我像这样在ubuntu上启动sparkR

然后我输入

但我收到一条错误消息：“log4j Warn No appenders could be found for logger”

我对 SPARKR 比较陌生。我下载了 SPARK 1.4 并设置 RStudio 以使用 SPARKR 库。但是我想知道如何将函数应用于分布式 DataFrame 的列中的每个值，有人可以帮忙吗？例如，

这完美地工作

如何使其适用于分布式 DataFrame。目的是将“_hello”附加到DF的列名称的每个值

在 SPARK 1.4 发布之前的 alpha 版本的 SPARKR 中似乎已经有了这个能力，为什么现在在 SPARK 1.4 正式版本中缺少这个能力？

我得到错误：

这是我的 R 代码：

我怎么解决这个问题？