Schnelle Übersetzung von über einen Socket übertragenen String-Daten in Objekte in Python

Question

Ich habe derzeit eine Python-Anwendung, wo Newline-terminierte ASCII-Zeichenfolgen über einen TCP/IP-Socket an mich übertragen werden. Ich habe eine hohe Datenrate dieser Zeichenfolgen, und ich muss sie so schnell wie möglich analysieren. Derzeit werden die Zeichenketten als CSV übertragen, und wenn die Datenrate hoch genug ist, beginnt meine Python-Anwendung hinter der Eingabedatenrate zurückzubleiben (was wahrscheinlich nicht sonderlich überraschend ist).

Die Strings sehen in etwa so aus:

chan,2007-07-13T23:24:40.143,0,0188878425-079,0,0,True,S-4001,UNSIGNED_INT,name1,module1,...

Ich habe ein entsprechendes Objekt, das diese Zeichenketten analysiert und alle Daten in einem Objekt speichert. Derzeit sieht das Objekt etwa so aus:

class ChanVal(object):
    def __init__(self, csvString=None,**kwargs):

        if csvString is not None:
            self.parseFromCsv(csvString)

        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):

        lst = csvString.split(',')

        self.eventTime=lst[1]
        self.eventTimeExact=long(lst[2])
        self.other_clock=lst[3]
        ...

Um die Daten aus dem Socket zu lesen, verwende ich ein einfaches "socket.socket(socket.AF_INET,socket.SOCK_STREAM)" (meine App ist der Server-Socket) und dann verwende ich das "select.poll()"-Objekt aus dem "select"-Modul, um den Socket mit seiner "poll(...)"-Methode ständig nach neuen Eingaben zu fragen.

Ich habe eine gewisse Kontrolle über den Prozess, der die Daten sendet (d.h. ich kann den Absender dazu bringen, das Format zu ändern), aber es wäre wirklich praktisch, wenn wir die ASCII-Verarbeitung so weit beschleunigen könnten, dass wir keine Festbreiten- oder Binärformate für die Daten verwenden müssten.

Bis jetzt habe ich also folgende Dinge ausprobiert, die nicht wirklich einen Unterschied gemacht haben:

1. Mit der String-"Split"-Methode und dann Indizierung der Liste der Ergebnisse direkt (siehe oben), aber "Split" scheint wirklich langsam zu sein.
2. Verwendung des "reader"-Objekts im "csv"-Modul zum Parsen der Zeichenketten
3. Ändern der gesendeten Strings in ein String-Format, das ich zur direkten Instanziierung eines Objekts über "eval" verwenden kann (z. B. Senden von etwas wie "ChanVal(eventTime='2007-07-13T23:24:40.143',eventTimeExact=0,...)")

Ich versuche, ein Format mit fester Breite oder ein binäres Format zu vermeiden, obwohl mir klar ist, dass diese Formate letztendlich wahrscheinlich viel schneller wären.

Letztendlich bin ich offen für Vorschläge, wie man den Socket besser abfragen kann, wie man die Daten besser formatieren/parsen kann (obwohl wir hoffentlich bei ASCII bleiben können) oder alles andere, was Ihnen einfällt.

Gracias.

Answer 1

Sie können Python nicht schneller machen. Aber Sie können Ihre Python-Anwendung schneller machen.

Grundsatz 1: Weniger tun.

Sie können nicht weniger Eingaben parsen über alle aber Sie können weniger Input-Parsing in dem Prozess machen, der auch den Socket liest und alles andere mit den Daten macht.

Generell gilt: Tun Sie dies.

Gliedern Sie Ihre Anwendung in eine Pipeline von einzelnen Schritten.

1. Den Socket lesen, in Felder zerlegen, ein benanntes Tupel erstellen, das Tupel in eine Pipe schreiben mit etwas wie pickle .

2. Lesen einer Pipe (mit pickle ), um das benannte Tupel zu konstruieren, etwas zu verarbeiten und in eine andere Pipe zu schreiben.

3. Lesen Sie eine Pipe, führen Sie eine Verarbeitung durch, schreiben Sie in eine Datei oder ähnliches.

