[PYTHON] Verwenden Sie Gokart, um Pandas-Assoziationen in Ihrer Datenpipeline zu testen.

schreiben

• So überprüfen Sie Pandas im Zusammenhang mit Gokart
• Der erste besteht darin, zu überprüfen, ob der Eingabe-PD-Datenrahmen normal endet, wenn er leer ist. ――Die zweite besteht darin, beim Dumping zu überprüfen, ob jede Spalte den erwarteten Typ hat.

Was ist Gokart?

--OSS entwickelt von M3 und fringe81 --Wraps luigi wurde von Spotify entwickelt, um die Verwendung zu vereinfachen. Insbesondere die Menge des zu schreibenden Codes wird reduziert.

Zielversion

• 0.3.11

Bestätigung der normalen Beendigung, wenn pd.Dataframe der Eingabe leer ist

• Der folgende Code verursacht einen Fehler, wenn pd.Dataframe leer ist. ――Es ist eine Voraussetzung, einen Unit-Test zu schreiben, aber es gab viele Fälle, in denen ich ihn nicht abholen konnte.

class DataTask(gokart.TaskOnKart):
    task_namespace = 'sample'

    def run(self):
        df = pd.DataFrame(dict(user=[1, 2], item=['a', 'b']))
        self.dump(df)


class TaskA(gokart.TaskOnKart):
    task_namespace = 'sample'

    def requires(self):
        return DataTask()

    def run(self):
        df = self.load_data_frame()
        #Wenn df leer ist`KeyError`Tritt ein.
        df['user'] = df['user'].apply(lambda x: f'user_{x}')
        self.dump(df)


if __name__ == '__main__':
    gokart.run()

• Sie können überprüfen, ob es funktioniert, auch wenn es leer ist, indem Sie Folgendes ausführen.
• Fügen Sie --test-run-pandas und --test-run-namespace = sample hinzu.

$ python main.py sample.TaskA --local-scheduler --test-run-pandas --test-run-namespace=sample

--Botschaft

gokart test results:
status=OK; namespace=sample; name=DataTask; id=10f87ddcf3df71d786a023ae5e0bbc98;
status=NG; namespace=sample; name=TaskA; id=44e9690a4d2182a9bed6b6d9730291bd; message=<class 'KeyError'>: user

Überprüfen Sie beim Dumping, ob jede Spalte den erwarteten Typ hat

--Überprüfen Sie, ob sich der Pandas-Typ aufgrund einer unbeabsichtigten Operation geändert hat.

• Sie können für jede Aufgabe prüfen, aber für jeden Namespace eine Kombination aus Spaltenname und Typ definieren.
• Erben Sie gokart.PandasTypeConfig'und definieren Sie die Regeln im Namespace.

class SamplePandasTypeCheck(gokart.PandasTypeConfig):
    task_namespace = 'sample'

    @classmethod
    def type_dict(cls) -> Dict[str, Any]:
        return {'user': int}


class DataTask(gokart.TaskOnKart):
    task_namespace = 'sample'

    def run(self):
        df = pd.DataFrame(dict(user=[1, 2], item=['a', 'b']))
        self.dump(df)


class TaskA(gokart.TaskOnKart):
    task_namespace = 'sample'

    def requires(self):
        return DataTask()

    def run(self):
        df = self.load_data_frame()
        #Der Benutzertyp wurde von int in str geändert.
        df['user'] = df['user'].apply(lambda x: f'user_{x}')
        self.dump(df)


if __name__ == '__main__':
    gokart.run()

• Wenn Sie die Aufgabe normal ausführen, wird der Typ automatisch überprüft.
• In diesem Fall schlägt "TaskA" fehl, weil "Benutzer" in "str" geändert wurde.

$ python main.py sample.TaskA --local-scheduler

--Botschaft

gokart.pandas_type_config.PandasTypeError: expected type is "<class 'int'>", but "<class 'str'>" is passed in column "user".

...

===== Luigi Execution Summary =====

Scheduled 2 tasks of which:
* 1 ran successfully:
    - 1 sample.DataTask(...)
* 1 failed:
    - 1 sample.TaskA(...)

This progress looks :( because there were failed tasks

===== Luigi Execution Summary =====