Repository: PrefectHQ/prefect
Pull request: Added worker for Snowflake's Snowpark Container Services
This PR adds a new worker to the prefect-snowflake integration. The worker can run service jobs in Snowflake's Snowpark Container Services. Closes #15674 

A few things worth mentioning:

- ~~Snowflake's Python API currently doesn't support running async service jobs, so I'm temporarily running jobs using SQL commands instead. I will submit a PR to resolve this as soon as Snowflake adds the required features to their Python API.~~ This has been resolved as of 2025-03-31.
- Most of the code was hacked together based on the official "custom worker" tutorial and copying things over from the Azure and Kubernetes workers. The same goes for the unit tests.
- Since a lot of the structure was copied over, let me know if I misunderstood how something should be done. For example, am I using run_sync_in_worker_thread correctly?
- I don't have a lot of experience creating unit tests in Python, so I've done the bare minimum with regards to unit testing.

I don't know the best way to test the worker after moving the modules over from our company repo to the Prefect repo. Perhaps someone could test it for me, or leave instructions in a comment. The code should be the same as it were before, it's just some import paths that have changed.

There are two bugs I've bumped into along the way which I don't think are caused by my code. I don't know if these issues only affect custom workers, but I think so:

- After creating a custom worker pool and changing any of the settings, you can no longer edit the pool using the GUI. See #15678 
- If I make changes to some of the settings in the pool, such as the Snowflake credentials, it seems like they don't apply until the pool is re-created.

Problem statement:
New worker for running job services in Snowpark Container Services
### Describe the current behavior

There is no worker suitable for running flows in Snowpark Container Services.

### Describe the proposed behavior

A new worker is added, capable of running flows as job services in Snowpark Container Services.

I've already developed a worker that does this. It isn't ready for release yet, might take a couple of weeks, When it's ready, I'm wondering if it'd make more sense to release it as a PR into prefect-snowflake.

It would be great to hear what someone from the Prefect team thinks. If you're already planning to add this, perhaps my contribution can be a starting point. If not, I'll develop and maintain this on my own, as a 3rd-party package.

### Example Use

Users could run their flows in Snowpark Container Services as an alternative to Kubernetes, Azure Container Instances etc.

### Additional context

Snowpark Container Services seems to be a Kubernetes-based, Snowflake-managed job service, capable of handling both long-running "services" (with auto restarts etc) as well as short, one-off "job services". Job services seems like the best fit for a Prefect integration.

Hints from issue discussion:
Hey @bjorhn! It's super cool to hear that you're developing a worker for Snowpark Container Services! I'm definitely open to exploring adding your implementation to `prefect-snowflake`. When you're ready, feel free to open a PR to `prefect-snowflake` and we can give it a look!

Also, if you run into any issues while developing the worker, feel free to comment on this issue and we can help!
The worker is done, since yesterday it's running all of our flow runs in our dev and test environments. If it stays stable, we will switch over to it for production work as well, hopefully in about a week. When we are confident it's running well in production, I'll create a PR to gather some feedback.
Hey @bjorhn ! Is your worker implementation in a state that you'd be comfortable sharing? Another group is building out a new worker type soon and I'd love to use your worker as another reference example.
I had to implement a temporary workaround for a problem I've since spoken with Snowflake's engineers about, so my plan was to wait until they push an API update that would let me remove the workaround, which is due in January.

I suppose it wouldn't do any harm to get the PR up anyway, it'll be a good opportunity to collect some feedback meanwhile. We have been running the worker in our production environment for 6 weeks now and it's been near flawless. 

I should be able to get something up this weekend. 
It's up!

Commit message: Formatting

