Python 변환의 오류, 중단점과 로그

실행 중의 값을 확인하려면 IDE의 디버거에서 중단점을 설정합니다. 변수와 DataFrame, 함수 호출 문맥을 볼 수 있습니다. Code Repositories의 Python 디버거와 로컬 VS Code 확장의 실행 환경은 각 도구의 지원 범위를 따릅니다.

Traceback 읽기

처리하지 않은 Python 예외는 함수 호출 경로인 traceback으로 나타납니다. 가장 최근 호출이 아래쪽에 있고 마지막 줄에 예외 유형과 메시지가 있습니다.

NameError라면 사용한 이름이 정의되어 있는지 확인합니다. 예를 들어 Stats 클래스를 정의하고 Statistics()를 호출하면 없는 이름에 대한 오류가 납니다. 마지막 메시지, 아래쪽 파일 및 줄 번호, 해당 코드와 그 위의 호출자를 순서대로 확인합니다.

다음은 값을 추가한 뒤 평균을 계산하는 이름이 일치한 예입니다.

class Stats:
    def __init__(self):
        self.nums = []

    def add(self, number):
        self.nums.append(number)

    def sum(self):
        return sum(self.nums)

    def avg(self):
        return self.sum() / len(self.nums)


def main():
    stats = Stats()
    stats.add(1)
    stats.add(2)
    stats.add(3)
    print(stats.avg())


if __name__ == "__main__":
    main()

이 예제의 avg는 값을 하나 이상 추가한 상태에서 사용합니다. 예외 메시지를 지우는 대신 실제로 호출한 이름과 정의 및 입력 조건을 맞춰야 합니다.

로그의 저장 범위

표준 단일 노드 변환은 print를 사용할 수 있습니다. 운영 Spark 변환의 로그에는 Python logging을 사용하며, 이 경로는 INFO 이상을 저장합니다. 일반적인 DEBUG 메시지가 저장될 것으로 기대하지 않습니다.

VS Code의 Output 또는 Builds의 Actions > View logs에서 로그를 봅니다. Code Repositories의 미리보기 디버거 콘솔과 운영 빌드의 로그 수집 범위도 구분합니다.

다음 예제는 실제 입력의 행 및 열 수를 기록하고 같은 데이터를 출력에 씁니다. 가상 자산 경로를 실제 값으로 바꾸고 사용하는 엔진을 환경에 준비합니다.

Polars

import logging
from transforms.api import Input, Output, transform

LOG = logging.getLogger(__name__)


@transform.using(
    output=Output("/Example Project/Outputs/logged_polars"),
    source=Input("/Example Project/Inputs/table"),
)
def compute(output, source):
    rows = source.polars()
    LOG.info("Number of rows: %d", rows.height)
    print("Number of columns: " + str(rows.width))
    output.write_table(rows)

DuckDB

이 예제는 SQL 결과를 Pandas로 구체화하여 크기를 기록합니다. 큰 데이터를 로그 때문에 불필요하게 수집하지 않도록 실제 사용 범위를 선택합니다.

import logging
from transforms.api import Input, Output, transform

LOG = logging.getLogger(__name__)


@transform.using(
    output=Output("/Example Project/Outputs/logged_duckdb"),
    source=Input("/Example Project/Inputs/table"),
)
def compute(ctx, output, source):
    rows = ctx.duckdb().conn.sql("SELECT * FROM source").fetchdf()
    LOG.info("Number of rows: %d", rows.shape[0])
    print("Number of columns: " + str(rows.shape[1]))
    output.write_table(rows)

Pandas

import logging
from transforms.api import Input, Output, transform

LOG = logging.getLogger(__name__)


@transform.using(
    output=Output("/Example Project/Outputs/logged_pandas"),
    source=Input("/Example Project/Inputs/table"),
)
def compute(output, source):
    rows = source.pandas()
    LOG.info("Number of rows: %d", rows.shape[0])
    print("Number of columns: " + str(rows.shape[1]))
    output.write_table(rows)

PySpark

import logging
from transforms.api import Input, Output, transform_df

LOG = logging.getLogger(__name__)


@transform_df(
    Output("/Example Project/Outputs/logged_spark"),
    source=Input("/Example Project/Inputs/table"),
)
def compute(source):
    LOG.info("Number of rows: %d", source.count())
    LOG.info("Number of columns: %d", len(source.columns))
    return source

Spark의 count는 실제 계산을 실행하는 action입니다. 로그를 위한 추가 전체 스캔 비용을 고려하며 운영 코드를 디버깅 로그 때문에 불필요하게 느리게 만들지 않습니다.

자원 문제 진단

메모리 부족이나 느린 단일 노드 빌드는 job의 Telemetry > Metrics에서 CPU와 메모리 그래프를 봅니다. 메모리에 묶인 작업인지, CPU 계산이 많은지, 할당량을 쓰지 못하는지 구분합니다.

이 Metrics 경로는 lightweight 엔진의 지표입니다. Spark에는 Spark 실행 진단을 적용합니다. 자세한 단일 노드 해석은 Metrics를 따릅니다.