Utilisez gokart pour tester les associations pandas sur votre pipeline de données.

écrire

• Comment vérifier les pandas liés à l'aide de gokart
• La première consiste à vérifier si l'entrée pd.Dataframe se termine normalement lorsqu'elle est vide. ―― La seconde consiste à vérifier si chaque colonne a le type attendu lors du vidage.

Qu'est-ce que le gokart?

--OSS développé par M3 et fringe81

• Enveloppe luigi développé par Spotify pour le rendre plus facile à utiliser. En particulier, la quantité de code à écrire est réduite.

Version cible

• 0.3.11

Confirmez s'il faut se terminer normalement lorsque pd.Dataframe de l'entrée est vide

• Le code ci-dessous provoque une erreur lorsque pd.Dataframe est vide. «C'est une condition préalable pour écrire un test unitaire, mais il y a eu de nombreux cas où je n'ai pas pu le saisir.

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()
        #Quand df est vide`KeyError`Se produit.
        df['user'] = df['user'].apply(lambda x: f'user_{x}')
        self.dump(df)


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

• Vous pouvez vérifier si cela fonctionne même s'il est vide en exécutant ce qui suit. --Ajoutez --test-run-pandas et --test-run-namespace = sample.

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

--Message

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

Vérifiez si chaque colonne a le type attendu lors du vidage

• Vérifiez si le type de pandas a changé en raison d'une opération involontaire.
• Vous pouvez vérifier chaque tâche, mais définissez une combinaison de nom_colonne et de type pour chaque espace de noms. --Hériter de gokart.PandasTypeConfig 'et définir les règles dans l'espace de noms.

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()
        #Le type d'utilisateur est passé de int à str.
        df['user'] = df['user'].apply(lambda x: f'user_{x}')
        self.dump(df)


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

--Si vous exécutez la tâche normalement, le type sera vérifié automatiquement.

• Dans ce cas, TaskA échoue car ʻuser a été changé en str`.

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

--Message

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 =====