Jeder dieser drei Prozesse, die mit OS-Pipes verbunden sind, läuft gleichzeitig. Das bedeutet, dass der erste Prozess den Socket liest und Tupel erstellt, während der zweite Prozess Tupel konsumiert und Berechnungen durchführt, während der dritte Prozess Berechnungen durchführt und eine Datei schreibt.

Diese Art von Pipeline maximiert die Leistungsfähigkeit Ihrer CPU. Ohne zu viele schmerzhafte Tricks.

Das Lesen und Schreiben in Pipes ist trivial, da Linux Ihnen zusichert, dass sys.stdin und sys.stdout Pipes sind, wenn die Shell die Pipeline erstellt.

Bevor Sie irgendetwas anderes tun, unterteilen Sie Ihr Programm in Pipeline-Stufen.

proc1.py

import cPickle
from collections import namedtuple

ChanVal= namedtuple( 'ChanVal', ['eventTime','eventTimeExact', 'other_clock', ... ] )
for line socket:
    c= ChanVal( **line.split(',') )
    cPickle.dump( sys.stdout )

proc2.py

import cPickle
from collections import namedtuple
ChanVal= namedtuple( 'ChanVal', ['eventTime','eventTimeExact', 'other_clock', ... ] )
while True:
    item = cPickle.load( sys.stdin )
    # processing
    cPickle.dump( sys.stdout )

Diese Idee der Verarbeitung von Nameduples durch eine Pipeline ist sehr skalierbar.

python proc1.py | python proc2.py

Answer 2

Sie müssen ein Profil Ihres Codes erstellen, um herauszufinden, wo die Zeit verbraucht wird.

Das bedeutet nicht notwendigerweise, den Python-Profiler zu benutzen

Sie können zum Beispiel einfach versuchen, dieselbe CSV-Zeichenfolge 1000000 Mal mit verschiedenen Methoden zu analysieren. Wählen Sie die schnellste Methode - dividieren Sie durch 1000000. Jetzt wissen Sie, wie viel CPU-Zeit das Parsen einer Zeichenkette in Anspruch nimmt

Versuchen Sie, das Programm in Teile zu zerlegen und herauszufinden, welche Ressourcen für jeden Teil tatsächlich benötigt werden.

Die Teile, die die meiste CPU pro Eingangsleitung benötigen, sind Ihre Flaschenhälse

Auf meinem Computer gibt das folgende Programm Folgendes aus

ChanVal0 took 0.210402965546 seconds
ChanVal1 took 0.350302934647 seconds
ChanVal2 took 0.558166980743 seconds
ChanVal3 took 0.691503047943 seconds

Sie sehen also, dass etwa die Hälfte der Zeit dort von parseFromCsv . Aber auch, dass die Extraktion der Werte und ihre Speicherung in der Klasse sehr viel Zeit in Anspruch nimmt.

Wenn die Klasse nicht sofort verwendet wird, kann es schneller sein, die Rohdaten zu speichern und Eigenschaften zu verwenden, um den csvString bei Bedarf zu analysieren.

from time import time
import re

class ChanVal0(object):
    def __init__(self, csvString=None,**kwargs):
        self.csvString=csvString
        for key in kwargs:
            setattr(self,key,kwargs[key])

class ChanVal1(object):
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):
        self.lst = csvString.split(',')

class ChanVal2(object):
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        for key in kwargs:
                setattr(self,key,kwargs[key])

    def parseFromCsv(self, csvString):
        lst = csvString.split(',')
        self.eventTime=lst[1]
        self.eventTimeExact=long(lst[2])
        self.other_clock=lst[3]

class ChanVal3(object):
    splitter=re.compile("[^,]*,(?P<eventTime>[^,]*),(?P<eventTimeExact>[^,]*),(?P<other_clock>[^,]*)")
    def __init__(self, csvString=None,**kwargs):
        if csvString is not None:
            self.parseFromCsv(csvString)
        self.__dict__.update(kwargs)

    def parseFromCsv(self, csvString):
        self.__dict__.update(self.splitter.match(csvString).groupdict())

s="chan,2007-07-13T23:24:40.143,0,0188878425-079,0,0,True,S-4001,UNSIGNED_INT,name1,module1"
RUNS=100000

for cls in ChanVal0, ChanVal1, ChanVal2, ChanVal3:
    start_time = time()
    for i in xrange(RUNS):
        cls(s)
    print "%s took %s seconds"%(cls.__name__, time()-start_time)