问题标签 [mrjob]

我有一个 status_changes 的日志文件，每个文件都有一个 driver_id、时间戳和持续时间。使用 driver_id 和时间戳，我想从 S3 获取适当的 GPS 日志。这些 GPS 日志以 bucket_name/yyyy/mm/dd/driver_id.log 形式存储在 S3 存储桶中。

在命令行中，我使用 status_change 日志文件作为输入运行它： $ python mileage.py status_changes.log

我的问题是：给定我构建的 S3 URI 字符串，我如何实际获取该 GPS 日志？

问题陈述：

我正在尝试使用 python MRJob 库在 Amazon EMR 中运行 map-reduce 作业，但在使用必要的库和包引导节点时遇到了问题。

细节：

我的示例python mrjob代码：

有一些像 NLTK 这样的库需要导入，还有一些我正在导入的本地模块，比如from sentClassifier import sentClassify

我想知道引导 EMR 节点以使这些方法和包可用的最佳方法是什么。该代码在我的本地机器上运行良好。

我的示例mrjob.conf文件：

但是工作失败了。

我已阅读以下参考资料并尝试了所有各种方法，但仍然没有运气：

• MRJob 文档 EMR Runner 配置
• MRJob 文档 Hadoop Runner 配置
• Salmon Run 的博文

错误日志：

任何帮助将非常感激

我有一个相对较大的文件 - 大约 10GB 需要处理。我怀疑它不适合我笔记本电脑的 RAM，如果 MRJob 决定在 RAM 或类似的东西中对其进行排序。

同时，我不想设置 hadoop 或 EMR - 工作并不紧急，我可以在睡觉前简单地启动 worker，并在第二天早上得到结果。换句话说，我对本地模式很满意。我知道，性能不会完美，但现在还可以。

那么它可以在一台弱机器上处理这样的“大”文件吗？如果是 - 您建议做什么（除了设置自定义 tmp 目录以指向文件系统，而不是指向会很快耗尽的 ramdisk）。假设我们使用 0.4.1 版本。

我有一个 mrjob 配置，其中包括将一个大文件从 s3 加载到 HDFS 中。我想在配置文件中包含这些命令，但似乎所有引导命令都在集群中的所有节点上执行。这是过度杀戮，也可能会产生同步问题。

是否有某种方法仅在 mrjob 配置中包含主节点的启动命令，或者是在集群启动执行这些操作后通过 SSH 进入头节点的唯一解决方案？

约夫

我的 MapReduce 程序的输入是一组二进制文件。我希望能够通过 mrjob 阅读它们。经过一番研究，似乎我必须编写一个自定义的 hadoop 流媒体 jar。有没有更简单的方法？还是这样的罐子很容易买到？更多详情如下。

输入文件只是一个 8 字节整数序列。我希望一次用 2 个整数调用我的映射器函数。

我首先想到我可以转换成pickle二进制格式，然后指定：

INPUT_PROTOCOL = mrjob.protocol.PickleProtocol。

但这给出了一个错误：无法解码输入。我也觉得 mrjob 只能使用 pickle ascii 格式（而不是二进制）。因为否则hadoop流将如何处理看起来像换行符的字节。mrjob 源代码似乎证实了这一点。

另一种选择是编写一个自定义的 hadoop 流 jar。mrjob 可以选择指定这样的 jar。但作为不熟悉 hadoop/Java 的人，我更喜欢基于 python 的解决方案。

我有一个 mrjob 程序，只是从 sql 数据库中获取数据，所以我不需要读取本地文件或任何输入文件，但是 mrjob 强迫我“从 STDIN 读取”，所以我只是创建一个空文件作为输入文件。真的很难看，有没有办法在没有输入文件的情况下运行作业？

我正在阅读本指南，了解如何让 mrjob 在 EMR 上工作。我遵循所有步骤，但是当我运行示例脚本时，我收到了这个错误：

我将为kmeans算法编写一个映射器和reducer，我认为最好的做法是将距离计算器放入mapper并发送到reducer，其中cluster id作为键，行坐标作为值。在 reducer 中，将执行更新质心。我是用python写的。

如您所知，我必须使用 Hadoop 流在STDIN和之间传输数据STOUT。据我所知，当我们print (key + "\t"+value)，它会被送到减速机。Reducer 将接收数据并计算新的质心，但是当我们打印新的质心时，我认为它不会将它们发送到映射器以计算新的集群，它只是将其发送到STDOUT并且如您所知，kmeans 是一种迭代算法。所以，我的问题是 Hadoop 流是否会遭受迭代程序的影响，我们应该MRJOB为迭代程序使用吗？

我正在尝试 MapReduce 日志，并且我想在 EMR 中处理它们之前按文件名过滤存储桶中的所有日志。另外，有些文件是 tar 目录，我希望 mrjob 解压缩它，然后过滤其中的文件以仅解析相关文件。

知道如何通过来自MrJob的文件名过滤存储桶 s3吗？我找到了mapper_pre_filter！方法，但它只逐行过滤输入。

我为 Hadoop 编写了一个简单的 k-means 聚类代码（两个独立的程序 - 映射器和化简器）。该代码正在处理我本地盒子上的一个二维点的小数据集。它是用 Python 编写的，我打算使用 Streaming API。

每次运行 mapper 和 reducer 后，都会生成新的中心。这些中心是下一次迭代的输入。

根据建议，我使用了适合多步骤的mrjob，job python，

这只是一次迭代，请告诉我在生成新中心后如何反馈给映射器。我的意思是，正如您在最后一步（“reducer”）中看到的那样，将生成新的中心，现在是时候再次将其反馈给映射器（第一步）以计算新中心的新距离，依此类推，直到满意的收敛。

（请不要告诉我 Mahout、spark 或任何其他实现，我知道它们。）