Erhalten zufälligen Satz von Zeilen aus PCollection

Question

Ich habe einen Datensatz der Form

user_id, date, other_columns 
1, 2017-03-10, ... 
2, 2017-03-10, ... 
3, 2017-03-10, ... 
... 

und ich brauche folgendes zu tun: Für jede Zeile im Datensatz ich eine neue Zeile erstellt werden soll, die die enthalten aktuelle Zeile und eine zufällige Teilmenge von N Zeilen für den gleichen Tag an verschiedenen Benutzer folgend entsprechen:

row, other_rows 
{'user_id': 1, 'date': '2017-03-10', ...}, [{'user_id': 2,...},...] 
{'user_id': 2, 'date': '2017-03-10', ...}, [{'user_id': 1,...},...] 
... 

ich habe es wie folgt implementiert, aber es ist sehr langsam für große Datensätze, wenn in der Cloud ausgeführt.

dataset 
| 'map-to-date' >> beam.Map(lambda x: (x['date'], x)) 
| 'group-by-date' >> beam.GroupByKey() 
| 'generate-output' >> beam.ParDo(GenerateOutputRows()) 

wo GenerateOutputRows wie folgt definiert ist:

class GenerateOutputRows(beam.DoFn): 
    def process(self, element): 
     (date, rows) = element 
     for r in rows: 
      other_users_rows = list(filter(lambda x: x['user_id'] != r['user_id'], 
              rows)) 
      yield (r, random.sample(other_users_rows, N)) 

könnte denken Sie an eine andere leistungsfähigere Art und Weise, um das gewünschte Ergebnis zu bekommen?

Answer 1

ich nicht sofort sehen algorithmisch effizientere Art und Weise, dies zu tun, es sei denn, Sie einige vereinfachende Annahmen zur Verfügung stellen kann, zum Beispiel:

• Innerhalb 1 Datum ist user_id einzigartig oder können mehrere Zeilen mit demselben dort sein Benutzer?
• Wenn ja, müssen dann mehrere Samples von other_users_rows für denselben Benutzer statistisch unabhängig sein? Wenn dies nicht der Fall ist, können Sie Caching verwenden und dasselbe Beispiel mehrfach für Zeilen mit derselben Benutzer-ID verwenden.
• Müssen die other_users_rows-Beispiele für verschiedene Benutzer statistisch unabhängig sein, oder wenn beispielsweise ein Beispiel für Benutzer A Benutzer B nicht enthält, dann ist es in Ordnung, genau dasselbe Beispiel für Benutzer B zu verwenden?

In der Regel ist dies eine algorithmische Frage eher als eine Wolke Dataflow/Apache Strahl Frage, weil der Engpass des Codes der O (rows.size()^2) ist Schleife in GenerateOutputRows und Beam kann nicht machen das schneller automatisch; Ich würde empfehlen, zu http://cstheory.stackexchange.com Empfehlungen für einen effizienteren Algorithmus zu gehen.