Serialize and Deserialize from S3/URLs

docs/source/changelog.rst

@@ -6,6 +6,7 @@ Changelog
     * Enhancements
         * Added drop_first as param in encode_features (:pr:`647`)
         * Generate transform features of direct features (:pr:`623`)
+        * Added serializing and deserializing from S3 and deserializing from URLs (:pr:`685`)
     * Fixes
         * Fix performance regression in DFS (:pr:`637`)
         * Fix deserialization of feature relationship path (:pr:`665`)

featuretools/entityset/deserialize.py

@@ -1,13 +1,27 @@
 import json
 import os
+import tarfile
+from pathlib import Path
 
+import boto3
 import pandas as pd
 
 from featuretools.entityset.relationship import Relationship
 from featuretools.entityset.serialize import FORMATS
-from featuretools.utils.gen_utils import check_schema_version
+from featuretools.utils.gen_utils import (
+    check_schema_version,
+    is_python_2,
+    use_s3fs_es,
+    use_smartopen_es
+)
+from featuretools.utils.wrangle import _is_s3, _is_url
 from featuretools.variable_types.variable import find_variable_types
 
+if is_python_2():
+    from backports import tempfile
+else:
+    import tempfile
+
 
 def description_to_variable(description, entity=None):
     '''Deserialize variable from variable description.
@@ -132,14 +146,15 @@ def read_entity_data(description, path):
 
 
 def read_data_description(path):
-    '''Read data description from disk.
+    '''Read data description from disk, S3 path, or URL.
 
         Args:
-            path (str): Location on disk to read `data_description.json`.
+            path (str): Location on disk, S3 path, or URL to read `data_description.json`.
 
         Returns:
             description (dict) : Description of :class:`.EntitySet`.
     '''
+
     path = os.path.abspath(path)
     assert os.path.exists(path), '"{}" does not exist'.format(path)
     file = os.path.join(path, 'data_description.json')
@@ -149,12 +164,38 @@ def read_data_description(path):
     return description
 
 
-def read_entityset(path, **kwargs):
-    '''Read entityset from disk.
+def read_entityset(path, profile_name=None, **kwargs):
+    '''Read entityset from disk, S3 path, or URL.
 
         Args:
-            path (str): Directory on disk to read `data_description.json`.
+            path (str): Directory on disk, S3 path, or URL to read `data_description.json`.
+            profile_name (str, bool): The AWS profile specified to write to S3. Will default to None and search for AWS credentials.
+                Set to False to use an anonymous profile.
             kwargs (keywords): Additional keyword arguments to pass as keyword arguments to the underlying deserialization method.
     '''
-    data_description = read_data_description(path)
-    return description_to_entityset(data_description, **kwargs)
+    if _is_url(path) or _is_s3(path):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            file_name = Path(path).name
+            file_path = os.path.join(tmpdir, file_name)
+            transport_params = {}
+            session = boto3.Session()
+
+            if _is_url(path):
+                use_smartopen_es(file_path, path)
+            elif isinstance(profile_name, str):
+                transport_params = {'session': boto3.Session(profile_name=profile_name)}
+                use_smartopen_es(file_path, path, transport_params)
+            elif profile_name is False:
+                use_s3fs_es(file_path, path)
+            elif session.get_credentials() is not None:
+                use_smartopen_es(file_path, path)
+            else:
+                use_s3fs_es(file_path, path)
+
+            tar = tarfile.open(str(file_path))
+            tar.extractall(path=tmpdir)
+            data_description = read_data_description(tmpdir)
+            return description_to_entityset(data_description, **kwargs)
+    else:
+        data_description = read_data_description(path)
+        return description_to_entityset(data_description, **kwargs)

featuretools/entityset/entityset.py

@@ -143,38 +143,47 @@ def metadata(self):
     def reset_data_description(self):
         self._data_description = None
 
-    def to_pickle(self, path, compression=None):
-        '''Write entityset to disk in the pickle format, location specified by `path`.
+    def to_pickle(self, path, compression=None, profile_name=None):
+        '''Write entityset in the pickle format, location specified by `path`.
+            Path could be a local path or a S3 path.
+            If writing to S3 a tar archive of files will be written.
 
             Args:
                 path (str): location on disk to write to (will be created as a directory)
                 compression (str) : Name of the compression to use. Possible values are: {'gzip', 'bz2', 'zip', 'xz', None}.
+                profile_name (str) : Name of AWS profile to use, False to use an anonymous profile, or None.
         '''
-        serialize.write_data_description(self, path, format='pickle', compression=compression)
+        serialize.write_data_description(self, path, format='pickle', compression=compression, profile_name=profile_name)
         return self
 
