View a markdown version of this page

Trabajar con parámetros en plantillas PySpark de análisis - AWS Clean Rooms
Acceder a los parámetrosPrácticas recomendadas de seguridad de parámetrosEjemplo completo de parámetros

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Trabajar con parámetros en plantillas PySpark de análisis

Los parámetros aumentan la flexibilidad de sus plantillas de PySpark análisis al permitir que se proporcionen diferentes valores en el momento de enviar los trabajos. Se puede acceder a los parámetros a través del objeto de contexto que se pasa a la función de punto de entrada.

nota

Los parámetros son cadenas proporcionadas por el usuario que pueden contener contenido arbitrario.

  • Revise el código para asegurarse de que los parámetros se gestionen de forma segura y evitar comportamientos inesperados en el análisis.

  • Diseñe el manejo de los parámetros para que funcione de forma segura, independientemente de los valores de los parámetros que se proporcionen en el momento del envío.

Acceder a los parámetros

Los parámetros están disponibles en el context['analysisParameters'] diccionario. Todos los valores de los parámetros son cadenas.

ejemplo Acceso seguro a los parámetros
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
        }
    }

Prácticas recomendadas de seguridad de parámetros

aviso

Los parámetros son cadenas proporcionadas por el usuario que pueden contener contenido arbitrario. Debe gestionar los parámetros de forma segura para evitar vulnerabilidades de seguridad en el código de análisis.

Patrones de manejo de parámetros inseguros para evitar:

  • Ejecutar parámetros como código: nunca utilice eval() ni exec() sobre los valores de los parámetros

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

  • Interpolación de cadenas SQL: nunca concatene los parámetros directamente en cadenas SQL

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

  • Operaciones de rutas de archivos no seguras: nunca utilice los parámetros directamente en las operaciones del sistema de archivos sin validarlos

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

Patrones de manejo seguro de parámetros:

  • Utilice los parámetros en DataFrame las operaciones: DataFrames maneje los valores de los parámetros de forma segura

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

  • Valide los valores de los parámetros: compruebe que los parámetros cumplen con los formatos esperados antes de usarlos

    # 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}")

  • Utilice listas de valores permitidos para los valores de los parámetros: siempre que sea posible, valide los parámetros comparándolos con valores válidos conocidos

    # 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}")

  • Conversión de tipos con gestión de errores: convierta los parámetros de cadena en los tipos esperados de forma segura

    # 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
importante

Recuerde que los parámetros omiten la revisión del código cuando los ejecutores de tareas proporcionan valores diferentes. Diseñe el manejo de los parámetros para que funcione de forma segura, independientemente de los valores de los parámetros que se proporcionen.

Ejemplo completo de parámetros

ejemplo Utilizar los parámetros de forma segura en un PySpark script
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