ODSC-39731/support_running_sparkpipelinemodel_in_df (#152)

Author: mingkang111

ads/common/object_storage_details.py

@@ -121,3 +121,20 @@ def is_valid_uri(uri: str) -> bool:
                 "It must follow the pattern `oci://<bucket_name>@<namespace>/<prefix>`."
             )
         return True
+
+    @staticmethod
+    def is_oci_path(uri: str = None) -> bool:
+        """Check if the given path is oci object storage uri.
+
+        Parameters
+        ----------
+        uri: str
+            The URI of the target.
+
+        Returns
+        -------
+        bool: return True if the path is oci object storage uri.
+        """
+        if not uri:
+            return False
+        return uri.startswith("oci://")

ads/common/utils.py

@@ -495,9 +495,7 @@ def print_user_message(
         )
 
     if is_documentation_mode() and is_notebook():
-
         if display_type.lower() == "tip":
-
             if "\n" in msg:
                 t = "<b>{}:</b>".format(title.upper().strip()) if title else ""
 
@@ -567,7 +565,6 @@ def print_user_message(
             )
 
         elif display_type.startswith("info"):
-
             user_message = msg.strip().replace("\n", "<br>")
 
             if see_also_links:
@@ -640,7 +637,6 @@ def ellipsis_strings(raw, n=24):
 
     result = []
     for s in sequence:
-
         if len(str(s)) <= n:
             result.append(s)
         else:
@@ -1136,36 +1132,44 @@ def is_data_too_wide(
     return col_num > max_col_num
 
 
-def get_files(directory: str):
+def get_files(directory: str, auth: Optional[Dict] = None):
     """List out all the file names under this directory.
 
     Parameters
     ----------
     directory: str
         The directory to list out all the files from.
+    auth: (Dict, optional). Defaults to None.
+        The default authentication is set using `ads.set_auth` API. If you need to override the
+        default, use the `ads.common.auth.api_keys` or `ads.common.auth.resource_principal` to create appropriate
+        authentication signer and kwargs required to instantiate IdentityClient object.
 
     Returns
     -------
     List
         List of the files in the directory.
     """
     directory = directory.rstrip("/")
-    if os.path.exists(os.path.join(directory, ".model-ignore")):
-        ignore_patterns = (
-            Path(os.path.join(directory), ".model-ignore")
-            .read_text()
-            .strip()
-            .split("\n")
-        )
+    path_scheme = urlparse(directory).scheme or "file"
+    storage_options = auth or authutil.default_signer()
+    model_ignore_path = os.path.join(directory, ".model-ignore")
+    if is_path_exists(model_ignore_path, auth=auth):
+        with fsspec.open(model_ignore_path, "r", **storage_options) as f:
+            ignore_patterns = f.read().strip().split("\n")
     else:
         ignore_patterns = []
     file_names = []
-    for root, dirs, files in os.walk(directory):
+    fs = fsspec.filesystem(path_scheme, **storage_options)
+    for root, dirs, files in fs.walk(directory):
         for name in files:
             file_names.append(os.path.join(root, name))
         for name in dirs:
             file_names.append(os.path.join(root, name))
 
+    # return all files in remote directory.
+    if directory.startswith("oci://"):
+        directory = directory.lstrip("oci://")
+
     for ignore in ignore_patterns:
         if not ignore.startswith("#") and ignore.strip() != "":
             matches = []
@@ -1228,7 +1232,7 @@ def copy_from_uri(
     force_overwrite: (bool, optional). Defaults to False.
         Whether to overwrite existing files or not.
     auth: (Dict, optional). Defaults to None.
-        The default authetication is set using `ads.set_auth` API. If you need to override the
+        The default authentication is set using `ads.set_auth` API. If you need to override the
         default, use the `ads.common.auth.api_keys` or `ads.common.auth.resource_principal` to create appropriate
         authentication signer and kwargs required to instantiate IdentityClient object.
 
@@ -1294,7 +1298,7 @@ def copy_file(
     force_overwrite: (bool, optional). Defaults to False.
         Whether to overwrite existing files or not.
     auth: (Dict, optional). Defaults to None.
-        The default authetication is set using `ads.set_auth` API. If you need to override the
+        The default authentication is set using `ads.set_auth` API. If you need to override the
         default, use the `ads.common.auth.api_keys` or `ads.common.auth.resource_principal` to create appropriate
         authentication signer and kwargs required to instantiate IdentityClient object.
     chunk_size: (int, optinal). Defaults to `DEFAULT_BUFFER_SIZE`
@@ -1357,7 +1361,7 @@ def remove_file(file_path: str, auth: Optional[Dict] = None) -> None:
     file_path: str
         The path of the source file, which can be local path or OCI object storage URI.
     auth: (Dict, optional). Defaults to None.
-        The default authetication is set using `ads.set_auth` API. If you need to override the
+        The default authentication is set using `ads.set_auth` API. If you need to override the
         default, use the `ads.common.auth.api_keys` or `ads.common.auth.resource_principal` to create appropriate
         authentication signer and kwargs required to instantiate IdentityClient object.
 
@@ -1570,3 +1574,26 @@ def extract_region(auth: Optional[Dict] = None) -> Union[str, None]:
         pass
 
     return None
+
+
+def is_path_exists(uri: str, auth: Optional[Dict] = None) -> bool:
+    """Check if the given path which can be local path or OCI object storage URI exists.
+
+    Parameters
+    ----------
+    uri: str
+        The URI of the target, which can be local path or OCI object storage URI.
+    auth: (Dict, optional). Defaults to None.
+        The default authentication is set using `ads.set_auth` API. If you need to override the
+        default, use the `ads.common.auth.api_keys` or `ads.common.auth.resource_principal` to create appropriate
+        authentication signer and kwargs required to instantiate IdentityClient object.
+
+    Returns
+    -------
+    bool: return True if the path exists.
+    """
+    path_scheme = urlparse(uri).scheme or "file"
+    storage_options = auth or authutil.default_signer()
+    if fsspec.filesystem(path_scheme, **storage_options).exists(uri):
+        return True
+    return False

ads/feature_engineering/schema.py

@@ -4,13 +4,21 @@
 # Copyright (c) 2021, 2022 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
+import asteval
+import fsspec
+import json
+import os
+import sys
+import yaml
 from abc import ABC, abstractmethod
+from cerberus import Validator
+from copy import deepcopy
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional
-from copy import deepcopy
-
-import yaml
+from string import Template
+from os import path
 from ads.common.serializer import DataClassSerializable
+from ads.common.object_storage_details import ObjectStorageDetails
 
 try:
     from yaml import CDumper as dumper
@@ -19,24 +27,13 @@
     from yaml import Dumper as dumper
     from yaml import Loader as loader
 
-import json
-import os
-import asteval
-from os import path
-
-import fsspec
-import yaml
-from cerberus import Validator
-import sys
-from string import Template
-
-
 SCHEMA_VALIDATOR_NAME = "data_schema.json"
 INPUT_OUTPUT_SCHENA_SIZE_LIMIT = 32000
 SCHEMA_VERSION = "1.1"
 DEFAULT_SCHEMA_VERSION = "1.0"
 SCHEMA_KEY = "schema"
 SCHEMA_VERSION_KEY = "version"
+DEFAULT_STORAGE_OPTIONS = None
 
 
 class SchemaSizeTooLarge(ValueError):
@@ -685,13 +682,16 @@ def to_json(self):
         """
         return json.dumps(self.to_dict()).replace("NaN", "null")
 
-    def to_json_file(self, file_path):
+    def to_json_file(self, file_path, storage_options: dict = None):
         """Saves the data schema into a json file.
 
         Parameters
         ----------
         file_path : str
             File Path to store the schema in json format.
+        storage_options: dict. Default None
+            Parameters passed on to the backend filesystem class.
+            Defaults to `storage_options` set using `DatasetFactory.set_default_storage()`.
 
         Returns
         -------
@@ -704,12 +704,19 @@ def to_json_file(self, file_path):
             ".json"
         ], f"The file `{basename}` is not a valid JSON file. The `{file_path}` must have the extension .json."
         if directory and not os.path.exists(directory):
-            try:
-                os.mkdir(directory)
-            except:
-                raise Exception(f"Error creating the directory.")
-        with open(os.path.join(directory, basename), "w") as json_file:
-            json.dump(self.to_dict(), json_file)
+            if not ObjectStorageDetails.is_oci_path(directory):
+                try:
+                    os.mkdir(directory)
+                except:
+                    raise Exception(f"Error creating the directory.")
+        if not storage_options:
+            storage_options = DEFAULT_STORAGE_OPTIONS or {"config": {}}
+        with fsspec.open(
+            os.path.join(directory, basename),
+            mode="w",
+            **(storage_options),
+        ) as f:
+            f.write(json.dumps(self.to_dict()))
 
     def to_yaml_file(self, file_path):
         """Saves the data schema into a yaml file.