View a markdown version of this page

PySpark 분석 템플릿에서 파라미터 작업 - AWS Clean Rooms
파라미터 액세스파라미터 보안 모범 사례전체 파라미터 예제

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

PySpark 분석 템플릿에서 파라미터 작업

파라미터는 작업 제출 시 다양한 값을 제공할 수 있도록 하여 PySpark 분석 템플릿의 유연성을 높입니다. 파라미터는 진입점 함수에 전달된 컨텍스트 객체를 통해 액세스할 수 있습니다.

참고

파라미터는 임의의 콘텐츠를 포함할 수 있는 사용자 제공 문자열입니다.

  • 코드를 검토하여 분석에서 예상치 못한 동작을 방지하기 위해 파라미터가 안전하게 처리되는지 확인합니다.

  • 제출 시 제공되는 파라미터 값에 관계없이 안전하게 작동하도록 파라미터 처리를 설계합니다.

파라미터 액세스

파라미터는 context['analysisParameters'] 사전에서 사용할 수 있습니다. 모든 파라미터 값은 문자열입니다.

예파라미터에 안전하게 액세스
def entrypoint(context):
    # Access parameters from context
    parameters = context['analysisParameters']
    threshold = parameters['threshold']
    table_name = parameters['table_name']
    
    # Continue with analysis using parameters
    spark = context['sparkSession']
    input_df = context['referencedTables'][table_name]
    
    # Convert threshold value
    threshold_val = int(threshold)
    
    # Use parameter in DataFrame operation
    filtered_df = input_df.filter(input_df.amount > threshold_val)
    
    return {
        "results": {
            "output": filtered_df
        }
    }

파라미터 보안 모범 사례

주의

파라미터는 임의의 콘텐츠를 포함할 수 있는 사용자 제공 문자열입니다. 분석 코드의 보안 취약성을 방지하려면 파라미터를 안전하게 처리해야 합니다.

피해야 할 안전하지 않은 파라미터 처리 패턴:

  • 파라미터를 코드로 실행 - 파라미터 값에 eval() 또는 exec()를 사용하지 않음

    # UNSAFE - Don't do this
eval(parameters['expression'])  # Can execute arbitrary code

  • SQL 문자열 보간 - 파라미터를 SQL 문자열에 직접 연결하지 않음

    # UNSAFE - Don't do this
sql = f"SELECT * FROM table WHERE column = '{parameters['value']}'"  # SQL injection risk

  • 안전하지 않은 파일 경로 작업 - 검증 없이 파일 시스템 작업에서 파라미터를 직접 사용하지 마세요.

    # UNSAFE - Don't do this
file_path = f"/data/{parameters['filename']}"  # Path traversal risk

안전한 파라미터 처리 패턴:

  • DataFrame 작업에서 파라미터 사용 - Spark DataFrames는 파라미터 값을 안전하게 처리합니다.

    # SAFE - Use parameters in DataFrame operations
threshold = int(parameters['threshold'])
filtered_df = input_df.filter(input_df.value > threshold)

  • 파라미터 값 검증 - 사용하기 전에 파라미터가 예상 형식을 충족하는지 확인합니다.

    # SAFE - Validate parameters before use
def validate_date(date_str):
    try:
        from datetime import datetime
        datetime.strptime(date_str, '%Y-%m-%d')
        return True
    except ValueError:
        return False

date_param = parameters['date_filter'] or '2024-01-01'
if not validate_date(date_param):
    raise ValueError(f"Invalid date format: {date_param}")

  • 파라미터 값에 허용 목록 사용 - 가능하면 알려진 정상 값과 비교하여 파라미터를 검증합니다. 

    # SAFE - Use allowlists
allowed_columns = ['column1', 'column2', 'column3']
column_param = parameters['column_name']
if column_param not in allowed_columns:
    raise ValueError(f"Invalid column: {column_param}")

  • 오류 처리가 포함된 유형 변환 - 문자열 파라미터를 예상 유형으로 안전하게 변환

    # SAFE - Convert with error handling
try:
    batch_size = int(parameters['batch_size'] or '1000')
    if batch_size <= 0 or batch_size > 10000:
        raise ValueError(f"Batch size must be between 1 and 10000")
except ValueError as e:
    print(f"Invalid parameter: {e}")
    raise
중요

파라미터는 작업 실행기가 서로 다른 값을 제공할 때 코드 검토를 우회합니다. 제공된 파라미터 값에 관계없이 안전하게 작동하도록 파라미터 처리를 설계합니다.

전체 파라미터 예제

예 PySpark 스크립트에서 안전하게 파라미터 사용
def entrypoint(context):
    try:
        # Access Spark session and tables
        spark = context['sparkSession']
        input_table = context['referencedTables']['sales_data']
        
        # Access parameters - fail fast if analysisParameters missing
        parameters = context['analysisParameters']
        
        # Validate and convert numeric parameter (handles empty strings with default)
        try:
            threshold = int(parameters['threshold'] or '100')
            if threshold <= 0:
                raise ValueError("Threshold must be positive")
        except (ValueError, TypeError) as e:
            print(f"Invalid threshold parameter: {e}")
            raise
        
        # Validate date parameter (handles empty strings with default)
        date_filter = parameters['start_date'] or '2024-01-01'
        from datetime import datetime
        try:
            datetime.strptime(date_filter, '%Y-%m-%d')
        except ValueError:
            raise ValueError(f"Invalid date format: {date_filter}")
        
        # Use parameters safely in DataFrame operations
        filtered_df = input_table.filter(
            (input_table.amount > threshold) &
            (input_table.date >= date_filter)
        )
        
        result_df = filtered_df.groupBy("category").agg(
            {"amount": "sum"}
        )
        
        return {
            "results": {
                "filtered_results": result_df
            }
        }
    
    except Exception as e:
        print(f"Error in analysis: {str(e)}")
        raise