In the previous article<\/a>, we looked at how we can use the generic Blob APIs from jclouds to upload content to S3. In this article we will use the S3 specific asynchronous API from jclouds<\/b> to upload content and leverage the multipart upload functionality provided by S3<\/a>.<\/p>\n The first part of the upload process is creating the jclouds API \u2013 this is a custom API for Amazon S3:<\/p>\n Amazon S3 has a 5 MB limit for each part to be uploaded. As such, the first thing we need to do is determine the right number of parts that we can split our content into so that we don\u2019t have parts below this 5 MB limit:<\/p>\n Were going to break the byte array into a set number of parts:<\/p>\n We\u2019re going to test<\/b> the logic of breaking the byte array into parts \u2013 we\u2019re going to generate some bytes, split the byte array, recompose it back together using Guava and verify<\/b> that we get back the original:<\/p>\n To generate the data, we simply use the support from Random<\/em>: Now that we have determined the correct number of parts for our content and we managed to break the content into parts, we need to generate the Payload objects<\/b> for the jclouds API:<\/p>\n The upload process is a flexible multi-step process \u2013 this means:<\/p>\n The first step in the Upload operation is to initiate the process<\/b>. This request to S3 must contain the standard HTTP headers \u2013 the Content<\/em>–MD5<\/em> header in particular needs to be computed. Were going to use the Guava hash function support here:<\/p>\n This is the md5 hash<\/b> of the entire byte array, not of the parts yet.<\/p>\n To initiate the upload<\/b>, and for all further interactions with S3, we\u2019re going to use the AWSS3AsyncClient \u2013 the asynchronous API we created earlier:<\/p>\n The key<\/b><\/em> is the handle assigned to the object \u2013 this needs to be a unique identifier specified by the client.<\/p>\n Also notice that, even though we\u2019re using the async version of the API, we\u2019re blocking<\/b> for the result of this operation \u2013 this is because we will need the result of the initialize to be able to move forward.<\/p>\n The result of the operation is an upload id<\/b> returned by S3 \u2013 this will identify the upload throughout it\u2019s lifecycle and will be present in all subsequent upload operations.<\/p>\n The next step is uploading the parts<\/b>. Our goal here is to send these requests in parallel<\/b>, as the upload parts operation represent the bulk of the upload process:<\/p>\n The part numbers need to be continuous but the order in which the requests are send is not relevant.<\/p>\n After all of the upload part requests have been submitted, we need to wait for their responses <\/b>so that we can collect the individual ETag value of each part:<\/p>\n If, for whatever reason, one of the upload part operations fails, the operation can be retried<\/b> until it succeeds. The logic above does not contain the retry mechanism, but building it in should be straightforward enough.<\/p>\n The final step of the upload process is completing the multipart operation<\/b>. The S3 API requires the responses from the previous parts upload as a Map<\/em>, which we can now easily create from the list of ETags that we obtained above:<\/p>\n And finally, send the complete request:<\/p>\n This will return final ETag of the finished object and will complete the entire upload process.<\/p>\n In this article we built a multipart enabled, fully parallel upload operation to S3, using the custom S3 jclouds API. This operation is ready to be used as is, but it can be improved<\/b> in a few ways. First, retry logic<\/b> should be added around the upload operations to better deal with failures. Next, for really large files, even though the mechanism is sending all upload multipart requests in parallel, a throttling mechanism<\/b> should still limit the number of parallel requests being sent. This is both to avoid bandwidth becoming a bottleneck as well as to make sure Amazon itself doesn\u2019t flag the upload process as exceeding an allowed limit of requests per second \u2013 the Guava RateLimiter<\/a> can potentially be very well suited for this. 1. Goal In the previous article, we looked at how we can use the generic Blob APIs from jclouds to upload content to S3. In this article we will use the S3 specific asynchronous API from jclouds to upload content and leverage the multipart upload functionality provided by S3. 2. Preparation 2.1. Set …<\/p>\n","protected":false},"author":104,"featured_media":112,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[8],"tags":[331,210],"class_list":["post-10930","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-enterprise-java","tag-amazon-aws","tag-cloud"],"yoast_head":"\n2. Preparation<\/h2>\n
\n <\/p>\n2.1. Set up the custom API<\/h4>\n
public AWSS3AsyncClient s3AsyncClient() {\r\n String identity = ...\r\n String credentials = ...\r\n BlobStoreContext context = ContextBuilder.newBuilder('aws-s3').\r\n credentials(identity, credentials).buildView(BlobStoreContext.class);\r\n RestContext<AWSS3Client, AWSS3AsyncClient> providerContext = context.unwrap();\r\n return providerContext.getAsyncApi();\r\n}<\/pre>\n2.2. Determining the number of parts for the content<\/h4>\n
public static int getMaximumNumberOfParts(byte[] byteArray) {\r\n int numberOfParts= byteArray.length \/ fiveMB; \/\/ 5*1024*1024\r\n if (numberOfParts== 0) {\r\n return 1;\r\n }\r\n return numberOfParts;\r\n}<\/pre>\n2.3. Breaking the content into parts<\/h4>\n
public static List<byte[]> breakByteArrayIntoParts(byte[] byteArray, int maxNumberOfParts) {\r\n List<byte[]> parts = Lists.<byte[]> newArrayListWithCapacity(maxNumberOfParts);\r\n int fullSize = byteArray.length;\r\n long dimensionOfPart = fullSize \/ maxNumberOfParts;\r\n for (int i = 0; i < maxNumberOfParts; i++) {\r\n int previousSplitPoint = (int) (dimensionOfPart * i);\r\n int splitPoint = (int) (dimensionOfPart * (i + 1));\r\n if (i == (maxNumberOfParts - 1)) {\r\n splitPoint = fullSize;\r\n }\r\n byte[] partBytes = Arrays.copyOfRange(byteArray, previousSplitPoint, splitPoint);\r\n parts.add(partBytes);\r\n }\r\n return parts;\r\n}<\/pre>\n@Test\r\npublic void given16MByteArray_whenFileBytesAreSplitInto3_thenTheSplitIsCorrect() {\r\n byte[] byteArray = randomByteData(16);\r\n int maximumNumberOfParts = S3Util.getMaximumNumberOfParts(byteArray);\r\n List<byte[]> fileParts = S3Util.breakByteArrayIntoParts(byteArray, maximumNumberOfParts);\r\n assertThat(fileParts.get(0).length + fileParts.get(1).length + fileParts.get(2).length,\r\n equalTo(byteArray.length));\r\n byte[] unmultiplexed = Bytes.concat(fileParts.get(0), fileParts.get(1), fileParts.get(2));\r\n assertThat(byteArray, equalTo(unmultiplexed));\r\n}<\/pre>\nbyte[] randomByteData(int mb) {\r\n byte[] randomBytes = new byte[mb * 1024 * 1024];\r\n new Random().nextBytes(randomBytes);\r\n return randomBytes;\r\n}<\/pre>\n2.4. Creating the Payloads<\/h4>\n
public static List<Payload> createPayloadsOutOfParts(Iterable<byte[]> fileParts) {\r\n List<Payload> payloads = Lists.newArrayList();\r\n for (byte[] filePart : fileParts) {\r\n byte[] partMd5Bytes = Hashing.md5().hashBytes(filePart).asBytes();\r\n Payload partPayload = Payloads.newByteArrayPayload(filePart);\r\n partPayload.getContentMetadata().setContentLength((long) filePart.length);\r\n partPayload.getContentMetadata().setContentMD5(partMd5Bytes);\r\n payloads.add(partPayload);\r\n }\r\n return payloads;\r\n}<\/pre>\n3. Upload<\/h2>\n
\n
3.1. Initiating the Upload operation<\/h4>\n
Hashing.md5().hashBytes(byteArray).asBytes();<\/pre>\n
ObjectMetadata metadata = ObjectMetadataBuilder.create().key(key).contentMD5(md5Bytes).build();\r\nString uploadId = s3AsyncApi.initiateMultipartUpload(container, metadata).get();<\/pre>\n
3.2. Uploading the Parts<\/h4>\n
List<ListenableFuture<String>> ongoingOperations = Lists.newArrayList();\r\nfor (int partNumber = 0; partNumber < filePartsAsByteArrays.size(); partNumber++) {\r\n ListenableFuture<String> future = s3AsyncApi.uploadPart(\r\n container, key, partNumber + 1, uploadId, payloads.get(partNumber));\r\n ongoingOperations.add(future);\r\n}<\/pre>\nFunction<ListenableFuture<String>, String> getEtagFromOp =\r\n new Function<ListenableFuture<String>, String>() {\r\n public String apply(ListenableFuture<String> ongoingOperation) {\r\n try {\r\n return ongoingOperation.get();\r\n } catch (InterruptedException | ExecutionException e) {\r\n throw new IllegalStateException(e);\r\n }\r\n }\r\n};\r\nList<String> etagsOfParts = Lists.transform(ongoingOperations, getEtagFromOp);<\/pre>\n3.3. Completing the Upload operation<\/h4>\n
Map<Integer, String> parts = Maps.newHashMap();\r\nfor (int i = 0; i < etagsOfParts.size(); i++) {\r\n parts.put(i + 1, etagsOfParts.get(i));\r\n}<\/pre>\ns3AsyncApi.completeMultipartUpload(container, key, uploadId, parts).get();<\/pre>\n
4. Conclusion<\/h2>\n
\n <\/p>\n