Patch to review:
diff --git a/src/integrations/prefect-snowflake/prefect_snowflake/workers/__init__.py b/src/integrations/prefect-snowflake/prefect_snowflake/workers/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/src/integrations/prefect-snowflake/prefect_snowflake/workers/spcs.py b/src/integrations/prefect-snowflake/prefect_snowflake/workers/spcs.py
new file mode 100644
index 000000000000..f88160a8c068
--- /dev/null
+++ b/src/integrations/prefect-snowflake/prefect_snowflake/workers/spcs.py
@@ -0,0 +1,621 @@
+"""Classes and helper functions for creating a worker that can run job services in Snowpark Container Services."""
+
+import datetime
+import os
+import sys
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, Optional
+from uuid import UUID
+
+import anyio
+import dateutil
+import snowflake.connector
+import yaml
+from pydantic import Field
+from slugify import slugify
+from snowflake.connector import SnowflakeConnection
+from snowflake.core import Root
+from snowflake.core.exceptions import NotFoundError
+from snowflake.core.service import ServiceResource
+
+from prefect.client.schemas.objects import FlowRun
+from prefect.utilities.asyncutils import run_sync_in_worker_thread
+from prefect.utilities.dockerutils import get_prefect_image_name
+from prefect.workers.base import (
+    BaseJobConfiguration,
+    BaseVariables,
+    BaseWorker,
+    BaseWorkerResult,
+)
+from prefect_snowflake import SnowflakeCredentials
+
+if TYPE_CHECKING:
+    from prefect.server.schemas.core import Flow
+    from prefect.server.schemas.responses import DeploymentResponse
+
+SPCS_DEFAULT_CPU_REQUEST = "1"
+SPCS_DEFAULT_MEMORY_REQUEST = "1G"
+SPCS_DEFAULT_CPU_LIMIT = None
+SPCS_DEFAULT_MEMORY_LIMIT = None
+
+DEFAULT_CONTAINER_ENTRYPOINT = "/opt/prefect/entrypoint.sh"
+
+
+def _get_default_job_manifest_template() -> dict[str, Any]:
+    """Returns the default job manifest template used by the SPCS job service."""
+    # fmt: off
+    return {
+        "spec": {
+            "containers": [
+                {
+                    "name": "{{ name }}",
+                    "image": "{{ image }}",
+                    "command": "{{ entrypoint }}",
+                    "args": " {{ command }}",
+                    "env": "{{ env }}",
+                    "secrets": "{{ secrets }}",
+                    "volumeMounts": "{{ volume_mounts }}",
+                    "resources": {
+                        "requests": {
+                            "cpu": "{{ cpu_request }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_request }}"
+                        },
+                        "limits": {
+                            "cpu": "{{ cpu_limit }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_limit }}"
+                        },
+                    },
+                }
+            ],
+            "volumes": "{{ volumes }}",
+            "logExporters": {
+                "eventTableConfig": {
+                    "logLevel": "{{ log_level }}"
+                }
+            },
+            "platformMonitor":
+            {
+                "metricConfig": {
+                    "groups": "{{ metrics_groups }}"
+                }
+            }
+        }
+    }
+    # fmt: on
+
+
+class SPCSWorkerConfiguration(BaseJobConfiguration):
+    """Defines variables which can be overridden in the work pool.
+
+    Also handles validation of the config, whether set in the work pool or in the deployment.
+    """
+
+    job_manifest: dict[str, Any] = Field(
+        json_schema_extra={"template": _get_default_job_manifest_template()}
+    )
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
+        description="The fully-qualified name of the compute pool to run services in.",
+        examples=["common.compute.my_compute_pool"],
+    )
+    cpu_request: str = Field(
+        default=SPCS_DEFAULT_CPU_REQUEST,
+        description=f"CPU allocation request for the job service. If not provided, a default value of {SPCS_DEFAULT_CPU_REQUEST} will be used.",
+    )
+    cpu_limit: str = Field(
+        default=SPCS_DEFAULT_CPU_LIMIT,
+        description="CPU allocation limit for the job service. If not provided, there will be no limit.",
+    )
+    gpu_count: Optional[int] = Field(
+        default=None,
+        description="The number of GPUs to use. If not provided, no GPUs will be used.",
+    )
+    memory_request: str = Field(
+        default=SPCS_DEFAULT_MEMORY_REQUEST,
+        description=(
+            "Memory allocation request for the job service."
+            f"If not provided, a default value of {SPCS_DEFAULT_MEMORY_REQUEST} will be used unless present on the task definition."
+        ),
+    )
+    memory_limit: str = Field(
+        default=SPCS_DEFAULT_MEMORY_LIMIT,
+        description="Memory allocation limit for the job service. If not provided, will be the same as the request.",
+    )
+    volume_mounts: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Defines where volumes appears in the container.",
+    )
+    volumes: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Shared file system that can be made available in your container.",
+    )
+    log_level: Literal["INFO", "ERROR", "NONE"] = Field(
+        default="INFO",
+        description="The level of application logs that Snowflake should collect and export to your event table.",
+    )
+    metrics_groups: list[str] = Field(
+        default_factory=list,
+        description="The list of predefined metrics groups that should be logged to the event table.",
+    )
+    query_warehouse: Optional[str] = Field(
+        default=None, description="The query warehouse of the job service."
+    )
+    service_comment: Optional[str] = Field(
+        default=None,
+        description="Specify a comment for the job service. Visible in certain Snowflake logs.",
+    )
+    stream_output: bool = Field(
+        default=False,
+        description="Direct flow log output back to the worker's console.",
+    )
+    service_watch_poll_interval: int = Field(
+        default=5,
+        description="The number of seconds to wait between Snowflake API calls while monitoring the state of the service.",
+    )
+
+    def prepare_for_flow_run(
+        self,
+        flow_run: "FlowRun",
+        deployment: Optional["DeploymentResponse"] = None,
+        flow: Optional["Flow"] = None,
+    ) -> None:
+        """Prepares the job configuration for a flow run."""
+        super().prepare_for_flow_run(flow_run, deployment, flow)
+
+        # There's only one container.
+        container = self.job_manifest["spec"]["containers"][0]
+
+        # Ensure an image is set.
+        container["image"] = self.image
+
+        # Set the container's environment variables by grabbing them from the worker.
+        container["env"] = {**self._base_environment(), **self.env}
+
+        # Convert the entrypoint from a string to a list, because that's what SPCS expects
+        if self.entrypoint:
+            container["command"] = self.entrypoint.split(" ")
+
+        # Convert the command from a string to a list, because that's what SPCS expects
+        if self.command:
+            container["args"] = self.command.split(" ")
+
+        # Set memory_limit to the same value as memory_request, if not specified.
+        if not self.memory_limit:
+            container["resources"]["limits"]["memory"] = self.memory_request
+
+        # Remove the platformMonitor section if there are no metrics groups.
+        if not self.metrics_groups:
+            self.job_manifest["spec"].pop("platformMonitor")
+
+
+class SPCSServiceTemplateVariables(BaseVariables):
+    """Defines variables which can be overridden by deployments.
+
+    Must include all variables in SPCSWorkerConfiguration. All validation should happen in SPCSWorkerConfiguration.
+    """
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
+        description="The fully-qualified name of the compute pool to run services in.",
+        examples=["common.compute.my_compute_pool"],
+    )
+    cpu_request: str = Field(
+        default=SPCS_DEFAULT_CPU_REQUEST,
+        description=f"CPU allocation request for the job service. If not provided, a default value of {SPCS_DEFAULT_CPU_REQUEST} will be used.",
+    )
+    cpu_limit: str = Field(
+        default=SPCS_DEFAULT_CPU_LIMIT,
+        description="CPU allocation limit for the job service. If not provided, there will be no limit.",
+    )
+    gpu_count: Optional[int] = Field(
+        default=None,
+        description="The number of GPUs to use. If not provided, no GPUs will be used.",
+    )
+    memory_request: str = Field(
+        default=SPCS_DEFAULT_MEMORY_REQUEST,
+        description=(
+            "Memory allocation request for the job service."
+            f"If not provided, a default value of {SPCS_DEFAULT_MEMORY_REQUEST} will be used unless present on the task definition."
+        ),
+    )
+    memory_limit: str = Field(
+        default=SPCS_DEFAULT_MEMORY_LIMIT,
+        description="Memory allocation limit for the job service. If not provided, will be the same as the request.",
+    )
+    volume_mounts: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Defines where volumes appears in the container.",
+    )
+    volumes: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Shared file system that can be made available in your container.",
+    )
+    log_level: Literal["INFO", "ERROR", "NONE"] = Field(
+        default="INFO",
+        description="The level of application logs that Snowflake should collect and export to your event table.",
+    )
+    metrics_groups: list[str] = Field(
+        default_factory=list,
+        description="The list of predefined metrics groups that should be logged to the event table.",
+    )
+    query_warehouse: Optional[str] = Field(
+        default=None, description="The query warehouse of the job service."
+    )
+    service_comment: Optional[str] = Field(
+        default=None,
+        description="Specify a comment for the job service. Visible in certain Snowflake logs.",
+    )
+    stream_output: bool = Field(
+        default=False,
+        description="Direct flow log output back to the worker's console.",
+    )
+    service_watch_poll_interval: int = Field(
+        default=5,
+        description="The number of seconds to wait between Snowflake API calls while monitoring the state of the service.",
+    )
+
+
+class SPCSWorkerResult(BaseWorkerResult):
+    """Result returned by the SPCSWorker."""
+
+
+class SPCSWorker(BaseWorker):
+    """A Prefect worker that runs flows as service jobs in Snowpark Container Services."""
+
+    type: str = "snowpark-container-service"
+    job_configuration = SPCSWorkerConfiguration
+    job_configuration_variables = SPCSServiceTemplateVariables
+    _description = "Execute flow runs within containers on Snowflake's Snowpark Container Services. Requires a Snowflake account."
+
+    async def run(
+        self,
+        flow_run: FlowRun,
+        configuration: SPCSWorkerConfiguration,
+        task_status: Optional[anyio.abc.TaskStatus] = None,
+    ) -> BaseWorkerResult:
+        """Run a flow as a service job in Snowpark Container Services.
+
+        Args:
+            flow_run: The flow run to run.
+            configuration: The configuration for the flow run.
+            task_status: The task status object for the current task. Used
+            to provide an identifier that can be used to cancel the task.
+
+        Returns:
+            The result of the flow run.
+
+        """
+        run_start_time = datetime.now(datetime.timezone.utc)
+
+        # If SNOWFLAKE_HOST is defined, we're probably running in Snowflake.
+        # That means we can run as the service account.
+        if os.getenv("SNOWFLAKE_HOST"):
+
+            def get_login_token() -> str:
+                with Path("/snowflake/session/token").open("r") as f:
+                    return f.read()
+
+            connection_parameters = {
+                "host": os.getenv("SNOWFLAKE_HOST"),
+                "account": os.getenv("SNOWFLAKE_ACCOUNT"),
+                "token": get_login_token(),
+                "authenticator": "oauth",
+            }
+        else:
+            connection_parameters = {
+                "account": configuration.snowflake_credentials.account,
+                "user": configuration.snowflake_credentials.user,
+                "private_key": configuration.snowflake_credentials.resolve_private_key(),
+                "role": configuration.snowflake_credentials.role,
+            }
+
+        session = snowflake.connector.connect(**connection_parameters)
+        root = Root(session)
+
+        # Create the execution environment and start execution
+        job_service_name = await run_sync_in_worker_thread(
+            self._create_and_start_service,
+            root,
+            session,
+            flow_run,
+            configuration,
+        )
+
+        if task_status:
+            # Use a unique ID to mark the run as started.
+            # This ID is later used to tear down infrastructure, if the flow run is cancelled.
+            task_status.started(job_service_name)
+
+        # Monitor the execution
+        job_status = await run_sync_in_worker_thread(
+            self._watch_service,
+            job_service_name,
+            root,
+            configuration,
+            run_start_time,
+        )
+
+        # We're done in Snowflake, close the connection.
+        session.close()
+
+        exit_code = (
+            job_status if job_status is not None else -1
+        )  # Get the result of the execution for reporting
+
+        return SPCSWorkerResult(
+            status_code=exit_code,
+            identifier=job_service_name,
+        )
+
+    def _create_and_start_service(
+        self,
+        root: Root,
+        session: SnowflakeConnection,
+        flow_run: FlowRun,
+        configuration: SPCSWorkerConfiguration,
+    ) -> str:
+        [database, schema, compute_pool] = configuration.compute_pool.split(".")
+
+        job_manifest_yaml = yaml.dump(configuration.job_manifest)
+        job_service_name = self._slugify_service_name(configuration.name, flow_run.id)
+
+        self._logger.info(
+            f"Starting job service {job_service_name} in compute pool {compute_pool}..."
+        )
+
+        # The Snowflake Python SDK currently doesn't support executing multiple
+        # async statements simultaneously. We need to run SQL commands using
+        # snowflake-connector for now.
+        # Note: This temporary workaround doesn't support setting QUERY_WAREHOUSE or COMMENT.
+        cur = session.cursor()
+
+        sql_cmd = f"""
+        EXECUTE JOB SERVICE
+        IN COMPUTE POOL {compute_pool}
+        NAME = {database}.{schema}.{job_service_name}
+        EXTERNAL_ACCESS_INTEGRATIONS = ({", ".join(configuration.external_access_integrations)})
+        FROM SPECIFICATION $${job_manifest_yaml}$$;
+        """
+
+        cur.execute_async(sql_cmd)
+
+        # Note: Uncomment this when the above workaround is no longer needed.
+        # job_service = JobService(
+        #     name=job_service_name,
+        #     compute_pool=compute_pool,
+        #     spec=ServiceSpec(job_manifest_yaml),
+        #     external_access_integrations=configuration.external_access_integrations,
+        #     query_warehouse=configuration.query_warehouse,
+        #     comment=configuration.service_comment,
+        # )
+
+        # root.databases[database].schemas[schema].services.execute_job(job_service, async_req=True)
+
+        return job_service_name
+
+    def _watch_service(
+        self,
+        job_service_name: str,
+        root: Root,
+        configuration: SPCSWorkerConfiguration,
+        run_start_time: datetime,
+    ) -> int:
+        [database, schema, compute_pool] = configuration.compute_pool.split(".")
+
+        pool = root.compute_pools[compute_pool]
+
+        while True:
+            pool_state = pool.fetch().state
+
+            # Wait until the compute pool is active. It might be idle, resizing, etc.
+            if pool_state == "ACTIVE":
+                break
+
+            self._logger.info(
+                f"Compute pool {compute_pool} is in state {pool_state}, checking for ACTIVE state again in {configuration.service_watch_poll_interval} seconds."
+            )
+            time.sleep(configuration.service_watch_poll_interval)
+
+        service = root.databases[database].schemas[schema].services[job_service_name]
+        last_log_time = run_start_time
+
+        while True:
+            # Sleep first, give the job service a chance to start.
+            time.sleep(configuration.service_watch_poll_interval)
+
+            try:
+                service_status = service.get_service_status()
+                container_status = service_status[0]
+
+                # If status == PENDING or similar, we'll get an exception if we try to retrieve logs.
+                if configuration.stream_output and container_status["status"] in (
+                    "ACTIVE",
+                    "DONE",
+                    "FAILED",
+                ):
+                    last_log_time = self._get_and_stream_output(
+                        service=service,
+                        container_status=container_status,
+                        last_log_time=last_log_time,
+                    )
+
+                # If status in ("DONE", "FAILED"), the container is no longer running.
+                if container_status["status"] in ("DONE", "FAILED"):
+                    break
+            except NotFoundError:
+                self._logger.info(
+                    f"Service {job_service_name} isn't running yet, polling for status again in {configuration.service_watch_poll_interval} seconds."
+                )
+
+        return 0
+
+    @staticmethod
+    def _slugify_service_name(service_name: str, flow_run_id: UUID) -> Optional[str]:
+        """Generates a service name to match the configured name, ensuring it's SPCS compatible."""
+        # Must match `/?[a-zA-Z0-9][a-zA-Z0-9_]+` in the end.
+        if not service_name:
+            return None
+
+        # Convert flow_run_id to a string.
+        flow_run_id = str(flow_run_id)
+
+        # Slugify flow_name and ensure that the generated name won't be too long for the max deployment name length (63).
+        slugified_flow_name = slugify(
+            service_name,
+            separator="_",
+            max_length=62 - len(flow_run_id),
+            regex_pattern=r"[^a-zA-Z0-9_]+",  # Only allow lowercase, uppercase letters, numbers, underscores.
+        ).strip("_")  # Remove leading and trailing underscores.
+
+        return f"{slugified_flow_name}_{flow_run_id.replace('-', '_')}"
+
+    def _get_and_stream_output(
+        self,
+        service: ServiceResource,
+        container_status: dict[str, Any],
+        last_log_time: datetime,
+    ) -> datetime:
+        """Fetches logs output from the job container and writes all entries after a given time to stderr.
+
+        Args:
+            service: The service we want to retrieve logs for.
+            container_status: The status object containing the instanceId and containerName of the container.
+            last_log_time: The timestamp of the last output line already streamed.
+
+        Returns:
+            The time of the most recent output line written by this call.
+
+        """
+        logs = self._get_logs(service=service, container_status=container_status)
+
+        return self._stream_output(logs, last_log_time)
+
+    def _get_logs(
+        self,
+        service: ServiceResource,
+        container_status: dict[str, Any],
+        max_lines: int = 100,
+    ) -> str:
+        """Gets the most container logs up to a given maximum.
+
+        Args:
+            service: The service we want to retrieve logs for.
+            container_status: The status object containing the instanceId and containerName of the container.
+            max_lines: The number of log lines to pull. Defaults to 100.
+
+        Returns:
+            A string containing the requested log entries, one per line.
+
+        """
+        instance_id = container_status["instanceId"]
+        container_name = container_status["containerName"]
+
+        return service.get_service_logs(
+            instance_id=instance_id, container_name=container_name, num_lines=max_lines
+        )
+
+    def _stream_output(self, log_content: str, last_log_time: datetime) -> datetime:
+        """Writes each entry from a string of log lines to stderr.
+
+        Args:
+            log_content: A string containing SPCS container logs.
+            last_log_time: The timestamp of the last output line already streamed.
+
+        Returns:
+            The time of the most recent output line written by this call.
+
+        """
+        if not log_content:
+            # Nothing to stream.
+            return last_log_time
+
+        log_lines = log_content.split("\n")
+
+        last_written_time = last_log_time
+
+        for log_line in log_lines:
+            # Skip if the line is blank or whitespace.
+            if not log_line.strip():
+                continue
+
+            line_parts = log_line.split(" ")
+            # The timestamp should always be before the first space in the line.
+            line_timestamp = line_parts[0]
+
+            try:
+                # Parse the timestamp and make it timezone-aware.
+                line_time = dateutil.parser.parse(line_timestamp).replace(
+                    tzinfo=datetime.timezone.utc
+                )
+
+                if line_time > last_written_time:
+                    print(log_line, file=sys.stderr)
+                    last_written_time = line_time
+            except dateutil.parser.ParserError as e:
+                self._logger.debug(
+                    ("Unable to parse timestamp from SPCS log line: %s"),
+                    log_line,
+                    exc_info=e,
+                )
+
+        return last_written_time
diff --git a/src/integrations/prefect-snowflake/pyproject.toml b/src/integrations/prefect-snowflake/pyproject.toml
index c57d27b28729..a5f7c1e5dcc6 100644
--- a/src/integrations/prefect-snowflake/pyproject.toml
+++ b/src/integrations/prefect-snowflake/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "prefect-snowflake"
-dependencies = ["snowflake-connector-python>=2.7.6", "prefect>=3.0.0"]
+dependencies = ["snowflake>=1.0.2", "snowflake-connector-python>=2.7.6", "prefect>=3.0.0"]
 dynamic = ["version"]
 description = "Prefect integrations for interacting with Snowflake"
 readme = "README.md"
