问题标签 [partitioner]

Is it must that we have to set number of reducers to use custom partitioner ? Example : Word Count problem, want to get all the stop words count in one partition and remaining words count to go to different partition. If I set number of reducers to two and stop words to go to one partition and others to go to the next partition, it will work, but I am restricting the number of reducers to two(or N ), which I don't want. What is the best approach here? Or I have to calculate and set the number of reducers based on the size of the input to get the best performance?

我正在为支持 .NET 3.5 的 Unity GameEngine 构建一个自定义插件，为了能够使用 TPL，我安装了 Reactive Extensions v1.0.2856.0（我认为是最后一个包含 TPL 的插件）。所以，在一个脚本中，我有一个大约 65000 次交互的循环。例如，我使用 Parallel.Foreach 循环没有问题，但问题是使用默认分区器，由于迭代次数少，工作分配不均，实际上循环变得有点慢。

所以我的问题是：有没有一种方法可以创建一个分区器，它可以为我的 2 个 CPU 内核提供大约一半的迭代工作。

我正在尝试了解如何实施Partitioner<T>.

我的情况：

我们从文件中读取并插入到 Azure 表存储中。我们使用任务来加快进程。该文件有近 10,000,000 行。

我尝试使用并行执行和分区来实现更优雅的解决方案。我想将这个 9,000,000 的大组划分为一组分区集，并将它们分布在 30-50 个线程中。

我认为我需要动态实现Partitioner<List<List<TableEntity>>>. 我有几个例子，一个带有 StringPartitioner 的例子，但它对我的情况并没有真正的帮助，因为我需要一个 IEnumerator 用于List<List<TableEntity>>.

如果这一切令人困惑，请告诉我，以便我澄清。

问题是如何在List<List<TableEntity>>.

我想问一下 Hadoop 分区器，它是在 Mappers 中实现的吗？如何衡量使用默认哈希分区器的性能 - 是否有更好的分区器来减少数据倾斜？

谢谢

我必须做一批：
从数据库中读取一些数据（每一行都是一个项目，这很好）
然后做一些过程来添加更多数据（更多数据总是更好;））
然后这是我的问题，我有将每个项目写入 xml 文件，其名称取决于项目的数据。

例如我有
ItemA (attr1=toto, attr2=foo, attr3=myNonKeyData...)=>进入 toto_foo.xml
ItemB (attr1=toto, attr2=foo, attr3=myNonKeyData...)=>进入 toto_foo .xml

ItemC (attr1=tata, attr2=foo...)=>进入 tata_foo.xml
...

我看不到如何只运行一次批处理。
我有太多的键和可能的输出文件来做分类器。
也许使用分区器可能是一个好主意，即使它似乎不是为此而设计的。

为了平滑所有 Kafka 分区之间的流量，我尝试在我的生产者上制作一个自定义分区器（扩展 kafka.producer.Partitioner），以替换每 10 分钟仅更改一次分区的默认分区器。

我的分区器使用简单的循环算法每 200 条消息从一个分区切换到另一个分区。由于生产者配置为 batch.num.messages=200 和 producer.type=async，因此每 200 条消息，即每个生产者的请求，流量都应该从一个分区跳到另一个分区。据我了解，考虑到请求的数量，这不应该修改代理的输入流量。

但是，经过一些测试，使用此自定义分区器时，请求量似乎乘以 2。

为什么我们会有这种增长？有没有办法解决这个问题，或者至少有办法限制流量增加？

谢谢你的帮助。

注意：我使用的是 Kafka 0.8.2-beta，生产者在 scala 2.10 中编码

我使用 Hadoop 全序分区器和随机采样器作为输入采样器。

但是当我增加我的从节点并将任务减少到 8 个时，我收到以下错误：

我不知道这个错误的原因。

如何设置inputsampler.randomsampler函数的三个参数的数量？

基于这里的这个例子，这是可行的。在我的数据集上尝试过同样的方法。

样本数据集：

将每一行视为字符串，我的 Mapper 输出为：

键-> 字符串[2]，值-> 字符串。

我的分区代码：

在我的数据集中，大多数 id 是 137176。Reducer 声明为 -2。我希望有两个输出文件，一个用于 137176，第二个用于剩余的 ID。我得到两个输出文件，但是 Id 均匀分布在两个输出文件上。我的程序出了什么问题？

在学习 Hadoop MapReduce 时，我遇到了如何创建自定义 Partitioner 类。我知道我们需要在我们的类中定义抽象的 getPartition 方法。此方法应该返回当前键值对的分区号（整数）。

现在，分区的数量将等于作业的 reduce 任务的数量。如果在自定义分区程序中，写一些逻辑来根据“值”而不是“键”选择分区怎么办？根据我的理解，这可能意味着具有相同键（但值不同）的记录可能会被不同的 reduce 任务处理，这不是 MapReduce 所保证的。这不是异常吗？为什么我们甚至需要 getPartition(key, value, numPartitions) 方法中的 'value' 参数？如果不正确，请纠正我的理解。

使用 Hadoop 流时，可以像这样设置和配置分区器和排序器：

我想在我的 Javamain()方法中做同样的事情。排序可以这样实现：

.setKeyFieldPartitionerOptions然而，类中的方法KeyFieldBasedPartitioner不是static：

在作业对象中，我只能设置一个类，但是：

在这种情况下如何设置上述选项？当然，我可以实现自己的分区器类，但是如果应该有一个简单的方法，为什么还要努力呢？