Pipeline- Funktionen?

[ravenheart]

Hallo, ich bearbeite ja im Moment csv-Dateien.
Viele Funktionen arbeiten nur zeilenweise, weshalb ich mir gedacht habe, dass es schön wäre, wenn eine bearbeitete Zeile sofort der nächsten Funktion zur Vefügung stünde (ähnlich wie bei pipes).

Nur frage ich mich, wie man das realisieren könnte.
Ein kleines absichtlich sehr einfach gehaltenes Beispiel

Code: Alles auswählen

csv.select_column("A,B").select_column("A")

Dieser Code soll zB aus

Code: Alles auswählen

A  B  C                    A    B                             A
1  2  3        zunächst    1    2         anschließend        1
4  5  6                    4    5                             4
7  8  9                    7    8                             7

machen.

Wenn ich jetzt Funktionen schreibe, werden die sequentiell ausgeführt, wobei eine immer auf das Ende der vorhergehenden wartet.

Kann man das hübscher bewerkstelligen?

Ich habe gesehen, dass es ei pipes-Modul gibt, allerdings wüsste ich nicht wie ich es nutzen könnte

[sma]

Mit Generatoren (Stichwort `yield`) kannst du so etwas erreichen.

Hier ein dummes Beispiel mit einem Generator, der einen endlosen Strom von Einsen produziert, den man dann konsumieren kann (in Python 2.x gibt es kein next() sondern eine entsprechende Methode):

Code: Alles auswählen

def produce():
    while True:
        yield 1

def consume():
    s = 0
    p = produce()
    for i in range(10):
        s += next(p)
    return s

print(consume())

Stefan

[EyDu]

Was spricht denn gegen einfache filter-Funktionen? Wenn man die mit "functools.partial" noch ein wenig anpasst, dann reicht das für die meisten Fälle aus.

in Python 2.x gibt es kein next() sondern eine entsprechende Methode

Python 2.6.5 (r265:79063, Apr 16 2010, 13:57:41)
[GCC 4.4.3] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> next
<built-in function next>

[syntor]

Ich hoffe mal ich habe dich richtig verstanden, und beziehe mich auf dein Code-Beispiel^^

Ich denke eher, du hast da ein falsches Bild im Kopf, wenn du sagst "eine Zeile der nächsten Funktion zur Verfügung zu stellen".


Zuerst einmal hast du dein csv Objekt, das drei Spalten mit Daten beinhaltet - oder Zeilen à 3 Spalten.
Die Methode select_column() gibt dir ein neuest Objekt zurück, welches nur noch die zuvor angeforderten Spalten beinhaltet.
Nun repräsentiert "csv.select_column("A, B")" dein neues csv Objekt. Auf dieses kannst du nun wiederum die Methode select_column anwenden.

[ravenheart]

So ist es in etwa gerade.
Ich habe eine Klasse, welche die methode select_column hat.

Nun ist es so dass ich in der Klasse als Variablen einmal die Spaltennamen als auch deren Position speichere

im Moment sieht das so aus:

Code: Alles auswählen

select_columns(liste der zu wählenden Spalten):
    indices = []
    for name in liste der spalten:
        indices.append( dict[name] )
    for line in csv:
        line = line.split()
        for i in indices:
            new_line = line[i] + "\t"
            schreibe new_line in temporäre datei
     lege neues csv-objekt an
     gebe dies zurück

Mir wäre es jedoch lieber, wenn diese Funktion bei wiederholtem Anwenden so reagiert, dass einzelne Zeilen zurückgegeben werden können.
Sprich: select_columns 1 berbeitet die 2te Zeile während der 2te Aufruf sich um die erste schon bearbeitete kümmern kann

[BlackJack]

@ravenheart: Wenn es gerade so ist, verstehe ich das Problem nicht. Allerdings scheint es mir auch nicht so zu sein. Was ist denn ein neues csv-Objekt in Deinem Sprachgebrauch was anscheinend für jedes Element in jeder Spalte "zurückgegeben" wird? Wobei da ja kein ``return`` sein kann, denn an der Stelle wo dieser Satz steht, würde dann ja nur das erste Element der ersten Zeile verarbeitet werden!?

Fang doch mal am Anfang an. `select_columns()` ist ja schon der zweite Schritt. Der erste wäre einen Generator für die einzelnen Datensätze so einer Datei zu schreiben. Wenn es so etwas wie Spaltennamen gibt, dann vielleicht als Iterator über `Record`-Objekte die jeweils einen Datensätz + Metadaten repräsentieren, also Zuordnung von Spaltennamen zu Index. Wenn es verschiedene Spaltentypen gibt, könnte man das an der Stelle auch schon erledigen.

[ravenheart]

sry falsch eingerückt...

Es wird ein csv-objekt mit allen Zeilen zurückgegeben. Korrigiere schnell den Code