diff --git a/src/integrations/prefect-snowflake/tests/test_spcs_worker.py b/src/integrations/prefect-snowflake/tests/test_spcs_worker.py
new file mode 100644
index 000000000000..0ffa7419cb83
--- /dev/null
+++ b/src/integrations/prefect-snowflake/tests/test_spcs_worker.py
@@ -0,0 +1,197 @@
+import uuid
+from unittest.mock import Mock
+
+import pytest
+from prefect_snowflake import SnowflakeCredentials
+from prefect_snowflake.workers.spcs import (
+    SPCSServiceTemplateVariables,
+    SPCSWorker,
+    SPCSWorkerConfiguration,
+)
+from pydantic import SecretStr
+
+from prefect.client.schemas import FlowRun
+from prefect.server.schemas.core import Flow
+from prefect.testing.utilities import AsyncMock
+from prefect.utilities.dockerutils import get_prefect_image_name
+
+
+# Helper functions
+async def create_job_configuration(
+    snowflake_credentials, worker_flow_run, overrides=None, run_prep=True
+):
+    """
+    Returns a basic initialized SPCS infrastructure block suitable for use in a variety of tests.
+    """
+    if overrides is None:
+        overrides = {}
+
+    values = {
+        "command": "test",
+        "env": {},
+        "snowflake_credentials": snowflake_credentials,
+        "name": None,
+        "service_watch_poll_interval": 1,
+        "stream_output": False,
+    }
+
+    for k, v in overrides.items():
+        values = {**values, k: v}
+
+    job_service_variables = SPCSServiceTemplateVariables(**values)
+
+    json_config = {
+        "job_configuration": SPCSWorkerConfiguration.json_template(),
+        "variables": job_service_variables.model_dump(),
+    }
+
+    job_service_configuration = await SPCSWorkerConfiguration.from_template_and_values(
+        json_config, values
+    )
+
+    if run_prep:
+        job_service_configuration.prepare_for_flow_run(worker_flow_run)
+
+    return job_service_configuration
+
+
+# Fixtures
+@pytest.fixture
+def snowflake_credentials():
+    account = "test_account"
+    user = "test_user"
+    password = "test_password"
+
+    return SnowflakeCredentials(account=account, user=user, password=password)
+
+
+@pytest.fixture
+def mock_prefect_client(worker_flow):
+    """
+    A fixture that provides a mock Prefect client
+    """
+    mock_client = Mock()
+    mock_client.read_flow = AsyncMock()
+    mock_client.read_flow.return_value = worker_flow
+
+    return mock_client
+
+
+@pytest.fixture
+def worker_flow():
+    return Flow(id=uuid.uuid4(), name="test-flow")
+
+
+@pytest.fixture
+def worker_flow_run(worker_flow):
+    return FlowRun(id=uuid.uuid4(), flow_id=worker_flow.id, name="test-flow-run")
+
+
+# Tests
+async def test_worker_valid_command_validation(snowflake_credentials, worker_flow_run):
+    # ensure the validator allows valid commands to pass through
+    command = "command arg1 arg2"
+
+    spcs_job_config = await create_job_configuration(
+        snowflake_credentials, worker_flow_run, {"command": command}
+    )
+
+    assert spcs_job_config.command == command
+
+
+def test_worker_invalid_command_validation(snowflake_credentials):
+    # ensure invalid commands cause a validation error
+    with pytest.raises(ValueError):
+        SPCSWorkerConfiguration(
+            command=["invalid_command", "arg1", "arg2"],
+            subscription_id=SecretStr("test"),
+            resource_group_name="test",
+            snowflake_credentials=snowflake_credentials,
+        )
+
+
+async def test_job_configuration_creation(snowflake_credentials, worker_flow_run):
+    config = {
+        "snowflake_credentials": snowflake_credentials,
+        "image": "my-image",
+        "image_registry": "my_image_registry",
+        "compute_pool": "common.compute.test_pool",
+        "external_access_integrations": ["test_integration"],
+        "cpu_request": "1",
+        "memory_request": "1G",
+        "env": {"TEST": "VALUE"},
+    }
+
+    spcs_job_config = await create_job_configuration(
+        snowflake_credentials, worker_flow_run, config
+    )
+
+    assert spcs_job_config.image == "my-image"
+    assert spcs_job_config.image_registry == "my_image_registry"
+    assert spcs_job_config.cpu_request == "1"
+    assert spcs_job_config.memory_request == "1G"
+    assert spcs_job_config.env.get("TEST") == "VALUE"
+    assert spcs_job_config.compute_pool == "common.compute.test_pool"
+    assert spcs_job_config.external_access_integrations == ["test_integration"]
+    assert (
+        spcs_job_config.snowflake_credentials.account == snowflake_credentials.account
+    )
+    assert spcs_job_config.snowflake_credentials.user == snowflake_credentials.user
+    assert (
+        spcs_job_config.snowflake_credentials.password == snowflake_credentials.password
+    )
+
+
+async def test_image_populated_in_template_when_not_provided(
+    worker_flow_run, snowflake_credentials
+):
+    config = await SPCSWorkerConfiguration.from_template_and_values(
+        base_job_template=SPCSWorker.get_default_base_job_template(),
+        values=SPCSServiceTemplateVariables(
+            snowflake_credentials=snowflake_credentials,
+        ).model_dump(exclude_unset=True),
+    )
+    config.prepare_for_flow_run(worker_flow_run)
+
+    assert config.image == get_prefect_image_name()
+    assert (
+        config.job_manifest["spec"]["containers"][0]["image"]
+        == get_prefect_image_name()
+    )
+
+
+@pytest.mark.parametrize(
+    "flow_name",
+    [
+        "Short Flow",
+        "Another Notification",
+        "Extremely Long Flow Name That Exceeds The Maximum Length Allowed",
+    ],
+)
+async def test_consistent_service_job_naming(
+    mock_prefect_client: mock_prefect_client,
+    worker_flow_run,
+    flow_name,
+):
+    max_length = 63
+
+    flow = Flow(id=worker_flow_run.flow_id, name=flow_name)
+    mock_prefect_client.read_flow.return_value = flow
+
+    flow_run = FlowRun(flow_id=flow.id, name=f"{flow_name} run")
+
+    service_job_name = SPCSWorker._slugify_service_name(
+        service_name=flow_name, flow_run_id=flow_run.id
+    )
+
+    assert " " not in service_job_name
+    assert all(c.isalnum() or c in ("-", "_") for c in service_job_name)
+    assert service_job_name.endswith(str(flow_run.id).replace("-", "_"))
+
+    name_without_id = service_job_name[:-37]
+
+    assert name_without_id.replace("_", " ").lower() in flow_name.lower()
+
+    assert (
+        len(service_job_name) <= max_length
+    ), f"Length: {len(service_job_name)}, Max: {max_length}"


