안녕하세요, 데이터플랫폼셀 김민수입니다. 이번 글은 데이터 분석 라이브러리 개발기 (1)에서 이어지는 <데이터 분석 라이브러리 개발기> 두번째 글입니다. DevPlay Analytics는 PySpark을 활용하여 Amazon S3에 적재된 데이터 분석을 도와주는 파이썬 라이브러리입니다. 데이터 분석이라는 특성상 보통의 라이브러리와 같은 방법으로 테스트를 구성하기에는 까다로운 부분이 많았는데요, 이 글에서는 DevPlay Analytics 라이브러리에서 테스트 환경을 구축하고 문서 기반으로 테스트를 작성한 과정을 소개하겠습니다.
개요
DevPlay Analytics의 모듈은 S3에 적재된 특정한 데이터를 PySpark DataFrame으로 로드하고 가공하여 다시 S3로 저장하는 역할을 담당합니다. 그러므로 각 모듈에 대한 테스트를 작성한다면 미리 S3에 테스트용 샘플 데이터 셋을 저장해둔 후 로드하고 다시 테스트용 S3 버킷에 저장하여 의도한 대로 동작하는지 확인해야 합니다.
파이썬 테스트 툴 소개
unittest
파이썬에서 널리 사용하는 테스트 툴로는 unittest와 pytest가 있습니다. unittest는 JUnit이라는 자바 테스팅 프레임워크로부터 강력한 영향을 받은 파이썬 내장 유닛테스트 모듈입니다. 파이썬 자체 내부 테스트에서도 사용되는 등 여러 대형 프로젝트에서 널리 사용되는 툴이기도 합니다. unittest는 JUnit과 유사하게 모든 테스트를 unittest.TestCase를 상속받는 클래스 형태로 정의합니다. 예를 들어 str.upper의 동작을 테스트한다면,
import unittest
class TestStringMethods(unittest.TestCase):
def test_upper(self):
self.assertEqual('foo'.upper(), 'FOO')TestStringMethods 라는 클래스를 선언하고 test_upper라는 메소드를 선언하는 방식으로 테스트를 작성합니다. unittest.TestCase에는 assertEqual과 같이 테스트에 필요한 메소드들이 선언되어 있고, 테스트 작성 시 상황에 맞는 테스트 메소드를 골라서 사용합니다.
DevPlay Analytics 라이브러리의 경우 S3 데이터를 입출력하기 위해 거의 모든 동작에서 Spark Session이 필요합니다. 데이터 모듈 테스트의 관심사는 데이터 모듈이 정확한 S3 경로에서 데이터를 로드하는지를 확인하는 것이므로 Spark Session을 초기화하고 종료하는 과정은 테스트에서 확인하려는 부분이 아닙니다.
from devplay_analytcs import build
from pyspark.sql import SparkSession
class TestDWMethods(unittest.TestCase):
def setUp(self):
self.spark = SparkSession \
.builder \
.master('local[*]') \
.getOrCreate()
self.devplay_analytics = build("test_game", "prod")
def tearDown(self):
self.spark.stop()
def test_load(self):
df = self.devplay_analytics.dw.load('f_economy') \
.where(F.col('date') == '2019-01-01')
self.assertEqual(df.count(), 1104)unittest에서는 테스트에 필요한 자원의 생성과 반환을 setUp과 tearDown이라는 메소드에 정의하여 실제 테스트 코드를 분리하여 작성합니다.
pytest
pytest는 파이썬 내장 프레임워크가 아닌 써드 파티 파이썬 테스팅 툴입니다. pytest는 간결한 문법을 지원해서 unittest를 사용하는 것보다 테스트 코드를 쉽게 작성할 수 있도록 도와줍니다.
def test_upper():
assert 'foo'.upper() == 'FOO'pytest는 함수 단위로 테스트를 작성합니다. 그리고 unittest에서 assertEqual 같은 테스트용 메소드를 사용했던 것과 달리, 파이썬의 assert 문을 사용하여 대부분의 테스트를 할 수 있습니다. 그리고 테스트 실패 시 단순히 assert 문의 값이 틀렸다는 것 뿐만 아니라, 구문의 구조를 분석하여 어떤 식으로 테스트에 실패하였는지에 대한 자세한 정보를 출력해줍니다. 예를 들어, assert 문에서 list 타입의 값을 비교했다면 list 내부의 어떤 값이 존재하지 않았는지를 자세히 출력합니다.
pytest가 unittest 대비 가장 편리하다고 생각되는 점은 자원의 생성과 반환을 정의하는 fixture 문법이 매우 편하다는 것입니다. pytest에서 fixture는 @pytest.fixture decorator를 사용하여 함수 형태로 정의합니다.
import pytest
from devplay_analytcs import build
from pyspark.sql import SparkSession
@pytest.fixture
def devplay_analytics():
spark = SparkSession \
.builder \
.master('local[*]') \
.getOrCreate()
devplay_analytics = build("test_game", "prod")
yield devplay_analytics
spark.stop()
def test_load(devplay_analytics):
df = devplay_analytics.dw.load('f_economy') \
.where(F.col('date') == '2019-01-01')
assert df.count() == 1104위 예시에서 devplay_analytics 함수는 DevPlay Analytics 객체를 생성하는 pytest fixture입니다. devplay_analytics 함수에서 yield 전의 동작을 보면, Spark Session을 초기화하고, DevPlay Analytics 객체를 yield 합니다. test_load 테스트에서 파라미터 이름을 devplay_analytics라고 작성하면, 테스트 실행 시 devplay_analytics fixture를 실행하여 yield에서 반환하는 값을 테스트 파라미터로 넘겨줍니다. 그리고 테스트 완료 시 devplay_analytics의 yield 이후의 동작을 실행하여 자원을 반환합니다.
데이터 분석 라이브러리에서는 여러 가지 fixture가 복잡하게 얽힌 형태로 사용되며 자원의 생성과 반환의 시점을 잘 제어되어야 하기에 pytest의 간결하고 강력한 fixture 문법을 사용하여 개발하는 게 효율적이었습니다. 또한, 셀에서 만드는 다른 파이썬 프로젝트에서 pytest를 이미 사용하고 있었기에 경험이 많기도 했고요. 따라서 DevPlay Analytics 라이브러리 개발에서도 pytest를 활용하여 테스트를 작성하게 되었습니다. Spark Session을 비롯하여 여러 자원을 정의하는 데 유용하게 pytest를 사용할 수 있었습니다.
하지만 라이브러리를 계속 개발하다 보니 DevPlay Analytics 라이브러리는 보통의 파이썬 라이브러리처럼 테스트하기에는 까다로운 부분이 많았습니다.
문제 정의 - 데이터 분석 라이브러리는 테스트가 쉽지 않다
데이터 분석 라이브러리 개발기에서 소개해 드린 것과 같이, DevPlay Analytics 라이브러리의 데이터 계층은 IO 계층과 설정 계층을 호출하는 식으로 구현되어 있습니다. 따라서, 데이터 계층 함수의 테스트를 작성하기 위해서는 IO 계층과 설정 계층을 mocking 한 후 데이터 계층 함수 기능에 대한 테스트를 작성해야 합니다. 설정 계층은 단순히 특정한 형태의 파일을 읽어서 설정값을 반환해주므로 해당 파일에 대한 fixture를 정의하면 쉽게 원하는 형태의 설정값들을 만들 수 있었습니다. 그런데, IO 계층은 비즈니스 로직 자체가 PySpark와 S3에 기대고 있는 부분이 많아 mocking 하기에는 까다로운 부분이 많았습니다.
그래서 처음에는 S3에 테스트용 버킷과 fixture 데이터를 올려 두는 방식으로 테스트를 구성했습니다.
from pyspark.sql import functions as F
def test_load(devplay_analytics: DevPlayAnalytics):
df = devplay_analytics.dw.load('f_economy') \
.where(F.col('date') == '2019-01-01')
assert df.count() == 1104위 테스트는 f_economy라는 DW 테이블을 로드한 후 해당 테이블의 row 수가 예상한 대로 있는지를 확인합니다. DevPlay Analytics 객체인 devplay_analytics도 fixture로 구성을 했는데, 해당 fixture에서 SparkSession 초기화 / 종료, 설정 계층의 파일에 대한 mocking 등도 담당하게 하여 devplay_analytics만 fixture로 참조하여 테스트를 작성할 수 있게 했습니다. 설정 파일 mock에 각 데이터 계층의 테스트 데이터에 해당하는 경로들을 적어두었기에 테스트 경로에 테스트용 데이터를 미리 적재해뒀습니다.
이렇게 실제로 S3에 데이터를 쌓아두고 이 데이터 기반으로 테스트를 작성하는 방식으로 각 데이터 계층 모듈의 기능을 어느 정도 테스트할 수 있었습니다. 하지만, 실제로 S3에 접근하여 데이터를 입출력해야 하다 보니 접근 권한이 있는 경우에만 테스트 코드를 실행해볼 수 있었습니다. 간단한 CI를 구성하여 테스트를 실행하고 싶었는데 개발 당시에는 GitHub Actions에서 테스트용 S3 버킷 접근 권한을 부여하기가 쉽지 않았습니다. 한편, 매번 테스트마다 새로운 데이터를 생성해야 했고, 테스트 자체에도 시간이 오래 걸려 좋은 접근 방법은 아닌 것 같았습니다. 또한, 테스트용 데이터는 S3에 한 벌만 존재하므로 각 테스트 실행 별로 환경을 격리할 수가 없는 문제도 있었습니다.
MinIO를 사용한 로컬 테스트 환경 구축
따라서 S3 API 자체를 테스트마다 따로 띄울 수 있으면 좋을 것 같았습니다. 그러던 중 MinIO라는 프로젝트를 발견했습니다. MinIO는 오픈소스 오브젝트 스토리지 프로젝트인데요, S3 SDK와 호환이 되는 특징이 있습니다.
MinIO를 활용하면 직접 S3에 접근하지 않고도 로컬에서 S3 SDK와 호환되는 API 서버를 띄운 후 테스트에 필요한 데이터를 로컬에서 정의할 수 있습니다. 그래서 S3에 데이터를 미리 저장해두는 방식 대신 MinIO를 사용하여, 테스트 시작 시 MinIO 서버를 띄운 후 이 서버에 테스트에 필요한 버킷과 데이터를 생성한 후 테스트를 실행하도록 구성했습니다.
MinIO 서버 fixture 사용
pytest에서 사용할 수 있는 MinIO 서버 fixture는 pytest-server-fixtures 라는 프로젝트에 구현되어 있어 DevPlay Analytics의 dev 디펜던시에 pytest-server-fixtures[s3]를 추가했습니다. 그리고 pytest-server-fixtures에 구현된 s3_server fixture를 사용하여 MinIO 서버를 띄운 후, MinIO 서버 접속 정보를 주입해 Spark Session을 생성했습니다. 이런 방식으로 생성한 Spark Session을 fixture로 정의하여 각 테스트에서 MinIO + Spark Session을 사용할 수 있도록 했습니다.
@pytest.fixture(scope='session')
def spark(s3_server: MinioServer):
spark = (SparkSession
.builder
.master('local[*]')
# s3a endpoint를 로컬의 MinIO 서버로 설정하여 S3 API 호출을 MinIO서버로 하도록 합니다.
.config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:{}'.format(s3_server.port))
# s3_server fixture에서 MinIO 서버 실행 시 설정한 access key, secret key를 설정해줍니다.
.config('spark.hadoop.fs.s3a.access.key', s3_server.aws_access_key_id)
.config('spark.hadoop.fs.s3a.secret.key', s3_server.aws_secret_access_key)
.config('spark.hadoop.fs.s3a.path.style.access', 'true')
.config('spark.hadoop.fs.s3a.impl', 'org.apache.hadoop.fs.s3a.S3AFileSystem')
.getOrCreate())
# 여기서 반환된 Spark Session을 각 테스트에서 사용합니다.
yield spark
# 테스트 완료 시 Spark Session을 종료합니다.
spark.stop()
# (spark fixture 자원 반환 시 s3_server fixture에서 MinIO 서버를 종료하고 리소스를 삭제합니다)그리고 테스트에 필요한 bucket과 데이터들도 역시 fixture로 정의하여 테스트에서 레퍼런스하여 사용할 수 있도록 하였습 니다.
@pytest.fixture(scope='session')
def bucket_logging_test_game(s3_server: MinioServer):
# MinIO 서버에 create_bucket API를 호출하여 버킷을 생성합니다.
s3_server.get_s3_client().create_bucket(Bucket='logging-test-game')
@pytest.fixture(scope='session')
def bucket_analytics_test_game(s3_server: MinioServer):
s3_server.get_s3_client().create_bucket(Bucket='analytics-test-game')@pytest.fixture(scope='session')
def data_dw_f_churn(spark: SparkSession, bucket_analytics_test_game):
# 여기서 spark는 앞서 정의한 spark fixture입니다.
# 따라서 DataFrame을 생성한 후 Parquet로 적재를 하면
# fixture에서 설정한 대로 MinIO 서버를 호출하여 저장됩니다.
spark.createDataFrame([
Row(date='2020-02-01', countryCode='KR', os='android', userCnt=6040),
Row(date='2020-02-01', countryCode='KR', os='ios', userCnt=1962),
]).write.partitionBy('date').parquet('s3a://analytics-test-game/dw/env=prod/f_churn')마지막으로, DevPlay Analaytics 객체를 생성할 때, Spark Session과 설정값들 등의 fixture를 전부 모아 객체를 생성하도록 하였습니다.
@pytest.fixture(scope='session')
def devplay_analytics(
# SparkSession
spark,
# data_analytics_config 설정
data_analytics_config,
# minio server
s3_server,
# boto3 resource (글에서 다루고 있지는 않지만, boto3 라이브러리의 S3 리소스도 MinIO로 통신하도록 monkey patching 했습니다)
boto3_resource,
):
return build(
game_code='test_game',
env_code='prod',
)MinIO 서버 fixture를 사용하여 테스트 실행
이렇게 만든 fixture들을 조합하여 각 모듈에 대한 테스트를 작성했습니다. DW 모듈 데이터 로드에 대해서 테스트를 작성하기 위하여 devplay_analytics , data_dw_f_churn 두 fixture를 사용하여 아래와 같이 테스트 코드를 작성할 수 있습니다.
def test_load(devplay_analytics: DevPlayAnalytics, data_dw_f_churn):
df = devplay_analytics.dw.load('f_churn')
assert df.collect() == [
Row(date='2020-02-01', countryCode='KR', os='android', userCnt=6040),
Row(date='2020-02-01', countryCode='KR', os='ios', userCnt=1962),
]이 테스트는 devplay_analytics fixture를 첫번째 인자로 사용하고 있으므로, ① DevPlay Analytics Fixture를 생성하고, ② Spark Session을 초기화 시키며, ③ MinIO 서버를 실행시켜 Spark Session에서 연결할 수 있게 합니다. 그리고 두번째 인자로 data_dw_f_churn fixture를 사용하므로, ④ Log Data Fixture를 생성하기 위해 ⑤ Spark Session Fixture와 ⑥ Bucket Fixture를 생성합니다. Bucket Fixture를 생성하기 위해서도 ⑦ MinIO Server Fixture가 필요합니다.
따라서 테스트 실행 시 MinIO 서버가 생성되고, MinIO 서버에 필요한 bucket과 데이터가 적재되며, 해당 MinIO 서버에 Spark로 접근하여 테스트 코드가 진행되게 됩니다.
다른 테스트를 한 번에 실행시킬 경우 MinIO 서버 fixture는 공용으로 사용하면서 (다른 버킷과 데이터가 필요할 경우) 버킷과 데이터를 생성합니다.
모든 테스트가 완료되면 MinIO 서버와 관련된 리소스들이 삭제됩니다.
소결론: MinIO fixture를 사용하면 쉽고 가볍게 로컬 테스트를 할 수 있습니다.
정리하자면 MinIO fixture를 통하여 Amazon S3에 의존성이 없으면서도, 테스트 시 완전히 격리된 환경에서 S3 입출력이 필요한 기능의 테스트를 빠르게 실행할 수 있었습니다.
테스트 코드를 작성하면서 레퍼런스 문서도 만들 수 있다면
doctest는 파이썬 코드 중 docstring에 작성된 파이썬 인터프리터 코드를 실행하여 결과가 정확하게 작동하는지 확인해주는 파이썬 모듈입니다.
def factorial(n):
"""Return the factorial of n, an exact integer >= 0.
>>> [factorial(n) for n in range(6)]
[1, 1, 2, 6, 24, 120]
"""
...위 코드를 doctest 모듈로 실행시키면 파이썬 인터프리터 부분(>>> 오른쪽 코드)을 실행하여 아래와 같은 결과가 나오는지 검증해줍니다. doctest는 파이썬 내장 기능이기도 하지만, pytest의 doctest integration을 사용해서 실행할 수도 있습니다.
DevPlay Analytics 라이브러리는 데이터 분석가분들이 쉽게 사용하실 수 있도록 docstring에 각 함수의 용례를 포함한 자세한 설명을 작성해두고 있습니다. 또한, Sphinx라는 파이썬 문서화 도구를 사용해서 docstring을 함수 레퍼런스 문서로 빌드하여 제공하고 있습니다. 그런데, 모듈의 동작이나 형상이 바뀔 때마다 docstring을 최신화되지 않고 빠뜨리기도 하여 관리가 쉽지 않았습니다.
그러다 보니 docstring에 작성한 용례들이 실제로 실행이 되는지 체크를 하면 좋겠다는 생각이 들었습니다. 한편, fixture 기반으로 각 모듈 테스트를 작성하다 보니 사실 테스트 코드 대부분이 실제 함수 동작을 설명하는 예시 코드와 비슷하다고 느껴졌습니다. 그래서 pytest doctest 모듈을 활용하여 docstring 코드를 실행하는 방식으로 테스트를 구성하게 되었습니다.
class DW:
def load(
self,
table: str,
):
"""
DW 테이블을 로드하여 불러옵니다.
.. testsetup::
>>> getfixture('data_dw_f_churn')
.. doctest::
>>> f_churn = devplay_analytics.dw.load('f_churn')
>>> f_churn \
.where(F.col('date') == devplay_analytics.env.target_date) \
.orderBy("countryCode", "os") \
.show(truncate=False)
+-----------+-------+-------+----------+
|countryCode|os |userCnt|date |
+-----------+-------+-------+----------+
|KR |android|6040 |2020-02-01|
|KR |ios |1962 |2020-02-01|
+-----------+-------+-------+----------+
<BLANKLINE>
:param table: 불러올 테이블의 이름
:return: DataFrame
"""
...이렇게 작성된 doctest를 pytest에서 --doctest-modules 옵션을 줘서 실행시키면 >>> 오른쪽에 작성된 코드를 실행하여 결과가 docstring과 같은지 테스트합니다. 함수로 정의했던 테스트와 다르게 작성해야 하는 점은, 테스트 실행 시 필요한 fixture를 getfixture 라는 헬퍼 함수를 사용해서 호출해야 한다는 점입니다.
작성된 doctest는 sphinx automodule을 사용하여 아래와 같은 파이썬 문서로 빌드할 수 있습니다.
소결론: doctest를 사용하여 테스트 코드를 작성하는 것만으로 레퍼런스 문서를 만들 수 있습니다.
라이브러리를 제작하면서 작성해야 했던 매우 많은 레퍼런스 문서를 테스트 코드로 활용하다 보니, 문서화 커버리지(즉, pytest 테스트 커버리지)를 거의 100%에 가깝게 넓힐 수 있었고, outdate 되지 않는 실제로 작동하는 코드만 작성된 레퍼런스 문서를 얻을 수 있었습니다.
결론
MinIO 기반의 fixture를 활용하는 pytest + doctest기반의 테스트를 구성하여
- Amazon S3 의존성 없이 데이터 분석 라이브러리의 테스트 코드를 작성하였고
- 모든 API 레퍼런스 문서가 테스트 코드로도 활용되는
테스트 환경을 만들 수 있었습니다. 이런 테스트 환경을 갖추었기에 라이브러리를 개발하는 데 있어서 적은 공수로도 탄탄하게 테스트를 할 수 있었 으며 방대한 문서를 쉽게 작성하였습니다.
저희 데이터플랫폼셀은 탄탄한 엔지니어링을 통해 이런 데이터가 잘 흐르게 하기 위한 기반을 만드는 일들을 하고 있습니다. 저희 셀에서 하는 일들에 관심이 있으시면 데브시스터즈 채용 페이지를 확인해주세요 :)
감사합니다.