Data subsystem (#1526)

Signed-off-by: Ketan Umare <ketan.umare@gmail.com>
Signed-off-by: Yee Hing Tong <wild-endeavor@users.noreply.github.com>

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 <users@flyte.org>
+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

@@ -213,7 +213,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

@@ -206,7 +206,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
@@ -224,7 +223,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