Code review comments:
diff_hunk: @@ -0,0 +1,621 @@
+"""Classes and helper functions for creating a worker that can run job services in Snowpark Container Services."""
+
+import datetime
+import os
+import sys
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, Optional
+from uuid import UUID
+
+import anyio
+import dateutil
+import snowflake.connector
+import yaml
+from pydantic import Field
+from slugify import slugify
+from snowflake.connector import SnowflakeConnection
+from snowflake.core import Root
+from snowflake.core.exceptions import NotFoundError
+from snowflake.core.service import ServiceResource
+
+from prefect.client.schemas.objects import FlowRun
+from prefect.utilities.asyncutils import run_sync_in_worker_thread
+from prefect.utilities.dockerutils import get_prefect_image_name
+from prefect.workers.base import (
+    BaseJobConfiguration,
+    BaseVariables,
+    BaseWorker,
+    BaseWorkerResult,
+)
+from prefect_snowflake import SnowflakeCredentials
+
+if TYPE_CHECKING:
+    from prefect.server.schemas.core import Flow
+    from prefect.server.schemas.responses import DeploymentResponse
+
+SPCS_DEFAULT_CPU_REQUEST = "1"
+SPCS_DEFAULT_MEMORY_REQUEST = "1G"
+SPCS_DEFAULT_CPU_LIMIT = None
+SPCS_DEFAULT_MEMORY_LIMIT = None
+
+DEFAULT_CONTAINER_ENTRYPOINT = "/opt/prefect/entrypoint.sh"
+
+
+def _get_default_job_manifest_template() -> dict[str, Any]:
+    """Returns the default job manifest template used by the SPCS job service."""
+    # fmt: off
+    return {
+        "spec": {
+            "containers": [
+                {
+                    "name": "{{ name }}",
+                    "image": "{{ image }}",
+                    "command": "{{ entrypoint }}",
+                    "args": " {{ command }}",
+                    "env": "{{ env }}",
+                    "secrets": "{{ secrets }}",
+                    "volumeMounts": "{{ volume_mounts }}",
+                    "resources": {
+                        "requests": {
+                            "cpu": "{{ cpu_request }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_request }}"
+                        },
+                        "limits": {
+                            "cpu": "{{ cpu_limit }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_limit }}"
+                        },
+                    },
+                }
+            ],
+            "volumes": "{{ volumes }}",
+            "logExporters": {
+                "eventTableConfig": {
+                    "logLevel": "{{ log_level }}"
+                }
+            },
+            "platformMonitor":
+            {
+                "metricConfig": {
+                    "groups": "{{ metrics_groups }}"
+                }
+            }
+        }
+    }
+    # fmt: on
+
+
+class SPCSWorkerConfiguration(BaseJobConfiguration):
+    """Defines variables which can be overridden in the work pool.
+
+    Also handles validation of the config, whether set in the work pool or in the deployment.
+    """
+
+    job_manifest: dict[str, Any] = Field(
+        json_schema_extra={"template": _get_default_job_manifest_template()}
+    )
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
+        description="The fully-qualified name of the compute pool to run services in.",
+        examples=["common.compute.my_compute_pool"],
+    )
+    cpu_request: str = Field(
+        default=SPCS_DEFAULT_CPU_REQUEST,
+        description=f"CPU allocation request for the job service. If not provided, a default value of {SPCS_DEFAULT_CPU_REQUEST} will be used.",
+    )
+    cpu_limit: str = Field(
+        default=SPCS_DEFAULT_CPU_LIMIT,
+        description="CPU allocation limit for the job service. If not provided, there will be no limit.",
+    )
+    gpu_count: Optional[int] = Field(
+        default=None,
+        description="The number of GPUs to use. If not provided, no GPUs will be used.",
+    )
+    memory_request: str = Field(
+        default=SPCS_DEFAULT_MEMORY_REQUEST,
+        description=(
+            "Memory allocation request for the job service."
+            f"If not provided, a default value of {SPCS_DEFAULT_MEMORY_REQUEST} will be used unless present on the task definition."
+        ),
+    )
+    memory_limit: str = Field(
+        default=SPCS_DEFAULT_MEMORY_LIMIT,
+        description="Memory allocation limit for the job service. If not provided, will be the same as the request.",
+    )
+    volume_mounts: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Defines where volumes appears in the container.",
+    )
+    volumes: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Shared file system that can be made available in your container.",
+    )
+    log_level: Literal["INFO", "ERROR", "NONE"] = Field(
+        default="INFO",
+        description="The level of application logs that Snowflake should collect and export to your event table.",
+    )
+    metrics_groups: list[str] = Field(
+        default_factory=list,
+        description="The list of predefined metrics groups that should be logged to the event table.",
+    )
+    query_warehouse: Optional[str] = Field(
+        default=None, description="The query warehouse of the job service."
+    )
+    service_comment: Optional[str] = Field(
+        default=None,
+        description="Specify a comment for the job service. Visible in certain Snowflake logs.",
+    )
+    stream_output: bool = Field(
+        default=False,
+        description="Direct flow log output back to the worker's console.",
+    )
+    service_watch_poll_interval: int = Field(
+        default=5,
+        description="The number of seconds to wait between Snowflake API calls while monitoring the state of the service.",
+    )
+
+    def prepare_for_flow_run(
+        self,
+        flow_run: "FlowRun",
+        deployment: Optional["DeploymentResponse"] = None,
+        flow: Optional["Flow"] = None,
+    ) -> None:
+        """Prepares the job configuration for a flow run."""
+        super().prepare_for_flow_run(flow_run, deployment, flow)
+
+        # There's only one container.
+        container = self.job_manifest["spec"]["containers"][0]
+
+        # Ensure an image is set.
+        container["image"] = self.image
+
+        # Set the container's environment variables by grabbing them from the worker.
+        container["env"] = {**self._base_environment(), **self.env}
+
+        # Convert the entrypoint from a string to a list, because that's what SPCS expects
+        if self.entrypoint:
+            container["command"] = self.entrypoint.split(" ")
+
+        # Convert the command from a string to a list, because that's what SPCS expects
+        if self.command:
+            container["args"] = self.command.split(" ")
+
+        # Set memory_limit to the same value as memory_request, if not specified.
+        if not self.memory_limit:
+            container["resources"]["limits"]["memory"] = self.memory_request
+
+        # Remove the platformMonitor section if there are no metrics groups.
+        if not self.metrics_groups:
+            self.job_manifest["spec"].pop("platformMonitor")
+
+
+class SPCSServiceTemplateVariables(BaseVariables):
+    """Defines variables which can be overridden by deployments.
+
+    Must include all variables in SPCSWorkerConfiguration. All validation should happen in SPCSWorkerConfiguration.
+    """
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
path: src/integrations/prefect-snowflake/prefect_snowflake/workers/spcs.py
text: @user1:
This probably shouldn't have a default value because my runs failed when it wasn't set.
```suggestion
```

