이 텍스트는 제가 여기와 제 개인 블로그에 올릴 데이터 처리 애플리케이션 테스트에 관한 일련의 텍스트 중 첫 번째입니다.
소프트웨어 엔지니어에서 데이터 엔지니어로 경력을 전환하면서 소프트웨어 엔지니어링에 대한 배경 지식이 없는 데이터 분야의 사람들과 대화를 나누기 시작했습니다. 이러한 대화에서 테스트 작성 방법
이라는 질문이 반복적으로 제기되었습니다.사실 테스트 작성은 코드 작성 방식의 변화가 필요하기 때문에 익숙하지 않은 사람들에게는 복잡한 작업처럼 보일 수 있습니다. 진실은 신비가 아니라 실천과 반복의 문제라는 것입니다. 이 기사의 주요 목표는 이제 막 시작하는 여러분에게 데이터를 처리하는 애플리케이션에 대한 테스트를 작성하여 코드의 품질과 안정성을 보장할 수 있는 방법을 보여주는 프로세스를 안내하는 것입니다.
이 텍스트는 앞으로 몇 주 동안 데이터 엔지니어링을 목표로 하는 코드에서 자동화된 테스트를 작성하는 방법을 공유할 시리즈의 일부입니다. 오늘 기사에서는 mock에 대해 조금 살펴보고 싶습니다. 여러 코드 시나리오에서 데이터 파이프라인은 연결, API 호출, 클라우드 서비스와의 통합 등을 수행하므로 이 애플리케이션을 테스트하는 방법에 약간의 혼란을 일으킬 수 있습니다. 오늘 우리는 모의 사용에 초점을 맞춘 테스트 작성을 위한 몇 가지 흥미로운 라이브러리를 탐색할 것입니다.
Mock은 테스트의 초점이 아닌 외부 종속성 또는 구성 요소의 동작을 모방하기 위해 테스트에서 사용되는 모의 개체입니다. 이를 통해 테스트 중인 코드 단위를 격리할 수 있으므로 테스트를 보다 쉽게 제어하고 예측할 수 있습니다. 모의 테스트의 사용은 단위 테스트 및 통합 테스트에서 일반적인 관행입니다.
그리고 다음과 같은 경우에 모의 객체를 사용해야 합니다.
데이터 파이프라인에서 Mocking을 사용하면 실제 인프라에 의존하지 않고 데이터베이스, 메시징 서비스 또는 API와 같은 외부 구성 요소의 표현을 생성할 수 있습니다. 이는 분산 처리를 위한 PySpark, 메시징을 위한 Kafka는 물론 AWS 및 GCP와 같은 클라우드 서비스와 같은 여러 기술을 통합하는 데이터 처리 환경에 특히 유용합니다.
데이터 파이프라인이 있는 이러한 시나리오에서 Mocking은 격리되고 빠른 테스트 실행을 촉진하여 비용과 실행 시간을 최소화합니다. 이를 통해 실제 연결이나 외부 인프라로 인해 간헐적인 오류가 발생하지 않고 각 통합이 예상대로 작동한다는 확신을 가지고 파이프라인의 각 부분을 정확하게 검증할 수 있습니다.
각 프로그래밍 언어에서는 이미 구현해야 할 Mock 기능을 제공하는 내부 모듈을 찾을 수 있습니다. Python에서 기본 unittest.mock 라이브러리는 모의 객체를 생성하기 위한 주요 도구로, 개체와 기능을 쉽고 제어적으로 시뮬레이션할 수 있습니다. Go에서 Mocking 프로세스는 언어에 기본 Mock 라이브러리가 없기 때문에 mockery와 같은 외부 패키지에서 일반적으로 지원됩니다. mockery는 Go의 기본 기능인 인터페이스에서 모의 객체를 생성하는 데 특히 유용합니다. Java에서 Mockito는 JUnit과 통합하여 강력한 단위 테스트를 용이하게 하는 모의 생성을 위한 인기 있고 강력한 라이브러리로 돋보입니다. 이러한 라이브러리는 특히 외부 데이터 소스 및 API의 시뮬레이션이 중요한 데이터 파이프라인 및 분산 시스템에서 격리된 구성 요소를 테스트하기 위한 필수 기반을 제공합니다.
Mock을 사용하는 방법에 대한 기본적인 예부터 시작해 보겠습니다. API를 호출하는 함수가 있고 이 함수에 대한 단위 테스트를 작성해야 한다고 가정해 보겠습니다.
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
테스트 시나리오에 올바르게 접근하려면 먼저 어떤 상황을 다루어야 하는지 이해해야 합니다. 함수가 REST 호출을 수행할 때 테스트에서는 최소한 두 가지 주요 시나리오, 즉 요청이 성공한 시나리오와 응답이 예상과 다른 시나리오를 고려해야 합니다. 동작을 관찰하기 위해 실제 URL로 코드를 실행할 수 있지만 이 접근 방식에는 다양한 유형의 응답을 제어할 수 없을 뿐만 아니라 URL 응답의 변경이나 최종 사용 불가능에 테스트가 취약해지기 때문에 단점이 있습니다. . 이러한 불일치를 피하기 위해 Mock을 사용하겠습니다.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
Python 단위 테스트 라이브러리의 @mock.patch 데코레이션을 사용하면 요청.get 호출을 테스트 컨텍스트 내에서 get 함수의 동작을 시뮬레이션하는 "가짜 개체"인 모의 개체로 대체하여 외부 종속성을 제거할 수 있습니다. .
모의 객체의 return_value에 대한 값을 정의함으로써 테스트 중인 함수에서 호출될 때 객체가 반환할 것으로 예상되는 것을 정확하게 지정할 수 있습니다. return_value 구조가 우리가 대체하려는 실제 객체와 동일하게 따르는 것이 중요합니다. 예를 들어 요청 모듈의 응답 객체에는 status_code와 같은 속성과 json()과 같은 메서드가 있습니다. 따라서 request.get 함수의 응답을 시뮬레이션하기 위해 모의 객체에서 직접 이러한 속성과 메서드에 예상 값을 할당할 수 있습니다.
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
이 특정 사례에서는 요청 응답을 시뮬레이션하는 데 중점을 둡니다. 즉, 외부 URL에 의존하지 않고 테스트 환경에 영향을 주지 않고 다양한 예상 결과로 함수의 동작을 테스트하는 것입니다.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
테스트에서 API 오류 응답을 시뮬레이션함으로써 기본을 넘어 404, 401, 500, 503과 같은 다양한 유형의 HTTP 상태 코드에 대해 애플리케이션 동작을 확인할 수 있습니다. 이는 더 넓은 적용 범위를 제공하고 애플리케이션이 적절하게 처리하는지 확인합니다. 각 유형의 실패에 대해 통화의 이러한 변화가 애플리케이션/데이터 처리에 어떤 영향을 미칠 수 있는지 이해합니다. POST 메서드 호출에서는 상태 코드와 호출의 기본 기능뿐만 아니라 전송 및 수신 응답의 스키마도 확인하여 추가 유효성 검사 계층을 추가하여 반환된 데이터가 다음을 따르도록 할 수 있습니다. 예상되는 형식입니다. 이러한 보다 상세한 테스트 접근 방식은 애플리케이션이 다양한 오류 시나리오를 처리할 수 있도록 준비되고 수신된 데이터가 항상 설계된 것과 일치하는지 확인하여 향후 문제를 예방하는 데 도움이 됩니다.
이제 순수 Python 코드에서 Mock을 사용하는 간단한 사례를 살펴봤으므로 Pyspark를 사용하는 코드 조각으로 사례를 확장해 보겠습니다.
PySpark 기능, 특히 필터, 그룹별, 조인과 같은 DataFrame 작업을 테스트하기 위해 모의 개체를 사용하는 것은 실제 Spark를 실행할 필요가 없도록 하여 테스트 시간을 줄이고 개발 환경을 단순화하는 효과적인 접근 방식입니다. Python의 unittest.mock 라이브러리를 사용하면 이러한 메서드의 동작을 시뮬레이션할 수 있으므로 Spark 인프라에 의존하지 않고 코드 흐름과 논리를 확인할 수 있습니다.
Spark의 데이터 프레임에 대해 필터, groupBy 및 조인 작업을 수행하는 변환이 있는 다음 함수를 살펴보겠습니다.
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
PySpark 테스트를 실행하려면 Spark 구성을 로컬에서 수행해야 합니다. 이 구성은 클래스의 모든 테스트에 사용될 Spark 인스턴스를 생성하는 setUpClass 메서드에서 수행됩니다. 이를 통해 PySpark를 독립적으로 실행할 수 있으므로 전체 클러스터에 의존하지 않고도 실제 변환 작업을 수행할 수 있습니다. 테스트가 완료된 후 TeaDownClass 메소드는 Spark 세션을 종료하여 모든 리소스가 올바르게 릴리스되고 테스트 환경이 깨끗한지 확인합니다.
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
test_transform_data 테스트에서는 변환에 사용될 데이터가 포함된 df 및 df_other에 대한 예제 DataFrame을 생성하는 것부터 시작합니다. 그런 다음 모의 객체를 적용하지 않고 변환_데이터 함수를 실행하여 필터, 그룹별 및 조인 작업이 실제로 발생하고 결과적으로 새 DataFrame이 생성되도록 합니다. 실행 후에는 Collect() 메소드를 사용하여 결과 DataFrame에서 데이터를 추출합니다. 이를 통해 이 데이터를 예상 값과 비교하여 실제적이고 정확한 방식으로 수행된 변환을 검증할 수 있습니다.
그러나 이러한 pyspark 기능 중 하나의 결과를 테스트하려는 시나리오도 있을 수 있습니다. 실행 시 병목 현상을 나타낼 수 있고 프로세스에 위험을 나타내지 않는 코드의 다른 부분을 모의해야 합니다. 따라서 이전 요청 예제에서 보았던 것처럼 함수/모듈을 모의하는 기술을 사용할 수 있습니다.
response.status_code = mock_get.return_value.status_code response.json() = mock_get.return_value.json.return_value
특정 작업에 대한 모의 테스트는 test_transform_data_with_mocked_join 메서드에서 수행되었으며, 여기서는 필터 메서드에 대해 특별히 모의를 적용했습니다. 이 모의는 조인 작업의 결과를 시뮬레이션된 DataFrame으로 대체하여 groupBy 및 조인과 같은 이전 작업이 실제 방식으로 실행될 수 있도록 합니다. 그런 다음 테스트에서는 결과 DataFrame을 예상 값과 비교하여 수행된 다른 변환을 방해하지 않고 조인 모의가 올바르게 사용되었는지 확인합니다.
이러한 하이브리드 접근 방식은 여러 가지 장점을 제공합니다. Join 및 groupBy와 같은 실제 PySpark 작업이 유지되도록 함으로써 필터와 같은 특정 작업을 모의 작업으로 대체하는 유연성을 잃지 않고 변환 논리를 검증할 수 있습니다. 결과적으로 더욱 강력하고 빠른 테스트가 가능해지며 전체 Spark 클러스터가 필요하지 않아 지속적인 코드 개발 및 검증이 더 쉬워집니다.
이 전략은 결과에 편향이 발생하지 않는 시나리오에서만 주의해서 사용해야 한다는 점을 강조하는 것이 중요합니다. 테스트의 목적은 처리가 올바르게 수행되는지 확인하는 것입니다. 실제로 함수를 테스트하지 않고 단순히 값을 할당해서는 안 됩니다. 단위 테스트 프로세스에 영향을 미치지 않을 것이라고 보장할 수 있는 모의 섹션을 만드는 것은 유효하지만 실제 동작을 검증하려면 함수를 실행해야 한다는 점을 기억하는 것이 중요합니다.
따라서 이 기능에 다른 유형의 처리를 추가하면 하이브리드 접근 방식이 훨씬 더 합리적입니다. 이 전략을 사용하면 실제 작업과 시뮬레이션 작업을 효과적으로 조합하여 더욱 강력하고 안정적인 테스트를 보장할 수 있습니다
Mock은 특히 PySpark 및 기타 클라우드 서비스를 사용할 때 효과적인 테스트를 만드는 데 귀중한 동맹입니다. Python에서 단위 테스트를 사용하여 탐색한 구현은 작업을 시뮬레이션하는 데 도움이 되었을 뿐만 아니라 데이터와 프로세스의 무결성을 유지하는 데에도 도움이 되었습니다. 모의가 제공하는 유연성을 통해 프로덕션 환경에 혼란을 가져올 염려 없이 파이프라인을 테스트할 수 있습니다. 자, 다음 도전을 할 준비가 되셨나요? 다음 텍스트에서는 AWS 및 GCP 서비스와의 통합 세계에 대해 자세히 알아보고 이러한 호출을 모의하고 파이프라인이 완벽하게 작동하는지 확인하는 방법을 보여 드리겠습니다. 다음 시간까지!
위 내용은 모조품, 그게 뭐죠?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
