Source code for apache_beam.io.aws.s3io

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""AWS S3 client
"""

# pytype: skip-file

import errno
import io
import logging
import re
import time
import traceback

from apache_beam.io.aws.clients.s3 import messages
from apache_beam.io.filesystemio import Downloader
from apache_beam.io.filesystemio import DownloaderStream
from apache_beam.io.filesystemio import Uploader
from apache_beam.io.filesystemio import UploaderStream
from apache_beam.utils import retry

try:
  # pylint: disable=wrong-import-order, wrong-import-position
  # pylint: disable=ungrouped-imports
  from apache_beam.io.aws.clients.s3 import boto3_client
  BOTO3_INSTALLED = True
except ImportError:
  BOTO3_INSTALLED = False

MAX_BATCH_OPERATION_SIZE = 100


[docs]def parse_s3_path(s3_path, object_optional=False): """Return the bucket and object names of the given s3:// path.""" match = re.match('^s3://([^/]+)/(.*)$', s3_path) if match is None or (match.group(2) == '' and not object_optional): raise ValueError('S3 path must be in the form s3://<bucket>/<object>.') return match.group(1), match.group(2)
[docs]class S3IO(object): """S3 I/O client.""" def __init__(self, client=None, options=None): if client is None and options is None: raise ValueError('Must provide one of client or options') if client is not None: self.client = client elif BOTO3_INSTALLED: self.client = boto3_client.Client(options=options) else: message = 'AWS dependencies are not installed, and no alternative ' \ 'client was provided to S3IO.' raise RuntimeError(message)
[docs] def open( self, filename, mode='r', read_buffer_size=16 * 1024 * 1024, mime_type='application/octet-stream'): """Open an S3 file path for reading or writing. Args: filename (str): S3 file path in the form ``s3://<bucket>/<object>``. mode (str): ``'r'`` for reading or ``'w'`` for writing. read_buffer_size (int): Buffer size to use during read operations. mime_type (str): Mime type to set for write operations. Returns: S3 file object. Raises: ValueError: Invalid open file mode. """ if mode == 'r' or mode == 'rb': downloader = S3Downloader( self.client, filename, buffer_size=read_buffer_size) return io.BufferedReader( DownloaderStream(downloader, mode=mode), buffer_size=read_buffer_size) elif mode == 'w' or mode == 'wb': uploader = S3Uploader(self.client, filename, mime_type) return io.BufferedWriter( UploaderStream(uploader, mode=mode), buffer_size=128 * 1024) else: raise ValueError('Invalid file open mode: %s.' % mode)
[docs] @retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def list_prefix(self, path, with_metadata=False): """Lists files matching the prefix. Args: path: S3 file path pattern in the form s3://<bucket>/[name]. with_metadata: Experimental. Specify whether returns file metadata. Returns: If ``with_metadata`` is False: dict of file name -> size; if ``with_metadata`` is True: dict of file name -> tuple(size, timestamp). """ bucket, prefix = parse_s3_path(path, object_optional=True) request = messages.ListRequest(bucket=bucket, prefix=prefix) file_info = {} counter = 0 start_time = time.time() if with_metadata: logging.debug("Starting the file information of the input") else: logging.debug("Starting the size estimation of the input") while True: #The list operation will raise an exception #when trying to list a nonexistent S3 path. #This should not be an issue here. #Ignore this exception or it will break the procedure. try: response = self.client.list(request) except messages.S3ClientError as e: if e.code == 404: break else: raise e for item in response.items: file_name = 's3://%s/%s' % (bucket, item.key) if with_metadata: file_info[file_name] = ( item.size, self._updated_to_seconds(item.last_modified)) else: file_info[file_name] = item.size counter += 1 if counter % 10000 == 0: if with_metadata: logging.info( "Finished computing file information of: %s files", len(file_info)) else: logging.info("Finished computing size of: %s files", len(file_info)) if response.next_token: request.continuation_token = response.next_token else: break logging.log( # do not spam logs when list_prefix is likely used to check empty folder logging.INFO if counter > 0 else logging.DEBUG, "Finished listing %s files in %s seconds.", counter, time.time() - start_time) return file_info
[docs] def checksum(self, path): """Looks up the checksum of an S3 object. Args: path: S3 file path pattern in the form s3://<bucket>/<name>. """ return self._s3_object(path).etag
[docs] @retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def copy(self, src, dest): """Copies a single S3 file object from src to dest. Args: src: S3 file path pattern in the form s3://<bucket>/<name>. dest: S3 file path pattern in the form s3://<bucket>/<name>. Raises: TimeoutError: on timeout. """ src_bucket, src_key = parse_s3_path(src) dest_bucket, dest_key = parse_s3_path(dest) request = messages.CopyRequest(src_bucket, src_key, dest_bucket, dest_key) self.client.copy(request)
# We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def copy_paths(self, src_dest_pairs): """Copies the given S3 objects from src to dest. This can handle directory or file paths. Args: src_dest_pairs: list of (src, dest) tuples of s3://<bucket>/<name> file paths to copy from src to dest Returns: List of tuples of (src, dest, exception) in the same order as the src_dest_pairs argument, where exception is None if the operation succeeded or the relevant exception if the operation failed. """ if not src_dest_pairs: return [] results = [] for src_path, dest_path in src_dest_pairs: # Copy a directory with self.copy_tree if src_path.endswith('/') and dest_path.endswith('/'): try: results += self.copy_tree(src_path, dest_path) except messages.S3ClientError as err: results.append((src_path, dest_path, err)) # Copy individual files with self.copy elif not src_path.endswith('/') and not dest_path.endswith('/'): src_bucket, src_key = parse_s3_path(src_path) dest_bucket, dest_key = parse_s3_path(dest_path) request = messages.CopyRequest( src_bucket, src_key, dest_bucket, dest_key) try: self.client.copy(request) results.append((src_path, dest_path, None)) except messages.S3ClientError as err: results.append((src_path, dest_path, err)) # Mismatched paths (one directory, one non-directory) get an error result else: e = messages.S3ClientError( "Can't copy mismatched paths (one directory, one non-directory):" + ' %s, %s' % (src_path, dest_path), 400) results.append((src_path, dest_path, e)) return results
# We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def copy_tree(self, src, dest): """Renames the given S3 directory and it's contents recursively from src to dest. Args: src: S3 file path pattern in the form s3://<bucket>/<name>/. dest: S3 file path pattern in the form s3://<bucket>/<name>/. Returns: List of tuples of (src, dest, exception) where exception is None if the operation succeeded or the relevant exception if the operation failed. """ assert src.endswith('/') assert dest.endswith('/') results = [] for entry in self.list_prefix(src): rel_path = entry[len(src):] try: self.copy(entry, dest + rel_path) results.append((entry, dest + rel_path, None)) except messages.S3ClientError as e: results.append((entry, dest + rel_path, e)) return results
[docs] @retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def delete(self, path): """Deletes a single S3 file object from src to dest. Args: src: S3 file path pattern in the form s3://<bucket>/<name>/. dest: S3 file path pattern in the form s3://<bucket>/<name>/. Returns: List of tuples of (src, dest, exception) in the same order as the src_dest_pairs argument, where exception is None if the operation succeeded or the relevant exception if the operation failed. """ bucket, object_path = parse_s3_path(path) request = messages.DeleteRequest(bucket, object_path) try: self.client.delete(request) except messages.S3ClientError as e: if e.code == 404: return # Same behavior as GCS - don't surface a 404 error else: logging.error('HTTP error while deleting file %s: %s', path, 3) raise e
# We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def delete_paths(self, paths): """Deletes the given S3 objects from src to dest. This can handle directory or file paths. Args: src: S3 file path pattern in the form s3://<bucket>/<name>/. dest: S3 file path pattern in the form s3://<bucket>/<name>/. Returns: List of tuples of (src, dest, exception) in the same order as the src_dest_pairs argument, where exception is None if the operation succeeded or the relevant exception if the operation failed. """ directories, not_directories = [], [] for path in paths: if path.endswith('/'): directories.append(path) else: not_directories.append(path) results = {} for directory in directories: dir_result = dict(self.delete_tree(directory)) results.update(dir_result) not_directory_results = dict(self.delete_files(not_directories)) results.update(not_directory_results) return results
# We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def delete_files(self, paths, max_batch_size=1000): """Deletes the given S3 file object from src to dest. Args: paths: List of S3 file paths in the form s3://<bucket>/<name> max_batch_size: Largest number of keys to send to the client to be deleted simultaneously Returns: List of tuples of (path, exception) in the same order as the paths argument, where exception is None if the operation succeeded or the relevant exception if the operation failed. """ if not paths: return [] # Sort paths into bucket: [keys] buckets, keys = zip(*[parse_s3_path(path) for path in paths]) grouped_keys = {bucket: [] for bucket in buckets} for bucket, key in zip(buckets, keys): grouped_keys[bucket].append(key) # For each bucket, delete minibatches of keys results = {} for bucket, keys in grouped_keys.items(): for i in range(0, len(keys), max_batch_size): minibatch_keys = keys[i:i + max_batch_size] results.update(self._delete_minibatch(bucket, minibatch_keys)) # Organize final results final_results = [(path, results[parse_s3_path(path)]) for path in paths] return final_results
@retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def _delete_minibatch(self, bucket, keys): """A helper method. Boto3 allows batch deletions for files within the same bucket. Args: bucket: String bucket name keys: List of keys to be deleted in the bucket Returns: dict of the form {(bucket, key): error}, where error is None if the operation succeeded """ request = messages.DeleteBatchRequest(bucket, keys) results = {} try: response = self.client.delete_batch(request) for key in response.deleted: results[(bucket, key)] = None for key, error in zip(response.failed, response.errors): results[(bucket, key)] = error except messages.S3ClientError as e: for key in keys: results[(bucket, key)] = e return results # We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def delete_tree(self, root): """Deletes all objects under the given S3 directory. Args: path: S3 root path in the form s3://<bucket>/<name>/ (ending with a "/") Returns: List of tuples of (path, exception), where each path is an object under the given root. exception is None if the operation succeeded or the relevant exception if the operation failed. """ assert root.endswith('/') paths = self.list_prefix(root) return self.delete_files(paths)
[docs] def size(self, path): """Returns the size of a single S3 object. This method does not perform glob expansion. Hence the given path must be for a single S3 object. Returns: size of the S3 object in bytes. """ return self._s3_object(path).size
# We intentionally do not decorate this method with a retry, since the # underlying copy and delete operations are already idempotent operations # protected by retry decorators.
[docs] def rename(self, src, dest): """Renames the given S3 object from src to dest. Args: src: S3 file path pattern in the form s3://<bucket>/<name>. dest: S3 file path pattern in the form s3://<bucket>/<name>. """ self.copy(src, dest) self.delete(src)
[docs] def last_updated(self, path): """Returns the last updated epoch time of a single S3 object. This method does not perform glob expansion. Hence the given path must be for a single S3 object. Returns: last updated time of the S3 object in second. """ return self._updated_to_seconds(self._s3_object(path).last_modified)
[docs] def exists(self, path): """Returns whether the given S3 object exists. Args: path: S3 file path pattern in the form s3://<bucket>/<name>. """ try: self._s3_object(path) return True except messages.S3ClientError as e: if e.code == 404: # HTTP 404 indicates that the file did not exist return False else: # We re-raise all other exceptions raise
def _status(self, path): """For internal use only; no backwards-compatibility guarantees. Returns supported fields (checksum, last_updated, size) of a single object as a dict at once. This method does not perform glob expansion. Hence the given path must be for a single S3 object. Returns: dict of fields of the S3 object. """ s3_object = self._s3_object(path) file_status = {} if hasattr(s3_object, 'etag'): file_status['checksum'] = s3_object.etag if hasattr(s3_object, 'last_modified'): file_status['last_updated'] = self._updated_to_seconds( s3_object.last_modified) if hasattr(s3_object, 'size'): file_status['size'] = s3_object.size return file_status @retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def _s3_object(self, path): """Returns a S3 object metadata for the given path This method does not perform glob expansion. Hence the given path must be for a single S3 object. Returns: S3 object metadata. """ bucket, object = parse_s3_path(path) request = messages.GetRequest(bucket, object) return self.client.get_object_metadata(request) @staticmethod def _updated_to_seconds(updated): """Helper function transform the updated field of response to seconds""" return ( time.mktime(updated.timetuple()) - time.timezone + updated.microsecond / 1000000.0)
[docs] def rename_files(self, src_dest_pairs): """Renames the given S3 objects from src to dest. Args: src_dest_pairs: list of (src, dest) tuples of s3://<bucket>/<name> file paths to rename from src to dest Returns: List of tuples of (src, dest, exception) in the same order as the src_dest_pairs argument, where exception is None if the operation succeeded or the relevant exception if the operation failed. """ if not src_dest_pairs: return [] # TODO: Throw value error if path has directory for src, dest in src_dest_pairs: if src.endswith('/') or dest.endswith('/'): raise ValueError('Cannot rename a directory') copy_results = self.copy_paths(src_dest_pairs) paths_to_delete = [src for (src, _, err) in copy_results if err is None] delete_results = self.delete_files(paths_to_delete) delete_results_dict = {src: err for (src, err) in delete_results} rename_results = [] for src, dest, err in copy_results: if err is not None: rename_results.append((src, dest, err)) elif delete_results_dict[src] is not None: rename_results.append((src, dest, delete_results_dict[src])) else: rename_results.append((src, dest, None)) return rename_results
[docs]class S3Downloader(Downloader): def __init__(self, client, path, buffer_size): self._client = client self._path = path self._bucket, self._name = parse_s3_path(path) self._buffer_size = buffer_size # Get object state. self._get_request = ( messages.GetRequest(bucket=self._bucket, object=self._name)) try: metadata = self._get_object_metadata(self._get_request) except messages.S3ClientError as e: if e.code == 404: raise IOError(errno.ENOENT, 'Not found: %s' % self._path) else: logging.error('HTTP error while requesting file %s: %s', self._path, 3) raise self._size = metadata.size @retry.with_exponential_backoff( retry_filter=retry.retry_on_server_errors_and_timeout_filter) def _get_object_metadata(self, get_request): return self._client.get_object_metadata(get_request) @property def size(self): return self._size
[docs] def get_range(self, start, end): return self._client.get_range(self._get_request, start, end)
[docs]class S3Uploader(Uploader): def __init__(self, client, path, mime_type='application/octet-stream'): self._client = client self._path = path self._bucket, self._name = parse_s3_path(path) self._mime_type = mime_type self.part_number = 1 self.buffer = b'' self.last_error = None self.upload_id = None self.parts = [] self._start_upload() # There is retry logic in the underlying transfer library but we should make # it more explicit so we can control the retry parameters. @retry.no_retries # Using no_retries marks this as an integration point. def _start_upload(self): # The uploader by default transfers data in chunks of 1024 * 1024 bytes at # a time, buffering writes until that size is reached. try: request = messages.UploadRequest( self._bucket, self._name, self._mime_type) response = self._client.create_multipart_upload(request) self.upload_id = response.upload_id except Exception as e: # pylint: disable=broad-except logging.error( 'Error in _start_upload while inserting file %s: %s', self._path, traceback.format_exc()) self.last_error = e raise e
[docs] def put(self, data): MIN_WRITE_SIZE = 5 * 1024 * 1024 MAX_WRITE_SIZE = 5 * 1024 * 1024 * 1024 # TODO: Byte strings might not be the most performant way to handle this self.buffer += data.tobytes() while len(self.buffer) >= MIN_WRITE_SIZE: # Take the first chunk off the buffer and write it to S3 chunk = self.buffer[:MAX_WRITE_SIZE] self._write_to_s3(chunk) # Remove the written chunk from the buffer self.buffer = self.buffer[MAX_WRITE_SIZE:]
def _write_to_s3(self, data): try: request = messages.UploadPartRequest( self._bucket, self._name, self.upload_id, self.part_number, data) response = self._client.upload_part(request) self.parts.append({ 'ETag': response.etag, 'PartNumber': response.part_number }) self.part_number = self.part_number + 1 except messages.S3ClientError as e: self.last_error = e if e.code == 404: raise IOError(errno.ENOENT, 'Not found: %s' % self._path) else: logging.error('HTTP error while requesting file %s: %s', self._path, 3) raise
[docs] def finish(self): if len(self.buffer) > 0: # writes with zero length or mid size files between # MIN_WRITE_SIZE = 5 * 1024 * 1024 # MAX_WRITE_SIZE = 5 * 1024 * 1024 * 1024 # as we will reach this finish() with len(self.buffer) == 0 # which will fail self._write_to_s3(self.buffer) if self.last_error is not None: raise self.last_error # pylint: disable=raising-bad-type request = messages.CompleteMultipartUploadRequest( self._bucket, self._name, self.upload_id, self.parts) self._client.complete_multipart_upload(request)