diff_hunk: @@ -0,0 +1,621 @@
+"""Classes and helper functions for creating a worker that can run job services in Snowpark Container Services."""
+
+import datetime
+import os
+import sys
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal, Optional
+from uuid import UUID
+
+import anyio
+import dateutil
+import snowflake.connector
+import yaml
+from pydantic import Field
+from slugify import slugify
+from snowflake.connector import SnowflakeConnection
+from snowflake.core import Root
+from snowflake.core.exceptions import NotFoundError
+from snowflake.core.service import ServiceResource
+
+from prefect.client.schemas.objects import FlowRun
+from prefect.utilities.asyncutils import run_sync_in_worker_thread
+from prefect.utilities.dockerutils import get_prefect_image_name
+from prefect.workers.base import (
+    BaseJobConfiguration,
+    BaseVariables,
+    BaseWorker,
+    BaseWorkerResult,
+)
+from prefect_snowflake import SnowflakeCredentials
+
+if TYPE_CHECKING:
+    from prefect.server.schemas.core import Flow
+    from prefect.server.schemas.responses import DeploymentResponse
+
+SPCS_DEFAULT_CPU_REQUEST = "1"
+SPCS_DEFAULT_MEMORY_REQUEST = "1G"
+SPCS_DEFAULT_CPU_LIMIT = None
+SPCS_DEFAULT_MEMORY_LIMIT = None
+
+DEFAULT_CONTAINER_ENTRYPOINT = "/opt/prefect/entrypoint.sh"
+
+
+def _get_default_job_manifest_template() -> dict[str, Any]:
+    """Returns the default job manifest template used by the SPCS job service."""
+    # fmt: off
+    return {
+        "spec": {
+            "containers": [
+                {
+                    "name": "{{ name }}",
+                    "image": "{{ image }}",
+                    "command": "{{ entrypoint }}",
+                    "args": " {{ command }}",
+                    "env": "{{ env }}",
+                    "secrets": "{{ secrets }}",
+                    "volumeMounts": "{{ volume_mounts }}",
+                    "resources": {
+                        "requests": {
+                            "cpu": "{{ cpu_request }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_request }}"
+                        },
+                        "limits": {
+                            "cpu": "{{ cpu_limit }}",
+                            "nvidia.com/gpu": "{{ gpu_count }}",
+                            "memory": "{{ memory_limit }}"
+                        },
+                    },
+                }
+            ],
+            "volumes": "{{ volumes }}",
+            "logExporters": {
+                "eventTableConfig": {
+                    "logLevel": "{{ log_level }}"
+                }
+            },
+            "platformMonitor":
+            {
+                "metricConfig": {
+                    "groups": "{{ metrics_groups }}"
+                }
+            }
+        }
+    }
+    # fmt: on
+
+
+class SPCSWorkerConfiguration(BaseJobConfiguration):
+    """Defines variables which can be overridden in the work pool.
+
+    Also handles validation of the config, whether set in the work pool or in the deployment.
+    """
+
+    job_manifest: dict[str, Any] = Field(
+        json_schema_extra={"template": _get_default_job_manifest_template()}
+    )
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
+        description="The fully-qualified name of the compute pool to run services in.",
+        examples=["common.compute.my_compute_pool"],
+    )
+    cpu_request: str = Field(
+        default=SPCS_DEFAULT_CPU_REQUEST,
+        description=f"CPU allocation request for the job service. If not provided, a default value of {SPCS_DEFAULT_CPU_REQUEST} will be used.",
+    )
+    cpu_limit: str = Field(
+        default=SPCS_DEFAULT_CPU_LIMIT,
+        description="CPU allocation limit for the job service. If not provided, there will be no limit.",
+    )
+    gpu_count: Optional[int] = Field(
+        default=None,
+        description="The number of GPUs to use. If not provided, no GPUs will be used.",
+    )
+    memory_request: str = Field(
+        default=SPCS_DEFAULT_MEMORY_REQUEST,
+        description=(
+            "Memory allocation request for the job service."
+            f"If not provided, a default value of {SPCS_DEFAULT_MEMORY_REQUEST} will be used unless present on the task definition."
+        ),
+    )
+    memory_limit: str = Field(
+        default=SPCS_DEFAULT_MEMORY_LIMIT,
+        description="Memory allocation limit for the job service. If not provided, will be the same as the request.",
+    )
+    volume_mounts: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Defines where volumes appears in the container.",
+    )
+    volumes: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Shared file system that can be made available in your container.",
+    )
+    log_level: Literal["INFO", "ERROR", "NONE"] = Field(
+        default="INFO",
+        description="The level of application logs that Snowflake should collect and export to your event table.",
+    )
+    metrics_groups: list[str] = Field(
+        default_factory=list,
+        description="The list of predefined metrics groups that should be logged to the event table.",
+    )
+    query_warehouse: Optional[str] = Field(
+        default=None, description="The query warehouse of the job service."
+    )
+    service_comment: Optional[str] = Field(
+        default=None,
+        description="Specify a comment for the job service. Visible in certain Snowflake logs.",
+    )
+    stream_output: bool = Field(
+        default=False,
+        description="Direct flow log output back to the worker's console.",
+    )
+    service_watch_poll_interval: int = Field(
+        default=5,
+        description="The number of seconds to wait between Snowflake API calls while monitoring the state of the service.",
+    )
+
+    def prepare_for_flow_run(
+        self,
+        flow_run: "FlowRun",
+        deployment: Optional["DeploymentResponse"] = None,
+        flow: Optional["Flow"] = None,
+    ) -> None:
+        """Prepares the job configuration for a flow run."""
+        super().prepare_for_flow_run(flow_run, deployment, flow)
+
+        # There's only one container.
+        container = self.job_manifest["spec"]["containers"][0]
+
+        # Ensure an image is set.
+        container["image"] = self.image
+
+        # Set the container's environment variables by grabbing them from the worker.
+        container["env"] = {**self._base_environment(), **self.env}
+
+        # Convert the entrypoint from a string to a list, because that's what SPCS expects
+        if self.entrypoint:
+            container["command"] = self.entrypoint.split(" ")
+
+        # Convert the command from a string to a list, because that's what SPCS expects
+        if self.command:
+            container["args"] = self.command.split(" ")
+
+        # Set memory_limit to the same value as memory_request, if not specified.
+        if not self.memory_limit:
+            container["resources"]["limits"]["memory"] = self.memory_request
+
+        # Remove the platformMonitor section if there are no metrics groups.
+        if not self.metrics_groups:
+            self.job_manifest["spec"].pop("platformMonitor")
+
+
+class SPCSServiceTemplateVariables(BaseVariables):
+    """Defines variables which can be overridden by deployments.
+
+    Must include all variables in SPCSWorkerConfiguration. All validation should happen in SPCSWorkerConfiguration.
+    """
+
+    image: str = Field(
+        default_factory=get_prefect_image_name,
+        description="The image to use for the Prefect container in the task. This value defaults to a Prefect base image matching your local versions.",
+        examples=["docker.io/prefecthq/prefect:3-latest"],
+    )
+    image_registry: str = Field(
+        default=None,
+        description="The fully qualified name of the Snowflake image registry.",
+    )
+    entrypoint: Optional[str] = Field(
+        default=DEFAULT_CONTAINER_ENTRYPOINT,
+        description=(
+            "The entrypoint of the container you wish you run."
+            "This value defaults to the entrypoint used by Prefect images and should only be changed when using a custom image that is not based on an official Prefect image."
+        ),
+    )
+    snowflake_credentials: SnowflakeCredentials = Field(
+        default_factory=SnowflakeCredentials,
+        description="Snowflake credentials to use when creating job services.",
+    )
+    secrets: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Snowflake secrets to inject into the container as env variables or files.",
+    )
+    external_access_integrations: list[str] = Field(
+        default_factory=list,
+        description="The names of the external access integrations that the service should be created with.",
+    )
+    compute_pool: str = Field(
+        default=None,
+        description="The fully-qualified name of the compute pool to run services in.",
+        examples=["common.compute.my_compute_pool"],
+    )
+    cpu_request: str = Field(
+        default=SPCS_DEFAULT_CPU_REQUEST,
+        description=f"CPU allocation request for the job service. If not provided, a default value of {SPCS_DEFAULT_CPU_REQUEST} will be used.",
+    )
+    cpu_limit: str = Field(
+        default=SPCS_DEFAULT_CPU_LIMIT,
+        description="CPU allocation limit for the job service. If not provided, there will be no limit.",
+    )
+    gpu_count: Optional[int] = Field(
+        default=None,
+        description="The number of GPUs to use. If not provided, no GPUs will be used.",
+    )
+    memory_request: str = Field(
+        default=SPCS_DEFAULT_MEMORY_REQUEST,
+        description=(
+            "Memory allocation request for the job service."
+            f"If not provided, a default value of {SPCS_DEFAULT_MEMORY_REQUEST} will be used unless present on the task definition."
+        ),
+    )
+    memory_limit: str = Field(
+        default=SPCS_DEFAULT_MEMORY_LIMIT,
+        description="Memory allocation limit for the job service. If not provided, will be the same as the request.",
+    )
+    volume_mounts: list[dict[str, str]] = Field(
+        default_factory=list,
+        description="Defines where volumes appears in the container.",
+    )
+    volumes: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Shared file system that can be made available in your container.",
+    )
+    log_level: Literal["INFO", "ERROR", "NONE"] = Field(
+        default="INFO",
+        description="The level of application logs that Snowflake should collect and export to your event table.",
+    )
+    metrics_groups: list[str] = Field(
+        default_factory=list,
+        description="The list of predefined metrics groups that should be logged to the event table.",
+    )
+    query_warehouse: Optional[str] = Field(
+        default=None, description="The query warehouse of the job service."
+    )
+    service_comment: Optional[str] = Field(
+        default=None,
+        description="Specify a comment for the job service. Visible in certain Snowflake logs.",
+    )
+    stream_output: bool = Field(
+        default=False,
+        description="Direct flow log output back to the worker's console.",
+    )
+    service_watch_poll_interval: int = Field(
+        default=5,
+        description="The number of seconds to wait between Snowflake API calls while monitoring the state of the service.",
+    )
+
+
+class SPCSWorkerResult(BaseWorkerResult):
+    """Result returned by the SPCSWorker."""
+
+
+class SPCSWorker(BaseWorker):
+    """A Prefect worker that runs flows as service jobs in Snowpark Container Services."""
+
+    type: str = "snowpark-container-service"
+    job_configuration = SPCSWorkerConfiguration
+    job_configuration_variables = SPCSServiceTemplateVariables
+    _description = "Execute flow runs within containers on Snowflake's Snowpark Container Services. Requires a Snowflake account."
+
+    async def run(
+        self,
+        flow_run: FlowRun,
+        configuration: SPCSWorkerConfiguration,
+        task_status: Optional[anyio.abc.TaskStatus] = None,
+    ) -> BaseWorkerResult:
+        """Run a flow as a service job in Snowpark Container Services.
+
+        Args:
+            flow_run: The flow run to run.
+            configuration: The configuration for the flow run.
+            task_status: The task status object for the current task. Used
+            to provide an identifier that can be used to cancel the task.
+
+        Returns:
+            The result of the flow run.
+
+        """
+        run_start_time = datetime.now(datetime.timezone.utc)
path: src/integrations/prefect-snowflake/prefect_snowflake/workers/spcs.py
text: @user1:
I had to update this while testing.
```suggestion
        run_start_time = datetime.datetime.now(datetime.timezone.utc)
```