Comment rédiger un test de traitement utilisant BigQuery

Cet article est l'article du 6ème jour du Calendrier de l'Avent 2014 du blog de l'ingénieur VOYAGE GROUP.

Bonjour, est engagé dans les affaires de scientifique de données de temps libre dans VOYAGE GROUP [@ hagino3000] est (https://twitter.com/hagino3000).

Je pense que de nombreuses personnes ont commencé à intégrer des données à des fins d'analyse dans BigQuery en suivant le mouvement BigQuery ces jours-ci. Cependant, lorsque BigQuery est utilisé, le code de test tel que les lots agrégés ne sera pas terminé dans l'environnement local et vous souhaiterez vous référer à BigQuery lui-même. Cet article présente plusieurs approches.

L'exemple de code utilise Python + nose + BigQuery-Python.

Qu'est-ce qui ne va pas

La raison de s'inquiéter à propos du code de test est que BigQuery possède les deux fonctionnalités suivantes.

  1. Je ne parviens pas à créer un environnement local
  2. Il faut environ 5 secondes pour interroger même une petite quantité de données

D'autant que la requête prend beaucoup de temps, je veux raccourcir cela dans le test.

Faire tout se moquer

Le code de test BigQuery-Python, par exemple, n'accède pas du tout à BigQuery.

https://github.com/tylertreat/BigQuery-Python/blob/master/bigquery/tests/test_client.py

Quoi qu'il en soit, il a le mérite de fonctionner à grande vitesse, mais il n'est pas possible de confirmer si le processus INSERT et l'instruction SELECT fonctionnent réellement. De plus, le code de test est plein de Mock.

Créer un jeu de données pour chaque module de test

Pour faire simple, il serait bien d'avoir un ensemble de données pour les tests unitaires, mais comme cela interfère lorsque plusieurs personnes exécutent le test en même temps, un ensemble de données est requis pour chaque test. Faites la même chose que Django fait avec Create Database chaque fois que vous exécutez un test.

Tout d'abord, le processus de création d'un ensemble de données jetables (+ table).

tests/helper.py


# coding=utf-8
from datetime import datetime
import glob
import json
import os
import random
import re


def setup_dataset(client, test_name):
    """
Préparer un ensemble de données pour les tests

    Parameters
    ----------
    client : bigquery.client
        See https://github.com/tylertreat/BigQuery-Python

    Returns
    -------
    dataset_id : string
ID de l'ensemble de données créé(ex. ut_hoge_test_359103)

    schemas : dict (key: string, value: list)
La clé est le nom de la table et la valeur est la liste de définition de schéma utilisée pour créer la table.
    """
    #Créer un jeu de données
    dataset_id = 'ut_%s_%d' % (test_name ,int(random.random() * 1000000))
    client.create_dataset(
        dataset_id,
        friendly_name='For unit test started at %s' % datetime.now())

    #Créer une table à partir d'un fichier de définition de schéma
    schemas = {}
    BASE_DIR = os.path.normpath(os.path.join(os.path.dirname(__file__), '../'))
    for schema_file in glob.glob(os.path.join(BASE_DIR, 'schema/*.json')):
        table_name = re.search(r'([^/.]+).json$', schema_file).group(1)
        schema = json.loads(open(schema_file).read())
        schemas[table_name] = schema
        client.create_table(dataset_id, table_name, schema)

    return dataset_id, schemas

Créez un ensemble de données avec le corps de test et la configuration. De plus, afin d'utiliser cet ensemble de données pour le traitement à tester, le processus d'acquisition de l'ID de l'ensemble de données est converti en Mock.

test_hoge.py


# coding=utf-8
import time

import mock
from nose.tools import eq_
from nose.plugins.attrib import attr

import myapp.bq
import myapp.calc_daily_state
from . import helper

dataset_id = None
bq_client = None

#Exécuter en parallèle
_multiprocess_can_split_ = True


def setup():
    global dataset_id
    global bq_client
    # BigQuery-Obtenir une instance de client Python
    bq_client = myapp.bq.get_client(readonly=False)
    #Créer un jeu de données
    dataset_id, schemas = helper.setup_dataset(bq_client, 'test_hoge')
    #Simulez le processus pour obtenir l'ID de l'ensemble de données
    myapp.bq.get_dataset_id = mock.Mock(return_value=dataset_id)
    #INSÉRER des données de test
    bq_client.push_rows(dataset_id, 'events', [....Abréviation....])
    #Il peut ne pas être possible d'interroger immédiatement après INSERT, alors dormez
    time.sleep(10)


@attr('slow')
def test_calc_dau():
    #Tests faisant référence à BigQuery
    ret = myapp.calc_daily_state.calc_dau('2014/08/01')
    eq_(ret, "....Abréviation....")


@attr('slow')
def test_calc_new_user():
    #Tests faisant référence à BigQuery
    ret = myapp.calc_daily_state.calc_new_user('2014/08/01')
    eq_(ret, "....Abréviation....")


def teadown():
    #Il semble préférable de supprimer l'ensemble de données et de le laisser en cas d'échec du test
    bq_client.delete_dataset(dataset_id)

Dans cet exemple, le traitement à tester était supposé être en lecture seule, de sorte que l'ensemble de données n'a été créé qu'une seule fois. Cela prend 5 secondes pour chaque cas de test, donc je veux adhérer à 1 test et 1 assertion.

Il faut environ 1 seconde pour créer un jeu de données de configuration et charger les données. Étant donné que chaque cas prend du temps, il est possible de raccourcir le temps dans une certaine mesure en parallélisant.

#5 Exécuter des tests en parallèle
nosetests --processes=5 --process-timeout=30

Multiprocess: parallel testing — nose 1.3.4 documentation http://nose.readthedocs.org/en/latest/plugins/multiprocess.html

Créer un ensemble de données pour chaque fonction de test

Dans l'exemple ci-dessus, l'ensemble de données a été créé par la configuration du module, mais quand il s'agit de tester un processus avec INSERT, il est nécessaire d'éliminer l'influence entre les tests. Cela prendra encore plus de temps. Parce que si vous essayez de vérifier le résultat immédiatement après INSERT et d'exécuter la requête, vous n'obtiendrez pas le résultat. Après INSERT, vous devez dormir pendant quelques secondes, puis exécuter la requête (qui prend environ 5 secondes) pour vérifier le résultat.

test_fuga.py


#Exécuter en parallèle
_multiprocess_can_split_ = True

@attr('slow')
class TestFugaMethodsWhichHasInsert(object):
    def setup(self):
        #Créer un jeu de données
        (Abréviation)
        self.dataset_id = dataset_id
        self.bq_client = bq_client
    
    def test_insert_foo(self):
        #Tester le traitement avec INSERT

    def test_insert_bar(self):
        #Tester le traitement avec INSERT

    def teardown(self):
        self.bq_client.delete_dataset(self.dataset_id)

Il est inévitable de s'endormir à la fin du test, laissons donc à l'outil CI. Dans ce cas, les données BigQuery peuvent être exécutées en parallèle, car l'influence entre les tests peut être supprimée.

Un compromis équilibré

Seule la méthode d'émission d'une requête et d'insertion de données utilise le BigQuery réel et sépare les répertoires comme un test lent. Pour les autres traitements, la partie qui renvoie le résultat de la requête est convertie en Mock.

Radical

N'écrivez pas de tests dans le code de la tâche d'analyse

Future école

Attendez que quelque chose comme DynamoDB Local apparaisse. Ou faites-le.

Résumé

Il n'existe actuellement aucune option indiquant qu'il s'agit de la meilleure solution.Par conséquent, si vous souhaitez réduire Mock et simplifier votre code de test, reportez-vous directement à BigQuery et vice versa, utilisez Mock. Supprimez les dépendances entre les tests afin que les tests puissent être exécutés en parallèle. La création d'un ensemble de données ne prend pas longtemps, vous pouvez donc le créer pour chaque test.

J'apprécierais que vous me disiez s'il y a un meilleur modèle.

Les frais de demain sont de @brtriver, attendez-le avec impatience.

Remarque: quel est le code de test pour google-bigquery-tools?

Voyons ce qu'il advient du code de test de la commande bq faite par Python. Vous pouvez émettre une requête avec bq query xxx```, vous devriez donc pouvoir avoir un tel test.

https://code.google.com/p/google-bigquery-tools/source/browse/bq/bigquery_client_test.py

Il n'y a pas de test pour exécuter la requête. (´ ・ ω ・ `)

Recommended Posts

Comment rédiger un test de traitement utilisant BigQuery
Comment écrire une instruction ShellScript Bash for
[Python] Comment écrire une docstring conforme à PEP8
Comparez comment écrire le traitement des listes par langue
Comment écrire une classe méta qui prend en charge à la fois python2 et python3
Comment faire un test unitaire Part.1 Modèle de conception pour l'introduction
Comment tester sur une page authentifiée par Django
Un mémorandum sur la façon d'écrire des pandas que j'ai tendance à oublier personnellement
Comment écrire un document tuple nommé en 2020
[Go] Comment écrire ou appeler une fonction
Comment écrire une instruction de cas bash ShellScript
[BigQuery] Comment utiliser l'API de BigQuery pour Python -Création de table-
Je veux écrire en Python! (2) Écrivons un test
[Go] Comment créer une erreur personnalisée pour Sentry
Comment écrire un type liste / dictionnaire de Python3
Comment créer un référentiel local pour le système d'exploitation Linux
Comment créer un environnement de développement pour TensorFlow (1.0.0) (Mac)
Écrire du code dans UnitTest une application Web Python
[Introduction à Python] Comment écrire des instructions répétitives à l'aide d'instructions for
Comment utiliser pip, un système de gestion de paquets indispensable pour utiliser Python
Comment tester cette exception est déclenchée dans python unittest
Comment gérer un README pour github et PyPI
Comment appeler une fonction
Comment créer un package Python (écrit pour un stagiaire)
Comment remplacer une valeur numérique pour une correspondance partielle (Remarque 1)
Comment pirater un terminal
Je n'oublierai jamais comment écrire un script shell, n'oubliez pas! !!
Comment écrire des conseils de type pour les variables qui sont affectées plusieurs fois sur une ligne
[Introduction à Python] Comment utiliser l'opérateur in dans l'instruction for?
Spigot (Paper) Introduction à la création d'un plug-in pour 2020 # 01 (Construction de l'environnement)
Une histoire qui avait du mal à traiter en boucle 3 millions de données d'identification
Ecrire un programme qui abuse du programme et envoie 100 e-mails
Comment créer une grande quantité de données de test dans MySQL? ??
Un script qui utilise boto pour télécharger un dossier spécifié sur Amason S3
Comment écrire une docstring pour créer un document tuple nommé avec sphinx
Comment continuer le traitement après avoir renvoyé une réponse sur aiohttp Server
Comment faire une traduction japonais-anglais
Ecrire un test piloté par table en C
[Django] Comment tester le formulaire [TDD]
Comment mettre un lien symbolique
[Pour les non-programmeurs] Comment marcher Kaggle
Comment écrire sobrement avec des pandas
Comment créer un package Conda
Écrire la sortie standard dans un fichier
Comment créer un robot - Avancé
Comment créer une fonction récursive
Comment créer un pont virtuel
Comment créer un Dockerfile (basique)
[Blender] Comment créer un plug-in Blender
Comment supprimer un conteneur Docker
Comment écrire une image de base Docker
Comment écrire un fichier wsgi.py indépendant de l'environnement Django1.9
Comment créer un robot - Basic
Remarques sur la rédaction de requirements.txt