问题标签 [input-split]

我是否能够将全部input split输入映射器而不是将每一行输入拆分为映射器。

为此，我需要实现自己的自定义输入格式。但如果我在写WholeFileInputFormat

这是否意味着映射器获得整行或整个输入拆分？

NLineInputFormat能解决我的问题吗？

Consider this scenario:

I have 4 files each 6 MB each. HDFS block size is 64 MB.

1 block will hold all these files. It has some extra space. If new files are added, it will accommodate here

Now when the input splits are calculated for Map-reduce job by Input format, (split size are usually HDFS block size so that each split can be loaded into memory for processing, there by reducing seek time.)

how many input splits are made here:

1. is it one because all the 4 files are contained with in a block?

2. or is it one input split per file?

3. how is this determined? what if I want all files to be processed as a single input split?

这个问题不是重复的： Hadoop 进程记录如何跨块边界拆分？

我有一个关于输入拆分计算的问题。根据 hadoop 指南

1) InputSplits 尊重记录边界

2）同时它说拆分是由Job Submitter计算的。我假设它在客户端运行。[MapReduce 作业运行剖析 - 经典 MRv1]

这是否意味着：

(a) 作业提交者读取块来计算输入拆分？如果是这种情况，那么它不会是非常低效的，并且不会达到 hadoop 的目的。

或者

(b) 作业提交者是否仅根据块大小和位置计算拆分，这仅仅是基于块大小和位置的估计，然后它是否成为 InputFormat 和 RecordReader 在映射器下运行以获取跨越主机边界的记录的责任。

谢谢

我查看了 NLineInputFormat 的 getSplitsForFile() fn。我发现为输入文件创建了一个 InputStream，然后每 n 行创建它的迭代和拆分。它有效率吗？特别是在启动映射器任务之前在 1 个节点上发生此读取操作时。如果 1 有 5gb 的文件怎么办。基本上，这意味着文件数据被搜索两次，一次是在拆分创建期间，一次是在从映射器任务读取期间。如果这是一个瓶颈，hadoop 作业如何覆盖它？

编辑以将我的用例提供给 clément-mathieu

我的数据集是大输入文件，每个大约 2gb。文件中的每一行代表一个需要插入到数据库表中的记录（在我的情况下是 cassandra）我想将我的数据库的批量事务限制为每 n 行。我已经使用 nlineinputformat 成功地做到了这一点。我唯一担心的是生产中是否存在隐藏的性能瓶颈。

1. 鉴于我有一个要使用 Hadoop 处理的文件，并且我知道文件的大小小于 HDFS 的块大小。这是否保证文件不会被拆分，并且我不需要为它写一个 InputSplit，因为默认的不会拆分它？

2. 假设使用 SequenceFileOutputFormat（或其他输出格式）保存的文件大于块大小，但仅包含一个键值对。这是否意味着文件块将存储在同一节点上（复制副本除外）并且 MapReduce 任务不会浪费太多时间来获取它们？这是否意味着我不需要编写自己的 inputSplit 因为密钥不会被拆分（密钥大小小于块大小并且只有一个密钥）？

我正在使用 Jackson 来处理 Hadoop 中以块形式出现的 JSON。这意味着，它们是被分割成块的大文件（在我的问题中是 128M，但这并不重要）。出于效率原因，我需要它是流式传输的（不可能在内存中构建整个树）。

我正在使用 JsonParser 和 ObjectMapper 的混合物来读取我的输入。目前，我使用的是不可拆分的自定义 InputFormat，因此我可以读取整个 JSON。

（有效）JSON 的结构类似于：

我想在 RecordReader 中读取的记录是“记录”元素中的元素。“...”表示那里有更多信息，符合我的记录。如果我只有一个分裂，那根本没有问题。我使用 JsonParser 进行细粒度（标题并移动到“记录”标记），然后我使用 ObjectMapper 和 JsonParser 将记录作为对象读取。详情：

现在，假设我有一个包含两个输入分割的文件（即“记录”中有很多元素）。有效的 JSON 从第一次拆分开始，我读取并保留标题（每条记录都需要它，在本例中为“日期”字段）。

拆分将剪切 Records 数组中的任何位置。所以让我们假设我得到第二次这样的分裂：

我可以在开始解析之前进行检查，以将 InputStream (FSDataInputStream) 移动到记录的开头 ("{" )，其中包含下一个 "TSSNAME"（这可以完成）。在开始时丢弃尾随的“垃圾”很好。所以我们得到了这个：

然后我将它处理到上面看到的 JsonParser/ObjectMapper 对。第一个对象“ZZZ”读取正常。但是对于下一个“ZZZ2”，它打破了：JSONParser 抱怨 JSON 格式错误。它遇到一个“，”不在数组中。所以它失败了。然后我无法继续阅读我的记录。

如何解决这个问题，所以我仍然可以从第二个（和第 n 个）拆分中读取我的记录？如何让解析器忽略逗号上的这些错误，或者让解析器提前知道它正在读取数组的内容？

我正在使用flume将twitter数据下载到hdfs中。虽然我有超过 2 GB 的数据，但我的块文件拆分小于 64 MB。即第一个文件 300KB，第二个文件 - 566 KB。为什么会这样？

我在一个 csv 文件中有一个数据集，它占据 HDFS 中的两个块，并在两个节点 A 和 B 上复制。每个节点都有一个数据集的副本。

当 Spark 开始处理数据时，我看到了 Spark 如何将数据集作为输入加载的两种方式。它要么将整个数据集加载到一个节点上的内存中并在其上执行大部分任务，要么将数据集加载到两个节点中并将任务溢出到两个节点上（基于我在历史服务器上观察到的情况）。对于这两种情况，都有足够的容量将整个数据集保存在内存中。

我多次重复相同的实验，Spark 似乎在这两种方式之间交替。假设 Spark 继承了 MapReduce 作业中的输入拆分位置。据我了解，MapReduce 应该能够利用两个节点。我不明白为什么 Spark 或 MapReduce 会在这两种情况之间交替出现。

当只使用一个节点进行处理时，性能更差。

我正在从 Hadoop 读取一个 138MB 的文件，并尝试为每条记录分配序列号。以下是我遵循的方法。

我使用级联读取整个文件，为每条记录分配当前切片编号和当前记录计数器。预计这将对每个块并行运行，并根据存在的块分配唯一的序列号和切片号，即文件的块 0 应该转到映射器编号 0，切片编号将为“0”，而对于块 1，映射器没有 1将切片编号分配为“1”（级联中的切片与 MapReduce 中的输入拆分相同）。还预计切片编号为“0”的记录应该比切片编号为“1”的记录多得多，因为块 0 将是 128 MB，块 1 将是 10 MB。

但是当我看到输出时，我看到两组输入记录的记录数几乎相同，即记录均匀分布在 2 个映射器中。

我还可以看到文件的第一条记录是由 mapper1 而不是 mapper0 读取的。

您能否帮助我理解为什么记录在映射器之间均匀分布？

来自 Hadoop MapReduce InputFormat接口上的 Apache 文档：

" [L] 基于输入大小的逻辑拆分对于许多应用程序来说是不够的，因为要尊重记录边界。在这种情况下，应用程序还必须实现一个RecordReader，它负责尊重记录边界并呈现记录面向单个任务的逻辑InputSplit的视图。”

WordCount示例应用程序是否基于输入大小的逻辑拆分不足？如果是这样，在源代码中的哪里找到 RecordReader 的实现？