hab gerade den echten quellcode nicht da, deswegen kann es zu fehlern kommen

[syntor]

Mir wäre es jedoch lieber, wenn diese Funktion bei wiederholtem Anwenden so reagiert, dass einzelne Zeilen zurückgegeben werden können.
Sprich: select_columns 1 berbeitet die 2te Zeile während der 2te Aufruf sich um die erste schon bearbeitete kümmern kann

Es tut mir Leid, aber ich kann nicht wirklich nachvollziehen, was du genau erreichen möchtest. Das was du jetzt beschreibst, sind zwei Aktionen die gleichzeitig ablaufen sollen? Das kannst du nur mit Threads erreichen - allerdings erachte ich das in diesem Fall weder als angebracht, noch als nötig.

Falls du nicht das meinst - kannst du vielleicht ein Beispiel machen so wie du es im ersten Post gemacht hast? Ohne Code?

[ravenheart]

Ich versuche es zu malen, ohne große Spalten:

Ausgangsdatei

Code: Alles auswählen

Header
Zeile 1
Zeile 2
Zeile 3
Zeile 3

Gewünschter Ablauf:

Code: Alles auswählen

Datei        erster Aufruf                     zweiter Aufruf
Header       modifiziert Header             modifiziert2 header
Zeile 1      modifizierte Zeile      --->   modifiziert2 zeile
Zeile 2      wird gerade bearbeitet       wartet darauf dass neue Zeilen kommen
Zeile 3
Zeile 3

Im Grunde das selbe Verhalten wünsche ich, als wenn ich

Code: Alles auswählen

cat dateien | grep irgendwas | grep blubb ...

ausführe.
grep irgendwas wartet ja auch nicht bis cat alle dateien ausgegeben hat

[syntor]

Ah ok, jetzt sehe ich was du meinst.

Um das mit der von dir gewünschten Syntax zu lösen, müsstest du wohl auf Threads zurückgreifen.

Einen ähnlichen Effekt erhältst du hiermit:

Code: Alles auswählen

def pipe(*fns):
    def piped(*args, **kwargs):
        res = fns[0](*args, **kwargs)
        for fn in fns[1:]:
            res = fn(res)
       return res
    return piped

a_b_c = pipe(a, b, c)
a_b_c(argumente)

Das wäre ähnlich zu "a | b | c"

[EyDu]

Wenn die Daten nicht wirklich parallel verarbeitet werden sollen, was bei Threads mit Python eh der Fall ist (mal von Fällen wie IO und parallel dazu Berechnungen), dann bieten sich Generatoren an. Das hatte sma schon ganz am Anfang beschrieben, darauf gab es jedoch keine Reaktion.

[ravenheart]

Generatoren klingen in meinen Ohren bisher sehr gut ^^

Auf die Frage ob die nicht in Frage kämen, muss ich gestehen, dass diese mir total unbekannt waren.
Leider sehe ich noch keinen Weg die zu implementieren.

Als ersten Versuch, wollte ich eine Klasse schreiben, die meinem csv-aufbau recht ähnlich ist und eine simple Funktion bietet:

Code: Alles auswählen

"""
file soll hier mal meine csv datei ersetzen
in meiner echten Klasse (Quellcode nicht vorliegend) entspricht file dem pfad zu einer csv-datei
"""
file = [ "AAAAA","BBBBB","CCCCC","DDDDD","EEEEE","FFFFF"]

class Simple:
    def __init__(self,file):
		self.file = file
# statt spalten auszuwählen soll jeder string einfach eins kürzer gemacht werden
    def select_one_less():
	for line in self.file:
	    yield line[:-1]

Nun frage ich mich wie ich es bewerkstelligen könnte, diesen Aufruf hintereinandetr auszuführen, also

Code: Alles auswählen

 Simple(file).select_one_less().select_one_less()

Ich bin gerade noch in der "Denkphase", wenn mir jmd aber eine Lösung geben kann und möchte, ist diese gern willkommen.

[syntor]

Zwar nicht über Generatoren, aber funktioniert mit der Syntax die du möchtest.

Code: Alles auswählen

"""
file soll hier mal meine csv datei ersetzen
in meiner echten Klasse (Quellcode nicht vorliegend) entspricht file dem pfad zu einer csv-datei
"""
file = [ "AAAAA","BBBBB","CCCCC","DDDDD","EEEEE","FFFFF"]

class Simple:
    def __init__(self,file):
		self.file = file
# statt spalten auszuwählen soll jeder string einfach eins kürzer gemacht werden
    def select_one_less():
	shorten = lambda line: line[:-1]
        return Simple(map(shorten, self.file))

[ravenheart]

Wenn ich das aber so betrachte, zweifle ich wieder an der Parallelität, die ich wünsche ...
Ist anscheinend nicht trivial ^^

