python - AppEngine MapReduce 如何在使用数据存储输入读取器时过滤 StructuredProperty？

Question

使用appengine mapreduce lib时，如何通过StructuredProperty进行过滤？

我尝试过:

class Tag(ndb.Model):
    # ...
    tag = ndb.StringProperty()
    value = ndb.FloatProperty(indexed=False)

class User(ndb.Model):
    # ...
    tags = ndb.StructuredProperty(Tag, repeated=True)

class SamplePipeline(base_handler.PipelineBase):


    def run(self, tags, start_time, account_type, gsbucketname):

        start_time = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
        filters = []

        for tag in tags:
            filters.append(("tags.tag", "=", tag))

        yield mapreduce_pipeline.MapperPipeline(
            "name",
            "mapper",
            "mapreduce.input_readers.DatastoreInputReader",
            output_writer_spec="mapreduce.output_writers.FileOutputWriter",
            params={
                "input_reader": {
                    "entity_kind": "User",
                    "batch_size": 500,
                    "filters": filters
                },
                "output_writer": {
                    "filesystem": "gs",
                    "gs_bucket_name": gsbucketname,
                },
                "root_pipeline_id": self.root_pipeline_id,
                "account_type": account_type
            },
            shards=255
        )

我得到了什么。

File "/Users/lucemia/vagrant_home/adex2/lib/mapreduce/input_readers.py", line 794, in _validate_filters_ndb
    prop, model_class._get_kind())
BadReaderParamsError: ('Property %s is not defined for entity type %s', u'tags.tag', 'User')

Answer 1

看起来 mapreduce/input_readers.py 的 _validate_filters_ndb() 函数确实存在错误。

作为解决方法，您可以在 mapreduce/input_readers.py 的第 792 行添加以下行，以跳过对其中包含 . 的属性的验证，例如StructuredProperty 类型的属性执行以下操作:

if "." in prop:
    continue

我们将致力于修复以正确验证 StructuredProperty 类型的属性，并在未来版本中提供正确的代码。