在最终的reduce作业中,appengine Mapreduce API是否根据自己的逻辑决定计算碎片的大小?在
我正在使用appengine mapreduce API并提供了shard_size
kwarg设置我的mapreduce碎片大小。在
切分大小在我的mapreduce作业中特别重要,因为我不想在reduce函数的最后一步执行时将太多的结果批处理到任何一个结果中。换句话说,我在硬编码切分大小,以便根据系统的外部约束均匀地划分用户。在
地图工作看起来很好,但是reducer只使用了我指定的碎片的一小部分。在
下面是我正在处理的代码类型的粗略概述:
SHARD_SIZE = 42
def map_fun(entity):
shard_key = random.randint(1, SHARD_SIZE)
yield (
shard_key,
db.model_to_protobuf(entity).SerializeToString().encode('base64')
)
def reduce_fun(key, entities):
batch = []
for entity in entities:
#check for stuff
batch.append(entity)
expensive_side_effect(batch)
class MyGreatPipeline(base_handler.PipelineBase):
def run(self, *args, **kw):
yield mapreduce_pipeline.MapreducePipeline(
'label'
'path.to.map_fun',
'path.to.reduce_fun',
'mapreduce.input_readers.DatastoreInputReader',
'mapreduce.output_writers.BlobstoreOutputWriter',
mapper_params={
'entity_kind': 'path.to.entity',
'queue_name': 'coolQueue'
},
reducer_params={},
shard_size = SHARD_SIZE
)
map_fun
专门为每个实体分配一个根据碎片大小随机确定的碎片。我很困惑为什么我的reducer的碎片会比SHARD_SIZE
少,因为有很多实体,而且重复选择相同的整数是极不可能的。在
我不明白你在这里做什么。使用映射阶段将内容分组到一个小的、分片的密钥上,然后在reduce time处理这些密钥看起来很奇怪。即使你的reduce工人和mapper工人的数量一样多,你最终还是要为每个关键点做太多的工作。在
正在处理的'batch'是随机任意的,所以我假设
expensive_side_effect()
不依赖于批的内容。为什么不在映射时做这个工作呢,释放一些reduced可以传递给输出编写器的东西呢?在相关问题 更多 >
编程相关推荐