Dieser Artikel ist der Artikel zum 6. Tag des Adventskalenders 2014 des VOYAGE GROUP Engineer Blog.
Hallo, ist in der VOYAGE GROUP @ hagino3000 im Bereich Freizeitdatenwissenschaftler tätig.
Ich denke, es gibt viele Leute, die begonnen haben, Daten für die Analyse in BigQuery zu speichern, indem sie heutzutage die BigQuery-Bewegung fahren. Wenn BigQuery verwendet wird, wird Testcode wie Aggregatstapel in der lokalen Umgebung jedoch nicht abgeschlossen, und Sie sollten auf BigQuery selbst verweisen. Dieser Artikel stellt verschiedene Ansätze vor.
Der Beispielcode verwendet Python + Nase + BigQuery-Python.
Der Grund für die Besorgnis über Testcode ist, dass BigQuery die folgenden zwei Funktionen bietet.
Zumal die Abfrage lange dauert, möchte ich dies im Test verkürzen.
BigQuery-Python-Testcode greift beispielsweise überhaupt nicht auf BigQuery zu.
https://github.com/tylertreat/BigQuery-Python/blob/master/bigquery/tests/test_client.py
Auf jeden Fall hat es den Vorteil, mit hoher Geschwindigkeit zu arbeiten, aber es ist nicht möglich zu bestätigen, ob der INSERT-Prozess und die SELECT-Anweisung tatsächlich funktionieren. Außerdem ist der Testcode voll von Mock.
Einfach ausgedrückt wäre es schön, einen Datensatz für Unit-Tests zu haben. Da dies jedoch stört, wenn mehrere Personen gleichzeitig den Test ausführen, ist für jeden Testlauf ein Datensatz erforderlich. Machen Sie jedes Mal, wenn Sie einen Test ausführen, dasselbe wie Django mit "Datenbank erstellen".
Zunächst der Prozess zum Erstellen eines Einwegdatensatzes (+ Tabelle).
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):
"""
Bereiten Sie einen Datensatz zum Testen vor
Parameters
----------
client : bigquery.client
See https://github.com/tylertreat/BigQuery-Python
Returns
-------
dataset_id : string
ID des erstellten Datensatzes(ex. ut_hoge_test_359103)
schemas : dict (key: string, value: list)
Der Schlüssel ist der Tabellenname und der Wert ist die Schemadefinitionsliste, die zum Erstellen der Tabelle verwendet wird.
"""
#Erstellen eines Datensatzes
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())
#Erstellen Sie eine Tabelle aus einer Schemadefinitionsdatei
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
Erstellen Sie einen Datensatz mit dem Testkörper und dem Setup. Um diesen Datensatz für die zu testende Verarbeitung zu verwenden, wird der Prozess des Erfassens der Datensatz-ID in Mock konvertiert.
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
#Parallel laufen
_multiprocess_can_split_ = True
def setup():
global dataset_id
global bq_client
# BigQuery-Holen Sie sich eine Python-Client-Instanz
bq_client = myapp.bq.get_client(readonly=False)
#Erstellen eines Datensatzes
dataset_id, schemas = helper.setup_dataset(bq_client, 'test_hoge')
#Verspotten Sie den Prozess, um die Dataset-ID zu erhalten
myapp.bq.get_dataset_id = mock.Mock(return_value=dataset_id)
#INSERT von Testdaten
bq_client.push_rows(dataset_id, 'events', [....Abkürzung....])
#Es ist möglicherweise nicht möglich, unmittelbar nach dem Einfügen eine Abfrage durchzuführen
time.sleep(10)
@attr('slow')
def test_calc_dau():
#Tests, die auf BigQuery verweisen
ret = myapp.calc_daily_state.calc_dau('2014/08/01')
eq_(ret, "....Abkürzung....")
@attr('slow')
def test_calc_new_user():
#Tests, die auf BigQuery verweisen
ret = myapp.calc_daily_state.calc_new_user('2014/08/01')
eq_(ret, "....Abkürzung....")
def teadown():
#Es scheint besser, den Datensatz zu löschen und zu belassen, wenn ein Test fehlgeschlagen ist
bq_client.delete_dataset(dataset_id)
In diesem Beispiel wurde angenommen, dass der zu testende Prozess ReadOnly ist, sodass der Datensatz nur einmal erstellt wurde. Es dauert 5 Sekunden für jeden Testfall, daher möchte ich 1 Test und 1 Assert einhalten.
Es dauert ungefähr 1 Sekunde, um ein Setup-Dataset zu erstellen und die Daten zu laden. Da jeder Fall Zeit braucht, ist es möglich, die Zeit durch Parallelisieren etwas zu verkürzen.
#5 Führen Sie die Tests parallel aus
nosetests --processes=5 --process-timeout=30
Multiprocess: parallel testing — nose 1.3.4 documentation http://nose.readthedocs.org/en/latest/plugins/multiprocess.html
Im obigen Beispiel wurde der Datensatz durch Einrichten des Moduls erstellt. Wenn Sie jedoch einen Prozess mit INSERT testen möchten, müssen Sie den Einfluss zwischen den Tests beseitigen. Dies wird noch länger dauern. Wenn Sie versuchen, das Ergebnis unmittelbar nach INSERT zu überprüfen und die Abfrage auszuführen, erhalten Sie das Ergebnis nicht. Nach dem EINFÜGEN müssen Sie einige Sekunden schlafen und dann die Abfrage ausführen (die etwa 5 Sekunden dauert), um das Ergebnis zu überprüfen.
test_fuga.py
#Parallel laufen
_multiprocess_can_split_ = True
@attr('slow')
class TestFugaMethodsWhichHasInsert(object):
def setup(self):
#Erstellen Sie einen Datensatz
(Abkürzung)
self.dataset_id = dataset_id
self.bq_client = bq_client
def test_insert_foo(self):
#Testen der Verarbeitung mit INSERT
def test_insert_bar(self):
#Testen der Verarbeitung mit INSERT
def teardown(self):
self.bq_client.delete_dataset(self.dataset_id)
Es ist unvermeidlich, am Ende des Tests einzuschlafen. Überlassen wir es also dem CI-Tool. In diesem Fall können BigQuery-Daten parallel ausgeführt werden, da der Einfluss zwischen Tests entfernt werden kann.
Nur die Methode zum Ausgeben einer Abfrage und zum Einfügen von Daten verwendet die tatsächliche BigQuery und trennt die Verzeichnisse als langsamen Test. Bei einer anderen Verarbeitung wird der Teil, der das Ergebnis der Abfrage zurückgibt, in Mock konvertiert.
Schreiben Sie keine Tests in den Code der Analyseaufgabe
Warten Sie, bis etwas wie DynamoDB Local angezeigt wird. Oder mach es.
Derzeit gibt es keine Option, dass dies die beste ist. Wenn Sie also Mock reduzieren und Ihren Testcode einfach halten möchten, wenden Sie sich direkt an BigQuery und umgekehrt an Mock. Entfernen Sie die Abhängigkeiten zwischen den Tests, damit die Tests parallel ausgeführt werden können. Das Erstellen eines Datensatzes dauert nicht lange, daher ist es in Ordnung, ihn für jeden Test zu erstellen.
Ich würde es begrüßen, wenn Sie mir sagen könnten, ob es ein besseres Muster gibt.
Die Gebühr von morgen ist @brtriver. Bitte freuen Sie sich darauf.
Mal sehen, was mit dem Testcode des von Python erstellten Befehls bq passiert. Sie können eine Abfrage mit `` `bq query xxx``` ausgeben, damit Sie einen solchen Test durchführen können.
https://code.google.com/p/google-bigquery-tools/source/browse/bq/bigquery_client_test.py
Es gibt keinen Test zum Ausführen der Abfrage. (´ ･ ω ･ `)