APEXMALHAR-2022 Developed S3 Output Module #483

chaithu14 · 2016-11-04T12:49:25Z

No description provided.

bhupeshchawda · 2016-11-09T05:08:57Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+    if (currentWindowId <= windowDataManager.getLargestCompletedWindow()) {
+      return;
+    }
+    S3BlockMetaData metaData = blockInfo.get(tuple.getBlockId());


Having just blockId as the key here might cause problems since the way block ids are computed, blocks in different files may have the same block id. I'll suggest to have something like filename as well, as part of the key.

Yes. You are right. I made the changes as you suggested.

bhupeshchawda · 2016-11-09T05:10:51Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+    }
+    processedBlocks.add(tuple.getBlockId());
+    long partSize = tuple.getRecord().length;
+    PartETag pt = null;


rename variable => partETag?

bhupeshchawda · 2016-11-09T06:15:00Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+   * Send the CompleteMultipartUploadRequest to S3 if all the blocks of a file are uploaded into S3.
+   * @param keyName file to upload into S3
+   */
+  private void emitFileMerge(String keyName)


Name seems misleading. Can we rename it to something like: verifyAndEmitFileMerge ?

bhupeshchawda · 2016-11-09T06:25:23Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+      return;
+    }
+    if (currentWindowId <= windowDataManager.getLargestCompletedWindow()) {
+      uploadedFiles.add(keyName);


Files can get uploaded across windows. We cannot assume this keyName is uploaded.

Yes. I fixed this.

bhupeshchawda · 2016-11-09T06:26:10Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+        uploadFileMetadata.getFileMetadata().getNumberOfBlocks() != partETags.size()) {
+      return;
+    }
+    if (currentWindowId <= windowDataManager.getLargestCompletedWindow()) {


During recovery, no need to do anything as the window is already processed.
We can move this if block to the beginning of the method.

bhupeshchawda · 2016-11-09T06:28:58Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+  }
+
+  @Override
+  public void committed(long windowId)


Can we save the uploadedFiles in the WAL every window? This will help us to clear up uploadParts and fileMetaDatas in the end window.

I think you missed this..

No. I made the changes.