-    def to_parquet(self, path, engine='auto', compression=None):
+    def to_parquet(self, path, engine='auto', compression=None, profile_name=None):
         '''Write entityset to disk in the parquet format, location specified by `path`.
+            Path could be a local path or a S3 path.
+            If writing to S3 a tar archive of files will be written.
 
             Args:
                 path (str): location on disk to write to (will be created as a directory)
                 engine (str) : Name of the engine to use. Possible values are: {'auto', 'pyarrow', 'fastparquet'}.
                 compression (str) : Name of the compression to use. Possible values are: {'snappy', 'gzip', 'brotli', None}.
+                profile_name (str) : Name of AWS profile to use, False to use an anonymous profile, or None.
         '''
-        serialize.write_data_description(self, path, format='parquet', engine=engine, compression=compression)
+        serialize.write_data_description(self, path, format='parquet', engine=engine, compression=compression, profile_name=profile_name)
         return self
 
-    def to_csv(self, path, sep=',', encoding='utf-8', engine='python', compression=None):
+    def to_csv(self, path, sep=',', encoding='utf-8', engine='python', compression=None, profile_name=None):
         '''Write entityset to disk in the csv format, location specified by `path`.
+            Path could be a local path or a S3 path.
+            If writing to S3 a tar archive of files will be written.
 
             Args:
                 path (str) : Location on disk to write to (will be created as a directory)
                 sep (str) : String of length 1. Field delimiter for the output file.
                 encoding (str) : A string representing the encoding to use in the output file, defaults to 'utf-8'.
                 engine (str) : Name of the engine to use. Possible values are: {'c', 'python'}.
                 compression (str) : Name of the compression to use. Possible values are: {'gzip', 'bz2', 'zip', 'xz', None}.
+                profile_name (str) : Name of AWS profile to use, False to use an anonymous profile, or None.
         '''
-        serialize.write_data_description(self, path, format='csv', index=False, sep=sep, encoding=encoding, engine=engine, compression=compression)
+        serialize.write_data_description(self, path, format='csv', index=False, sep=sep, encoding=encoding, engine=engine, compression=compression, profile_name=profile_name)
         return self
 
     def to_dictionary(self):

featuretools/entityset/serialize.py

@@ -1,6 +1,22 @@
+import datetime
 import json
 import os
 import shutil
+import tarfile
+
+import boto3
+
+from featuretools.utils.gen_utils import (
+    is_python_2,
+    use_s3fs_es,
+    use_smartopen_es
+)
+from featuretools.utils.wrangle import _is_s3, _is_url
+
+if is_python_2():
+    from backports import tempfile
+else:
+    import tempfile
 
 FORMATS = ['csv', 'pickle', 'parquet']
 SCHEMA_VERSION = "1.0.0"
@@ -58,7 +74,7 @@ def entityset_to_description(entityset):
 
 
 def write_entity_data(entity, path, format='csv', **kwargs):
-    '''Write entity data to disk.
+    '''Write entity data to disk or S3 path.
 
     Args:
         entity (Entity) : Instance of :class:`.Entity`.
@@ -97,23 +113,58 @@ def write_entity_data(entity, path, format='csv', **kwargs):
     return {'location': location, 'type': format, 'params': kwargs}
 
 
-def write_data_description(entityset, path, **kwargs):
-    '''Serialize entityset to data description and write to disk.
+def write_data_description(entityset, path, profile_name=None, **kwargs):
+    '''Serialize entityset to data description and write to disk or S3 path.
 
     Args:
         entityset (EntitySet) : Instance of :class:`.EntitySet`.
