Wie verwendet man MapReduce, um Datenspeicher-Entities, die eine Abfrage erfüllen, gemeinsam zu aktualisieren?

Question

Ich möchte die mapreduce-Bibliothek verwenden, um alle Entitäten zu aktualisieren, die eine Abfrage erfüllen. Es gibt ein paar Komplikationen:

1. Die Abfrage, die die Entitäten findet Kontrollen zu aktualisieren, wenn der Wert einer bestimmte Eigenschaft „property1“ in einer langen Liste von Werten enthalten ist (~ 10000 Einträge) aus einem CSV
2. Für jede Entität Datei Erfüllung der Abfrage, eine andere Eigenschaft „property2“ muss aktualisiert werden, um gleich den Wert in der zweiten Spalte und derselben Zeile der cSV-Datei

ich weiß, wie die cSV-Upload Datei zu Blobstore und lesen Sie jede Zeile mit einem Blobstore-Eingabe-Reader. Mir ist auch der Datastore-Eingabe-Leser bekannt, der Entitäten mithilfe einer Abfrage abruft.

Meine Frage ist, wie kann ich eine Mapper-Klasse erstellen, die Eingabedaten aus dem Blobstore liest, holt die Datastore-Entitäten und aktualisiert sie so effizient wie möglich?

Answer 1

Da die Liste der möglichen Werte für property1 lang ist, um eine Abfrage-Filter scheint nicht wie eine gute Option (weil Sie einen IN-Filter verwenden müssten, was eigentlich runs one query per value)

Eine Alternative, bei der MR verwendet wird, besteht darin, die CSV-Datei unter Verwendung eines Map (von property1 bis property2) in den Speicher zu laden und anschließend einen MR-Job zu erstellen, der alle Entitäten iteriert und deren property1 Teil der Schlüssel ist Map, modifiziere es mit dem gemappten Wert.

Wie @Ryan B sagt, müssen Sie MR nicht verwenden, wenn Sie nur Batch-Puts nutzen möchten, da Sie mit dem DatastoreService einen Iterable bis put verwenden können.

Answer 2

Sie können einen DatastoreInputReader verwenden und in der Kartenfunktion herausfinden, ob die Eigenschaft1 tatsächlich in der CSV enthalten ist: Das Lesen von einer CSV wäre jedes Mal sehr langsam, was Sie tun können, ist memcache, um diese Information bereitzustellen Es wird nur einmal von seinem eigenen Datastore-Modell gelesen. Um das Datenspeichermodell zu füllen, würde ich empfehlen, property1 als benutzerdefinierte ID jeder Zeile zu verwenden. Auf diese Weise ist die Abfrage einfach. Sie würden den Datenspeicher nur für die Werte aktualisieren, die sich tatsächlich ändern, und den Mutationspool verwenden, um ihn performant zu machen (op.db.Put()). Ich lasse Sie Pseudo-Code (sorry ... Ich habe es nur in Python), wie die verschiedenen Stücke aussehen würde, ich Sie weiter empfehlen, diesen Artikel auf MapReduce auf Google App Engine zu lesen: http://sookocheff.com/posts/2014-04-15-app-engine-mapreduce-api-part-1-the-basics/

#to get the to_dict method 
from google.appengine.ext import ndb 
from mapreduce import operation as op 
from mapreduce.lib import pipeline 
from mapreduce import mapreduce_pipeline 

class TouchPipeline(pipeline.Pipeline): 
    """ 
    Pipeline to update the field of entities that have certain condition 
    """ 

    def run(self, *args, **kwargs): 
     """ run """ 
     mapper_params = { 
      "entity_kind": "yourDatastoreKind", 
     } 
     yield mapreduce_pipeline.MapperPipeline(
      "Update entities that have certain condition", 
      handler_spec="datastore_map", 
      input_reader_spec="mapreduce.input_readers.DatastoreInputReader", 
      params=mapper_params, 
      shards=64) 


class csvrow(ndb.Model): 
    #you dont store property 1 because you are going to use its value as key 
    substitutefield=ndb.StringProperty() 

def create_csv_datastore(): 
    # instead of running this, make a 10,000 row function with each csv value, 
    # or read it from the blobstore, iterate and update the values accordingly 
    for i in range(10000): 
    #here we are using our own key as id of this row and just storing the other column that 
    #eventually will be subtitute if it matches 
    csvrow.get_or_insert('property%s' % i, substitutefield = 'substitute%s').put() 


def queryfromcsv(property1): 
    csvrow=ndb.Key('csvrow', property1).get() 
    if csvrow: 
    return csvrow.substitutefield 
    else: 
    return property1 

def property1InCSV(property1): 
    data = memcache.get(property1) 
    if data is not None: 
     return data 
    else: 
     data = self.queryfromcsv(property1) 
     memcache.add(property1, data, 60) 
     return data 

def datastore_map(entity_type): 
    datastorepropertytocheck = entity_type.property1 
    newvalue = property1InCSV(datastorepropertytocheck) 
    if newvalue!=datastoreproperty: 
    entity_type.property11 = newvalue 
    #use the mutation pool 
    yield op.db.Put(entity)