Ich kann halt keinen Generator verwenden und im selben Zug yield nutzen... verflixt nochmal, da muss es doch ein Schlupfloch geben ^^

[syntor]

Parellelität hast du aber auch mit Generatoren nicht.
Weshalb möchtest du denn, dass parellel laufen haben?

[ravenheart]

Weil es im Moment so aussieht:

Es gibt ein Skript select_columns, das die Standardeingabe ausliest und auch dorthin wieder schreibt.
Dadurch kann es Pipes unterstützen und läuft annäherungsweise parallel.

Code: Alles auswählen

 cat csv | select_columns.py -c A,B,C

Ich möchte diese Skriptsammlung jetzt aber durch eine API ersetzen (ich hoffe der Begriff ist hier nicht unangebracht).
Das ganze werden die Leute aber nur benutzen wenn es
- den gleichen Funktionsumfang hat
- nicht wesentlich langsamer läuft
- leichter zu bedienen ist
- nicht wesentlich mehr Speicher / Festplatte benötigt

Was ich im Moment habe ist meiner Meinung nach leichter oder auf jeden Fall genau so gut bedienbar wie die Skriptsammlung
Der Funktionsumfang ist annähernd der selbe.

Ich benötige aber im Moment eine Menge Festplattenspeicher und Zugriffe, da ich grob skizziert folgendes mache:

Code: Alles auswählen

class csv:
    def __init__(self, path):
        self.file = path
    def select_columns(self, cols)
        file = open(path)
        tmp = tempfile.Tempfile
        for line in file:
            bearbeite 
            schreibe in tmp
        newcsv = csv(tmp)
        return newcsv

Das ist ganz grob skizziert, natürlich würde das oben nicht ganz funktionieren, da die tempfiles anders handzuhaben sind, aber das ist gelöst und klappt ganz gut.
Problem dabei:
ich schreibe für jede Operation, die ich durchführe eine neue Datei auf der Festplatte . Das bedeutet dass ich viel Festplattenspeicher brauche (aber ich kann davon ausgehen, dass mehr als genug zur Verfügung steht) und andererseits werde ich durch die Festplattenzugriffe auch langsamer.

So wie ich das bisher habe klappt alles ganz gut, aber ich würde das ganze gerne optimieren, um schneller zu sein und weniger Speicher zu belegen

[ms4py]

Vielleicht ist ja `multiprocessing.Pipe` genau das, was du suchst.

[BlackJack]

@ravenheart: Ich denke wir sollten nochmal die Begrifflichkeiten klären: Willst Du Parallelität, also dass die Schritte wirklich *gleichzeitig* ausgeführt werden, oder sollen die Daten nur "lazy" verarbeitet werden, also mehrere Schritte hintereinander ausgeführt, die jeweils nur die unmittelbar benötigten Daten erzeugen und weiterreichen statt alles zu verarbeiten und das dann an den nächsten Schritt weitergeben?

Ich denke letzteres, denn Parallelität macht IMHO hier nicht soviel Sinn, da der Folgeschritt immer vom (Teil)Ergebnis des Vorgängers abhängt und das ganze damit doch wieder in eine größtenteils sequenzielle Abarbeitung zwingt.

Die einfache Antwort bleibt dafür IMHO weiterhin Generatoren. Das funktioniert so direkt nicht mit Deiner Schreibweise. Statt ``table.select(a).select(b)`` wäre es dann ``select(select(table, a), b)``. Wenn Du unbedingt auf die verketteten Aufrufe bestehst, wirst Du dafür eine Klasse schreiben müssen die "iterable" ist und von den Methoden zurückgegeben wird und selber diese Methoden besitzt. In der `__iter__()`-Methode kann dann ja wieder ``yield`` zum Einsatz kommen.

Was mir persönlich an so *einer* Klasse nicht gefallen würde, ist, dass sie alle möglichen Verarbeitungsschritte kennen muss, damit man wirklich alle möglichen `select()`\s, `filter()`, oder `aggregate()`\s darauf aufrufen kann.

Also wenn's wirklich "gross" und "OOP" werden soll, würde ich die einzelnen Arbeitsschritte wohl in eigenen Klassen implementieren, die jeweils über ihre Datensätze "iterable" sind und die Metadaten wie Spaltennamen und Typen kennen, und eine "magische" Klasse der man diese Klassen bekannt machen kann/muss, und die dann das verketten der Aufrufe ermöglicht. Und die einzelnen Datensatz-Objekte kennen dann natürlich auch die Metadaten der Zeile, die sie repräsentieren, und erlauben Indexzugriff, Schlüsselzugriff, und Attributzugriff.

[ravenheart]

Deleted

[ravenheart]

Hmm, kannst du eventuell noch schnell eine Art Klassendiagramm hinschreiben, damit ich mir das besser vorstellen kann?