def METHOD_NAME(
    self,
    command_type: t.Optional[commands.CommandType],
    command_payload_file_path: t.Optional[str],
    ddl_concurrent_tasks: t.Optional[int],
) -> SparkSubmitHook:
    application_args = {
        "dialect": "spark",
        "command_type": command_type.value if command_type else None,
        "ddl_concurrent_tasks": ddl_concurrent_tasks,
        "payload_path": command_payload_file_path.split("/")[-1]
        if command_payload_file_path
        else None,
    }
    return SparkSubmitHook(
        conf=self._spark_conf,
        conn_id=self._connection_id,
        total_executor_cores=self._total_executor_cores,
        executor_cores=self._executor_cores,
        executor_memory=self._executor_memory,
        driver_memory=self._driver_memory,
        keytab=self._keytab,
        principal=self._principal,
        proxy_user=self._proxy_user,
        name=self._application_name,
        num_executors=self._num_executors,
        application_args=[f"--{k}={v}" for k, v in application_args.items() if v is not None],
        files=command_payload_file_path,
    )