[PYTHON] Es ist sehr nützlich, Target in Luigi die Methoden save () und load () hinzuzufügen

Das ist alles für den Titel!

… Weil es etwas ist, mit ein paar Beispielen.

Wie jeder weiß, der es verwendet hat, abstrahiert der Workflow-Manager Luigi die von einer Aufgabe generierten Daten und behandelt sie als geerbte Klasse von "Ziel". Dies funktioniert vorerst, wenn Sie die Methode exist () implementieren und prüfen können, ob sie bereits vorhanden ist. Wenn es sich beispielsweise um eine lokale Datei handelt, wird "LocalTarget" verwendet. Wenn Sie sie jedoch vorerst gemäß dem Lernprogramm schreiben, sieht dies wie folgt aus.

from luigi import Task, ExternalTask, LocalTarget
import pandas as pd


class RawFile(ExternalTask):

    def output(self):
        return LocalTarget('path/to/file.csv')


class Aggregation(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = pd.read_csv(self.input()[0], sep='\t', parse_dates=[7, 8], encoding='cp932')
        ...

Nun ... nun, hier kommt eine andere Aufgabe, die von "RawFile" abhängt. Lassen Sie uns darüber nachdenken. Ich fange allmählich an, diese Art von Code zu schreiben ...

class Plot(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = pd.read_csv(

Warte eine Minute. Erinnern Sie sich an die richtigen Optionen, die Sie für jedes Projekt, an dem Sie beteiligt sind, an den CSV-Datei-Parser übergeben können? ** Ich erinnere mich nicht. ** Möchten Sie jedes Mal eine Option schreiben? Ich möchte nicht ** absolut ** oft schreiben. Luigi hat Target in den Pfad der Originaldaten abstrahiert, aber das ist stumpf. Das ist richtig, ** Sie müssen nur die Lesung abstrahieren **.

from luigi import Task, ExternalTask, LocalTarget
import pandas as pd


class RawFileTarget(LocalTarget):

    path = 'path/to/file.csv'

    def __init__(self):
        super(RawFileTarget, self).__init__(path)

    def load(self):
        return pd.read_csv(self.fn, sep='\t', parse_dates=[7, 8], encoding='cp932')

class RawFile(ExternalTask):

    def output(self):
        return RawFileTarget()

Definieren wir "Ziel", das "load ()" definiert.

Auf diese Weise jede abhängige Aufgabe

class Aggregation(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = self.input()[0].load()
        ...


class Plot(Task):

    def requires(self):
        yield RawFile()

    def run(self):
        df = self.input()[0].load()
        ...

Es ist erfrischend. Später erhielt die für die Datenverarbeitung zuständige Person eine CSV-Datei mit anderen Spezifikationen als der mit den Worten "Entschuldigung, die, die ich zuvor erhalten habe, ich habe einen Fehler gemacht ...", aber ich habe sie falsch verstanden. Unabhängig davon, ob die übergebene Datei Excel ist oder nicht, können Sie einfach "RawFileTarget.load ()" neu schreiben. Glücklich!

Wie der Titel schon sagt, schreibe ich gerne "save ()", damit es "load ()" entspricht, aber ich werde das Codebeispiel weglassen.