코딩 작업 추정: 무엇이 잘못될 수 있나요?

“기존 DataFrame에 해시 추가” 작업이 며칠이 걸리던 작업에서 거의 전체 스프린트를 소모하게 된 과정은 다음과 같습니다.

2022년 2분기에 저는 REST 서비스에서 시장 데이터를 가져와 BigQuery 테이블에 저장하는 데이터 파이프라인 작업을 시작했습니다. 이는 파이프라인에 대한 높은 수준의 설명입니다. 흥미로운 부분은 데이터를 쿼리하여 DataFrame으로 변환한 다음 AirFlow의 GCSToBigQueryOperator

를 사용하여 BigQuery 테이블에 업로드하는 방식입니다.

처음에는 작성이 간단해 보였지만 Airflow의 '멱등원' 원칙으로 인해 약간의 어려움이 추가되었습니다. 이 REST 서비스에서 가져올 항목은 다른 테이블에 의해 결정되었으며 JOB이 멱등성인 경우에도 참조로 사용된 테이블은 두 번의 실행 사이에서 변경될 수 있습니다. 추가 시간을 투자한 후 2022년 3분기 말까지 데이터 엔지니어 파이프라인과의 대화가 준비되었습니다.

2024년 1분기로 가보겠습니다. 이 무렵에는 데이터에 액세스하는 사용자가 더 많아졌고 쿼리 패턴이 파티션을 제대로 사용하지 않는다는 것을 깨달았습니다. 또는 문자열 열을 기반으로 데이터에 액세스하고 싶었지만 BigQuery에서는 문자열 열을 기준으로 분할하는 것이 불가능합니다. 이로 인해 대량의 데이터를 스캔하고 일일 할당량에 자주 도달하게 되었습니다.

이로 인해 문자열 열을 기준으로 데이터를 분할하는 방법을 고려하게 되었습니다. 우리 데이터 엔지니어는 추가 모듈로 작업과 함께 FarmHash를 사용하여 해당 문자열 열을 정수로 변환할 것을 제안했습니다. 개념 증명에서 이로 인해 스캔이 거의 90% 감소하고 쿼리 성능이 3~5배 향상되었습니다. 우리는 이것을 최종 해결책으로 진행하기로 결정했습니다. 우리에게 필요한 것은 다음과 같습니다.

1. Farmhash 지문으로 테이블 만들기
2. 지문 계산을 위한 파이프라인 변경
3. 데이터를 업로드하세요.

Python에서 FarmHash 지문을 계산하려면 pyfarmhash 모듈이 있습니다. 모듈을 설치하고 아래 코드를 사용하여 해시를 계산했는데 로컬에서는 모두 원하는 대로 작동했습니다.

def get_hash(val: str) -> int:
    return additonal_logic(pyfarmhash.fingerprint64(...))

df[‘hash’] = df[‘Col’].apply(get_hash)

모든 테스트를 통과했으므로 이제 코드를 Airflow에 푸시하고 실행할 차례입니다. 나는 이 단계에서 어떤 일도 잘못될 것이라고 예상하지 못했습니다. 실제로 모든 것이 계획대로, 예상 시간 내에 진행되어 기뻤습니다.

기쁜 마음과 자신감으로 변경 사항을 적용하고 작업을 시작한 다음 10~15분 정도 기다려서 작업이 완료되었습니다. 그러는 동안 나는 다른 작업으로 전환했습니다. 얼마 지나지 않아 Airflow로부터 예상치 못한 실패 이메일을 받았습니다. 로그를 보다가 pyfarmhash 모듈을 설치하다가 실패했다는 것을 보고 깜짝 놀랐습니다!

문제의 이해를 돕기 위해 업무 구조를 설명해야 합니다. 작업에는 다음 단계가 있습니다.

1. 마루 형식의 데이터 다운로드
2. GCS 버킷에 업로드
3. 기존 데이터를 삭제합니다. 있다면. (중복 데이터 방지)
4. BQ 테이블에 데이터를 업로드하세요.

이 과정에서 데이터를 다운로드하는 task-1은 별도의 Python 모듈입니다. 이를 실행하기 위해 Airflow의 PythonVirtualenvOperator를 사용했습니다. 이 연산자를 사용하면 패키지를 요구 사항으로 지정한 다음 새로 생성된 가상 환경에 설치할 수 있습니다. 패키지가 설치되면 해당 패키지의 모든 종속 항목도 설치되어 실행 준비가 됩니다.

데이터를 다운로드하는 모듈에 pyfarmhash를 종속성으로 추가했고 나머지는 변경되지 않았습니다. 그리고 그것은 실패했습니다! 왜요?

pyfarmhash는 C/C++로 구현된 해싱 라이브러리입니다. 설치 시 패키지를 컴파일하려면 GCC가 필요하지만 Airflow 호스트에는 존재하지 않습니다. Airflow 호스트에 GCC가 없는 것이 이해가 되었지만 불행하게도 이것이 저에게 방해가 되었습니다.

pyfarmhash 패키지의 순수 Python 구현을 찾았지만 아무것도 없었습니다. 그러다가 휠 패키지를 찾아보았지만 역시 아무것도 없었습니다. 휠 패키지를 만들어서 밀어넣는 것도 고려했는데, 그러면 장기적으로 휠 패키지를 내부적으로 제공해야 하는 책임이 생겼을 것입니다. 추가적인 해결 방법과 유사한 단계를 피하고 싶었습니다. 모든 옵션을 살펴보고 Airflow를 유지 관리하는 팀과 논의했습니다. 그들은 Docker 이미지를 생성하고 이를 KubernetesPodOperator에서 실행할 것을 제안했습니다. 외부 환경에 의존하지 않고 환경을 제어하고 필요한 모든 것을 포함할 수 있다는 점에서 이는 좋은 선택이었습니다. 또한 이 솔루션에는 해결 방법이 없었습니다. 유일한 단기적인 단점은 구현하는 데 시간이 더 필요하다는 점이었습니다.

Docker 기반 솔루션을 시작하기 전에 이미 이 작업에 약 16~20시간을 소비했습니다. Docker 기반 솔루션의 경우 추가로 다음이 필요했습니다.

1. 다운로드 및 삭제 로직을 시작할 수 있는 진입점을 갖도록 Python 패키지를 변경합니다.
2. Docker 패키지를 생성하고 테스트합니다(이것이 제 두 번째 Docker 이미지였습니다).

더 이상 Airflow에서 PythonVirtualEnvOperator를 사용하지 않을 것이기 때문에 이를 완전히 제거하고 워크플로도 개선하기로 결정했습니다. 다운로드 및 제거 논리를 시작하기 위한 진입점을 갖도록 Python 패키지를 변경해야 했습니다

Docker 이미지가 준비된 최종 솔루션을 갖추는 데 추가로 30~36시간이 걸렸습니다. 이는 영업일 기준 6~7일이 소요되었으며, 초기 2일을 포함하면 긴 스프린트 작업이 되었습니다.

돌이켜 보면 작업 솔루션을 버리고, 모듈 구조를 변경하고, Docker 이미지를 만들고, 작업에 Docker 이미지를 사용하도록 10개 이상의 AirFlow 작업을 변경하고, 이러한 현실을 처리하고 초기 좌절감을 극복해야 했다는 사실이 궁금합니다. 이 모든 것은 "단일 Python 모듈을 컴파일하려면 "gcc"가 필요하기 때문입니다!"

위 내용은 코딩 작업 추정: 무엇이 잘못될 수 있나요?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!