bhupeshchawda · 2016-11-09T06:33:16Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+    for (int i = 0; i < blocks.length; i++) {
+      blockInfo.put(blocks[i], new S3BlockMetaData(tuple.getKeyName(), tuple.getUploadId(), i + 1));
+    }
+    if (blocks.length < 1) {


Refactor?
if (blocks.length > 0) { blockInfo.get(blocks[blocks.length - 1]).setLastBlock(true); }

bhupeshchawda · 2016-11-09T06:35:08Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+  protected transient AmazonS3 s3Client;
+  private transient Set<Long> processedBlocks;
+  private transient long currentWindowId;
+  private transient List<AbstractBlockReader.ReaderRecord<Slice>> recordData;


Rename to waitingTuples ?

bhupeshchawda · 2016-11-09T07:01:53Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+  /**
+   * Process the blocks which are in wait state.
+   */
+  private void processWaitBlocks()


This method seems idempotent. Can we call this during handleIdleTime() as well? This may shorten the time spent in endWindow().

Blocks has to wait till it receives the file meta data. I added this method call after receives the file meta data of the waited block.

I still think we should add it to handleIdleTime() and remove it from the processUploadFileMetadata() call.
Most of the times, it will have the meta data before it receives the actual data right? So only in some cases will we have the blocks waiting for meta data. Better to do it in the handleIdleTime and endWindow calls than blocking the operator thread to process waiting blocks.

bhupeshchawda · 2016-11-09T08:03:29Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+    if (processedBlocks.contains(tuple.getBlockId())) {
+      return;
+    }
+    processedBlocks.add(tuple.getBlockId());


Why do we need processedBlocks. Can't we directly make changes to blockInfo?

Yes we can do. I added this because of there is an issue in FSInputModule that the same block id is emitting. PR for this fix is merged. I removed the "processedBlocks" as it is not needed.

bhupeshchawda

I have added a few more comments. Please check.

bhupeshchawda · 2016-11-22T07:22:58Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3InitiateFileUpload.java

+   * Creates the empty object metadata for initiate multipart upload request.
+   * @return the ObjectMetadata
+   */
+  public ObjectMetadata createObjectMetadata()


This can be private?

No. By default it creates empty objectMetadata. User can override this and set the properties like encryption algorithm, etc. For more details, please refer below link:
http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/model/ObjectMetadata.html

bhupeshchawda · 2016-11-22T09:13:24Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.apex.malhar.lib.fs.s3output;


Can we move this to package: org.apache.apex.malhar.lib.io.fs.s3

@chaithu14 should we have a separate maven module for this similar to kafka?
@tweise Suggestions?

bhupeshchawda · 2016-11-22T09:29:33Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+  private String secretAccessKey;
+  private String endPoint;
+  private Map<String, S3BlockMetaData> blockInfo = new HashMap<>();
+  private Map<Long, String> blockId2FilePath = new HashMap<>();


Rename to blockIdToFilePath?

The purpose of this seems to be for tracking whether the block has been received from the ``BlockMetaDatainput port. If so, can we change it to aSet``` instead of a ```Map```?

No. We need filepath of the received block to construct unique block id.

bhupeshchawda · 2016-11-22T09:30:57Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+  public final transient DefaultOutputPort<UploadBlockMetadata> output = new DefaultOutputPort<>();
+
+  /**
+   * This input port receives incoming tuples.


incoming tuples. => incoming tuples (Block Data)

bhupeshchawda · 2016-11-22T09:47:32Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+   * @param tuple UploadFileMetadata
+   */
+  protected void processUploadFileMetadata(S3InitiateFileUpload.UploadFileMetadata tuple)
+  {


Can we skip processing this if currentWindowId <= windowDataManager.getLargestCompletedWindow() ? Similar to other process calls..

This constructs the blockInfo Map and clearing this in replay call. So, I think it's better to process even if currentWindowId <= windowDataManager.getLargestCompletedWindow().

bhupeshchawda · 2016-11-22T10:31:13Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+    if (partETags.size() <= 1) {
+      uploadedFiles.add(keyName);
+      currentWindowRecoveryState.add(keyName);
+      return;


LOG.debug("Uploaded file {} successfully", keyName);

bhupeshchawda · 2016-11-22T10:33:00Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+   */
+  protected void processUploadBlock(S3BlockUpload.UploadBlockMetadata tuple)
+  {
+    List<PartETag> listOfUploads = uploadParts.get(tuple.getKeyName());


Skip processing if windowId <= windowDataManager.getLargestCompletedWindow()

No. Need to process this because I am not saving the uploadParts in WAL.

bhupeshchawda · 2016-11-22T10:33:31Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+   */
+  protected void processFileMetadata(S3InitiateFileUpload.UploadFileMetadata tuple)
+  {
+    String keyName = tuple.getKeyName();


Skip processing if windowId <= windowDataManager.getLargestCompletedWindow()

bhupeshchawda · 2016-11-22T10:35:29Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+  }
+
+  @Override
+  public void committed(long windowId)


I think you missed this..

bhupeshchawda · 2016-11-22T10:47:02Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3InitiateUploadTest.java

@@ -0,0 +1,115 @@
+/**


The tests seem to be excluded from the build.
Can you try looking for some embedded S3 server for testing? Or write mock tests which simulate the behavior of S3 in order to have the unit testing?

Added mock test

bhupeshchawda · 2016-11-23T11:58:28Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+  @NotNull
+  private String secretAccessKey;
+  private String endPoint;
+  protected List<String> uploadedFiles = new ArrayList<>();


Can this e transient?

bhupeshchawda · 2016-11-25T05:33:51Z

library/pom.xml

@@ -193,6 +193,7 @@
        <configuration>
          <excludes>
            <exclude>**/S3InputModuleAppTest.java</exclude>
+            <exclude>**/S3OutputModuleAppTest.java</exclude>


Since we have the mock tests now, we can remove this test as it will not run anyway.

bhupeshchawda · 2016-11-25T05:36:17Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3InitiateUploadTest.java

+  private static final String APPLICATION_PATH_PREFIX = "target/s3outputtest/";
+  private String applicationPath;
+  private Attribute.AttributeMap.DefaultAttributeMap attributes;
+  Context.OperatorContext context;


bhupeshchawda · 2016-11-25T05:44:01Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3OutputModuleMockTest.java

+    PutObjectResult objResult = new PutObjectResult();
+    objResult.setETag("SuccessFullyUploaded");
+
+    UploadPartResult partResult = new UploadPartResult();


can you try it out using multiple parts?

This might be difficult to be done. Because it depends on UploadPartRequest and this is the passed argument to client.uploadPart().

bhupeshchawda · 2016-11-25T05:48:25Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3OutputModuleMockTest.java

+    LocalMode.Controller lc = lma.getController();
+    lc.setHeartbeatMonitoringEnabled(true);
+    lc.runAsync();
+


Try using setExitCondition() as in https://github.com/apache/apex-malhar/blob/master/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java.

bhupeshchawda · 2016-11-25T05:49:47Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3OutputModuleTest.java

@@ -0,0 +1,54 @@
+package org.apache.apex.malhar.lib.fs.s3output;


bhupeshchawda · 2016-11-25T05:50:08Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3OutputModuleMockTest.java

+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.when;
+
+public class S3OutputModuleMockTest


some docs on what is being validated

bhupeshchawda · 2016-11-25T05:50:30Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3output/S3InitiateUploadTest.java

+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.when;
+
+public class S3InitiateUploadTest


some docs on what is being validated

bhupeshchawda · 2016-11-25T08:44:59Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3FileMerger.java

+   */
+  private void verifyAndEmitFileMerge(String keyName)
+  {
+    if (currentWindowId <= windowDataManager.getLargestCompletedWindow()) {


Can this be moved to the processFileMetadata and processBlockUpload methods?

No. This can't be moved to processFileMetadata and processBlockUpload methods. Because I am not saving the uploadParts and fileMetadatas in WAL.

bhupeshchawda · 2016-11-25T08:46:29Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3output/S3BlockUpload.java

+/**
+ * This operator can be used to upload the block into S3 bucket using multi-part feature or putObject API.
+ * Upload the block into S3 using multi-part feature only if the number of blocks of a file is > 1.
+ */


Better to mention that this is useful in context of the S3 Output Module. Same for all operators in this module.

bhupeshchawda · 2016-11-25T08:49:06Z

Also check the build - some rat check failure..

bhupeshchawda

Added a couple of minor comments...

bhupeshchawda · 2016-11-28T08:32:52Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModuleTest.java

+
+import static org.apache.apex.malhar.lib.fs.s3.S3OutputModuleMockTest.client;
+
+public class S3OutputModuleTest extends S3OutputModule


Nit: Can you rename this class to S3OutputTestModule?
Otherwise this seems to be a test class without any tests :-)

bhupeshchawda · 2016-11-28T08:36:28Z

library/src/test/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModuleMockTest.java

+    });
+    lc.run(10000);
+
+    lc.shutdown();


No need of shutdown(); lc.run() will shutdown automatically.

Ok. Updated

bhupeshchawda

Changes look good to me...Will merge soon if there are no other comments.

yogidevendra · 2016-11-29T05:57:45Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3BlockUpload.java

+ */
+
+@InterfaceStability.Evolving
+public class S3BlockUpload implements Operator, Operator.CheckpointNotificationListener, Operator.IdleTimeHandler


Could you please change file name to S3BlockUploadOperator or S3BlockUploader?

Changed to S3BlockUploadOperator.

yogidevendra · 2016-11-29T06:00:08Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3InitiateFileUpload.java

+ * This operator is useful in context of S3 Output Module.
+ */
+@InterfaceStability.Evolving
+public class S3InitiateFileUpload implements Operator, Operator.CheckpointNotificationListener


Could you please change filename to S3FileUploadInitiator or S3InitiateFileUploadOperator?

Changed to S3InitiateFileUploadOperator.

yogidevendra · 2016-11-29T06:06:04Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModule.java

+ *  - S3BlockUpload
+ *  - S3FileMerger
+ *
+ * Initial BenchMark Results


This will be very useful. Excellent.

yogidevendra · 2016-11-29T06:09:14Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModule.java

+   * AWS access key
+   */
+  @NotNull
+  private String accessKey;


Could we make properties for S3OutputModule consistent with S3InputModule.
Either we should take accessKey, secretAccessKey seperately for both the modules.
OR take URL input for both the modules.

There could be discussion around pros, cons of both the approaches. Thus, you might tackle it as separate JIRA.

Ok. I will create the JIRA for the same.

Request you to put link to new JIRA in reply to this comment.

Created JIRA for this task: https://issues.apache.org/jira/browse/APEXMALHAR-2360

yogidevendra · 2016-11-29T06:11:33Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModule.java

+ *  @tags S3, Output
+ */
+@InterfaceStability.Evolving
+public class S3OutputModule implements Module


I would recommend to move this as a separate pom project under malhar. Just like we have malhar-kafka, malhar-hive

I added this under malhar library because the existing S3InputModule is under malhar library.

yogidevendra · 2016-11-29T06:14:24Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3OutputModule.java

+   * Creates the number of instances of S3FileMerger operator.
+   */
+  @Min(1)
+  private int noOfFileMergers = 1;


in the input module we have property called readerCount.
To make it consistent, we should rename this property to mergerCount.

yogidevendra · 2016-11-29T08:11:40Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3BlockUpload.java

+        return;
+      }
+      for (Map.Entry<String, UploadBlockMetadata> ubm: recoveredData.entrySet()) {
+        output.emit(ubm.getValue());


Avoid abbreviations in the variable names.

yogidevendra · 2016-11-29T08:11:49Z

library/src/main/java/org/apache/apex/malhar/lib/fs/s3/S3BlockUpload.java

+   */
+  private void processWaitBlocks()
+  {
+    Iterator<AbstractBlockReader.ReaderRecord<Slice>> ite = waitingTuples.iterator();


Avoid abbreviations in the variable names.

yogidevendra · 2016-11-30T05:39:42Z

Changes looks good to me.

bhupeshchawda reviewed Nov 9, 2016

View reviewed changes

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch 2 times, most recently from a77a280 to dc945a4 Compare November 21, 2016 10:37

bhupeshchawda reviewed Nov 22, 2016

View reviewed changes

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch from dc945a4 to 3fb799f Compare November 23, 2016 13:06

bhupeshchawda reviewed Nov 25, 2016

View reviewed changes

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch 3 times, most recently from b57099e to 91bb9f6 Compare November 28, 2016 05:17

bhupeshchawda reviewed Nov 28, 2016

View reviewed changes

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch from 91bb9f6 to 6ab63bd Compare November 28, 2016 08:55

chaithu14 closed this Nov 28, 2016

chaithu14 reopened this Nov 28, 2016

bhupeshchawda approved these changes Nov 28, 2016

View reviewed changes

yogidevendra suggested changes Nov 29, 2016

View reviewed changes

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch 3 times, most recently from 376d1eb to cb28681 Compare November 30, 2016 05:17

APEXMALHAR-2022 Development of S3 Output Module

a5e8fa3

chaithu14 force-pushed the APEXMALHAR-2022-S3Output-multiPart branch from cb28681 to a5e8fa3 Compare November 30, 2016 11:03

chaithu14 closed this Nov 30, 2016

chaithu14 reopened this Nov 30, 2016

chaithu14 closed this Nov 30, 2016

chaithu14 reopened this Nov 30, 2016

asfgit merged commit a5e8fa3 into apache:master Dec 1, 2016

		@@ -0,0 +1,54 @@
		package org.apache.apex.malhar.lib.fs.s3output;


		import static org.apache.apex.malhar.lib.fs.s3.S3OutputModuleMockTest.client;

		public class S3OutputModuleTest extends S3OutputModule

APEXMALHAR-2022 Developed S3 Output Module #483

APEXMALHAR-2022 Developed S3 Output Module #483

Conversation

chaithu14 commented Nov 4, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bhupeshchawda left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bhupeshchawda commented Nov 25, 2016

bhupeshchawda left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bhupeshchawda left a comment

Choose a reason for hiding this comment