특성 저장소 특성 프로세서 SDK

@feature_processor 데코레이터로 변환 함수를 장식하여 특성 저장소 특성 프로세서 정의를 선언합니다. SageMaker AI SDK for Python(Boto3)은 구성된 입력 데이터 소스에서 데이터를 자동으로 로드하고, 데코레이션된 변환 함수를 적용한 다음 변환된 데이터를 대상 특성 그룹에 수집합니다. 데코레이트된 변환 함수는 @feature_processor데코레이터의 예상 시그니처를 준수해야 합니다. @feature_processor 데코레이터에 대한 자세한 내용은 Amazon SageMaker 특성 저장소 설명서 읽기의 @feature_processor 데코레이터를 참조하세요.

@feature_processor 데코레이터를 사용하면 함수에 제공된 입력 인수와 반환 값이 Spark DataFrames인 Spark 런타임 환경에서 변환 함수가 실행됩니다. 변환 함수의 입력 파라미터 개수는 @feature_processor데코레이터에 구성된 입력 개수와 일치해야 합니다.

@feature_processor 데코레이터에 대한 자세한 내용은 Python용 특성 프로세서 특성 저장소 SDK (Boto3)를 참조하세요.

다음 코드는 @feature_processor데코레이터 사용 방법에 대한 기본 예제입니다. 보다 구체적인 예제 사용 사례는 일반 사용 사례에 대한 특성 처리 코드 예시섹션을 참조하세요.

특성 프로세서 SDK는 다음 명령을 사용하여 SageMaker Python SDK 및 기타 SDK에서 설치할 수 있습니다.

pip install sagemaker[feature-processor]

다음 예제에서 us-east-1은 리소스의 리전, 111122223333은 리소스 소유자 계정 ID, your-feature-group-name은 특성 그룹 이름입니다.

다음은 기본 특성 프로세서 정의로, @feature_processor데코레이터가 Amazon S3의 CSV 입력을 로드하여 변환 함수(예: transform)에 제공하도록 구성하고 특성 그룹에 수집할 수 있도록 준비합니다. 마지막 줄에서 실행합니다.

from sagemaker.feature_store.feature_processor import CSVDataSource, feature_processor

CSV_DATA_SOURCE = CSVDataSource('s3://your-bucket/prefix-to-csv/')
OUTPUT_FG = 'arn:aws:sagemaker:us-east-1:111122223333:feature-group/your-feature-group-name'

@feature_processor(inputs=[CSV_DATA_SOURCE], output=OUTPUT_FG)
def transform(csv_input_df):
   return csv_input_df
   
transform()

@feature_processor 파라미터는 다음과 같습니다.

선택적으로 래핑된 함수 파라미터(함수 시그니처에 제공된 경우 인수로 제공됨)는 다음과 같습니다.

다음 코드는 params및 spark파라미터를 사용하는 예제입니다.

from sagemaker.feature_store.feature_processor import CSVDataSource, feature_processor

CSV_DATA_SOURCE = CSVDataSource('s3://your-bucket/prefix-to-csv/')
OUTPUT_FG = 'arn:aws:sagemaker:us-east-1:111122223333:feature-group/your-feature-group-name' 

@feature_processor(inputs=[CSV_DATA_SOURCE], output=OUTPUT_FG)
def transform(csv_input_df, params, spark):
   
   scheduled_time = params['system']['scheduled_time']
   csv_input_df.createOrReplaceTempView('csv_input_df')
   return spark.sql(f'''
        SELECT *
        FROM csv_input_df
        WHERE date_add(event_time, 1) >= {scheduled_time}
   ''')
   
transform()

scheduled_time 시스템 파라미터(함수의 params인수에 제공됨)는 각 실행의 재시도를 지원하는 중요한 값입니다. 이 값은 특성 프로세서의 실행을 고유하게 식별하는 데 도움이 될 수 있으며, 코드의 실제 실행 시간과 관계없이 입력 범위를 보장하기 위해 날짜 범위 기반 입력(예: 최근 24시간 분량의 데이터만 로드)의 참조 포인트로 사용할 수 있습니다. 특성 프로세서가 일정에 따라 실행되는 경우(특성 프로세서 파이프라인의 일정 예약 및 이벤트 기반 실행 참조) 해당 값은 예정된 실행 시간으로 고정됩니다. 데이터 백필 또는 누락된 이전 실행 재실행과 같은 사용 사례를 지원하기 위해 SDK의 Execute API를 사용하여 동기 실행 중에 인수를 재정의할 수 있습니다. 이 값은 특성 프로세서가 다른 방식으로 실행되는 경우의 현재 시간입니다.

Spark 코드 작성에 대한 자세한 내용은 Spark SQL 프로그래밍 설명서를 참조하세요.

일반 사용 사례에 대한 더 많은 코드 샘플은 일반 사용 사례에 대한 특성 처리 코드 예시섹션을 참조하세요.

@feature_processor로 데코레이션된 변환 함수는 값을 반환하지 않는다는 점에 유의하세요. 함수를 프로그래밍 방식으로 테스트하려면 래핑된 함수에 대한 패스스루 역할을 하도록 @feature_processor데코레이터를 제거하거나 몽키 패치하면 됩니다. @feature_processor 데코레이터에 대한 자세한 내용은 Amazon SageMaker 특성 저장소 Python SDK를 참조하세요.