def METHOD_NAME(self, spark: SparkSession, args: Dict[str, Any]) -> None:
    logger: Logger = self.get_logger(spark=spark)
    # Arguments
    input_location: str = args[constants.GCS_TO_GCS_INPUT_LOCATION]
    input_format: str = args[constants.GCS_TO_GCS_INPUT_FORMAT]
    gcs_temp_view: str = args[constants.GCS_TO_GCS_TEMP_VIEW_NAME]
    sql_query: str = args[constants.GCS_TO_GCS_SQL_QUERY]
    output_partition_column: str = args[constants.GCS_TO_GCS_OUTPUT_PARTITION_COLUMN]
    output_mode: str = args[constants.GCS_TO_GCS_OUTPUT_MODE]
    output_format: str = args[constants.GCS_TO_GCS_OUTPUT_FORMAT]
    output_location: str = args[constants.GCS_TO_GCS_OUTPUT_LOCATION]
    logger.info(
        "Starting Cloud Storage to Cloud Storage with tranformations Spark job with parameters:\n"
        f"{pprint.pformat(args)}"
    )
    # Read
    input_data = ingest_dataframe_from_cloud_storage(
        spark,
        args,
        input_location,
        input_format,
        "gcs.gcs.input.",
        avro_format_override=constants.FORMAT_AVRO
    )
    if sql_query:
        # Create temp view on source data
        input_data.createOrReplaceTempView(gcs_temp_view)
        # Execute SQL
        output_data = spark.sql(sql_query)
    else:
        output_data = input_data
    # Write
    if output_partition_column:
        writer: DataFrameWriter = output_data.write.mode(output_mode).partitionBy(output_partition_column)
    else:
        writer: DataFrameWriter = output_data.write.mode(output_mode)
    persist_dataframe_to_cloud_storage(writer, args, output_location, output_format, "gcs.gcs.output."