Data subsystem (#1526)

Signed-off-by: Ketan Umare <[email protected]>
Signed-off-by: Yee Hing Tong <[email protected]>

Dockerfile.dev

@@ -0,0 +1,38 @@
+# This Dockerfile is here to help with end-to-end testing
+# From flytekit
+# $ docker build -f Dockerfile.dev --build-arg PYTHON_VERSION=3.10 -t localhost:30000/flytekittest:someversion .
+# $ docker push localhost:30000/flytekittest:someversion
+# From your test user code
+# $ pyflyte run --image localhost:30000/flytekittest:someversion
+
+ARG PYTHON_VERSION
+FROM python:${PYTHON_VERSION}-slim-buster
+
+MAINTAINER Flyte Team <[email protected]>
+LABEL org.opencontainers.image.source https://github.com/flyteorg/flytekit
+
+WORKDIR /root
+ENV PYTHONPATH /root
+
+ARG VERSION
+ARG DOCKER_IMAGE
+
+RUN apt-get update && apt-get install build-essential vim -y
+
+COPY . /code/flytekit
+WORKDIR /code/flytekit
+
+# Pod tasks should be exposed in the default image
+RUN pip install -e .
+RUN pip install -e plugins/flytekit-k8s-pod
+RUN pip install -e plugins/flytekit-deck-standard
+RUN pip install scikit-learn
+
+ENV PYTHONPATH "/code/flytekit:/code/flytekit/plugins/flytekit-k8s-pod:/code/flytekit/plugins/flytekit-deck-standard:"
+
+WORKDIR /root
+RUN useradd -u 1000 flytekit
+RUN chown flytekit: /root
+USER flytekit
+
+ENV FLYTE_INTERNAL_IMAGE "$DOCKER_IMAGE"

doc-requirements.txt

@@ -216,7 +216,7 @@ frozenlist==1.3.3
     # via
     #   aiosignal
     #   ray
-fsspec==2023.1.0
+fsspec==2023.3.0
     # via
     #   -r doc-requirements.in
     #   dask

flytekit/__init__.py

@@ -205,7 +205,6 @@
 from flytekit.core.condition import conditional
 from flytekit.core.container_task import ContainerTask
 from flytekit.core.context_manager import ExecutionParameters, FlyteContext, FlyteContextManager
-from flytekit.core.data_persistence import DataPersistence, DataPersistencePlugins
 from flytekit.core.dynamic_workflow_task import dynamic
 from flytekit.core.gate import approve, sleep, wait_for_input
 from flytekit.core.hash import HashMethod
@@ -223,7 +222,6 @@
 from flytekit.core.workflow import WorkflowFailurePolicy, reference_workflow, workflow
 from flytekit.deck import Deck
 from flytekit.extras import pytorch, sklearn, tensorflow
-from flytekit.extras.persistence import GCSPersistence, HttpPersistence, S3Persistence
 from flytekit.loggers import logger
 from flytekit.models.common import Annotations, AuthRole, Labels
 from flytekit.models.core.execution import WorkflowExecutionPhase

flytekit/configuration/__init__.py

@@ -352,7 +352,7 @@ class PlatformConfig(object):
     This object contains the settings to talk to a Flyte backend (the DNS location of your Admin server basically).
 
     :param endpoint: DNS for Flyte backend
-    :param insecure: Whether to use SSL
+    :param insecure: Whether or not to use SSL
     :param insecure_skip_verify: Whether to skip SSL certificate verification
     :param console_endpoint: endpoint for console if different from Flyte backend
     :param command: This command is executed to return a token using an external process

flytekit/core/checkpointer.py

@@ -126,7 +126,7 @@ def save(self, cp: typing.Union[Path, str, io.BufferedReader]):
                 fa.upload_directory(str(cp), self._checkpoint_dest)
             else:
                 fname = cp.stem + cp.suffix
-                rpath = fa._default_remote.construct_path(False, False, self._checkpoint_dest, fname)
+                rpath = fa._default_remote.sep.join([str(self._checkpoint_dest), fname])
                 fa.upload(str(cp), rpath)
             return
 
@@ -138,7 +138,7 @@ def save(self, cp: typing.Union[Path, str, io.BufferedReader]):
         with dest_cp.open("wb") as f:
             f.write(cp.read())
 
-        rpath = fa._default_remote.construct_path(False, False, self._checkpoint_dest, self.TMP_DST_PATH)
+        rpath = fa._default_remote.sep.join([str(self._checkpoint_dest), self.TMP_DST_PATH])
         fa.upload(str(dest_cp), rpath)
 
     def read(self) -> typing.Optional[bytes]:

flytekit/core/context_manager.py

@@ -84,7 +84,7 @@ class Builder(object):
         decks: List[Deck]
         raw_output_prefix: Optional[str] = None
         execution_id: typing.Optional[_identifier.WorkflowExecutionIdentifier] = None
-        working_dir: typing.Optional[utils.AutoDeletingTempDir] = None
+        working_dir: typing.Optional[str] = None
         checkpoint: typing.Optional[Checkpoint] = None
         execution_date: typing.Optional[datetime] = None
         logging: Optional[_logging.Logger] = None
@@ -202,12 +202,10 @@ def raw_output_prefix(self) -> str:
         return self._raw_output_prefix
 
     @property
-    def working_directory(self) -> utils.AutoDeletingTempDir:
+    def working_directory(self) -> str:
         """
         A handle to a special working directory for easily producing temporary files.
-
         TODO: Usage examples
-        TODO: This does not always return a AutoDeletingTempDir
         """
         return self._working_directory