-        path (str) : Location on disk to write `data_description.json` and entity data.
-        kwargs (keywords) : Additional keyword arguments to pass as keywords arguments to the underlying serialization method.
+        path (str) : Location on disk or S3 path to write `data_description.json` and entity data.
+        profile_name (str, bool): The AWS profile specified to write to S3. Will default to None and search for AWS credentials.
+            Set to False to use an anonymous profile.
+        kwargs (keywords) : Additional keyword arguments to pass as keywords arguments to the underlying serialization method or to specify AWS profile.
     '''
-    path = os.path.abspath(path)
-    if os.path.exists(path):
-        shutil.rmtree(path)
-    for dirname in [path, os.path.join(path, 'data')]:
-        os.makedirs(dirname)
+    if _is_s3(path):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.makedirs(os.path.join(tmpdir, 'data'))
+            dump_data_description(entityset, tmpdir, **kwargs)
+            file_path = create_archive(tmpdir)
+
+            transport_params = {}
+            session = boto3.Session()
+            if isinstance(profile_name, str):
+                transport_params = {'session': boto3.Session(profile_name=profile_name)}
+                use_smartopen_es(file_path, path, transport_params, read=False)
+            elif profile_name is False:
+                use_s3fs_es(file_path, path, read=False)
+            elif session.get_credentials() is not None:
+                use_smartopen_es(file_path, path, read=False)
+            else:
+                use_s3fs_es(file_path, path, read=False)
+    elif _is_url(path):
+        raise ValueError("Writing to URLs is not supported")
+    else:
+        path = os.path.abspath(path)
+        if os.path.exists(path):
+            shutil.rmtree(path)
+        os.makedirs(os.path.join(path, 'data'))
+        dump_data_description(entityset, path, **kwargs)
+
+
+def dump_data_description(entityset, path, **kwargs):
     description = entityset_to_description(entityset)
     for entity in entityset.entities:
         loading_info = write_entity_data(entity, path, **kwargs)
         description['entities'][entity.id]['loading_info'].update(loading_info)
     file = os.path.join(path, 'data_description.json')
     with open(file, 'w') as file:
         json.dump(description, file)
+
+
+def create_archive(tmpdir):
+    file_name = "es-{date:%Y-%m-%d_%H:%M:%S}.tar".format(date=datetime.datetime.now())
+    file_path = os.path.join(tmpdir, file_name)
+    tar = tarfile.open(str(file_path), 'w')
+    tar.add(str(tmpdir) + '/data_description.json', arcname='/data_description.json')
+    tar.add(str(tmpdir) + '/data', arcname='/data')
+    tar.close()
+    return file_path

featuretools/feature_base/features_deserializer.py

@@ -1,5 +1,7 @@
 import json
 
+import boto3
+
 from featuretools.entityset.deserialize import \
     description_to_entityset as deserialize_es
 from featuretools.feature_base.feature_base import (
@@ -12,23 +14,31 @@
     TransformFeature
 )
 from featuretools.primitives.utils import PrimitivesDeserializer
-from featuretools.utils.gen_utils import check_schema_version
+from featuretools.utils.gen_utils import (
+    check_schema_version,
+    use_s3fs_features,
+    use_smartopen_features
+)
+from featuretools.utils.wrangle import _is_s3, _is_url
 
 
-def load_features(features):
-    """Loads the features from a filepath, an open file, or a JSON formatted string.
+def load_features(features, profile_name=None):
+    """Loads the features from a filepath, S3 path, URL, an open file, or a JSON formatted string.
 
     Args:
         features (str or :class:`.FileObject`): The location of where features has
         been saved which this must include the name of the file, or a JSON formatted
         string, or a readable file handle where the features have been saved.
 
+        profile_name (str, bool): The AWS profile specified to write to S3. Will default to None and search for AWS credentials.
+            Set to False to use an anonymous profile.
+
     Returns:
         features (list[:class:`.FeatureBase`]): Feature definitions list.
 
     Note:
-        Features saved in one version of Featuretools are not guaranteed to work in another.
-        After upgrading Featuretools, features may need to be generated again.
+        Features saved in one version of Featuretools or python are not guaranteed to work in another.
+        After upgrading Featuretools or python, features may need to be generated again.
 
     Example:
         .. ipython:: python
@@ -51,7 +61,7 @@ def load_features(features):
     .. seealso::
         :func:`.save_features`
     """
-    return FeaturesDeserializer.load(features).to_list()
+    return FeaturesDeserializer.load(features, profile_name).to_list()
 
 
 class FeaturesDeserializer(object):
@@ -73,13 +83,27 @@ def __init__(self, features_dict):
         self._primitives_deserializer = PrimitivesDeserializer()
 
     @classmethod
-    def load(cls, features):
+    def load(cls, features, profile_name):
         if isinstance(features, str):
             try:
                 features_dict = json.loads(features)
             except ValueError:
-                with open(features, 'r') as f:
-                    features_dict = json.load(f)
+                if _is_url(features):
+                    features_dict = use_smartopen_features(features)
+                elif _is_s3(features):
+                    session = boto3.Session()
+                    if isinstance(profile_name, str):
+                        transport_params = {'session': boto3.Session(profile_name=profile_name)}
+                        features_dict = use_smartopen_features(features, transport_params)
+                    elif profile_name is False:
+                        features_dict = use_s3fs_features(features)
+                    elif session.get_credentials() is not None:
+                        features_dict = use_smartopen_features(features)
+                    else:
+                        features_dict = use_s3fs_features(features)
+                else:
+                    with open(features, 'r') as f:
+                        features_dict = json.load(f)
             return cls(features_dict)
         return cls(json.load(features))