You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by yogidevendra <gi...@git.apache.org> on 2016/03/17 14:43:12 UTC
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
GitHub user yogidevendra opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
commit 8f597c91e135a7a6ec0a6857a0d43827238d602c
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-03-04T07:12:17Z
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators BlockWriter, Synchronizer, FileStitcher,
FileMerger, HDFSFileMerger
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
GitHub user yogidevendra reopened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
GitHub user yogidevendra reopened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
commit 622d8393ed4ab421e88d37bd054e8a0979c1b09e
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-03-04T07:12:17Z
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators BlockWriter, Synchronizer, FileStitcher,
FileMerger, HDFSFileMerger
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
4. Incorporating review comments
5. Setting block threshold same as number of readers
6. Adding limit on mvn memory usage.
commit 0c5494f203af73bc1b76fc54246fc173fe10a468
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-04-19T11:23:58Z
Adding max violations for benchmark
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56624309
--- Diff: apps/filecopy/src/main/java/com/datatorrent/apps/copy/HDFSFileCopySampleApp.java ---
@@ -0,0 +1,50 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.apps.copy;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DAG.Locality;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.lib.io.fs.HDFSFileCopyModule;
+import com.datatorrent.lib.io.fs.HDFSInputModule;
+
+/**
+ * Application for HDFS to HDFS file copy
+ */
+@ApplicationAnnotation(name = "HDFSFileCopySampleApp")
+public class HDFSFileCopySampleApp implements StreamingApplication
--- End diff --
should class name be SampleApp ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56982569
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
+
+ private long bytesWritten;
+ private double windowTimeSec;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ windowTimeSec = (context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)
+ * context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) * 1.0) / 1000.0;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ bytesWrittenPerSec = 0;
+ bytesWritten = 0;
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ OutputFileMetadata outputFileMetadata;
--- End diff --
using super implementation for endWindow(). No need to override super implementation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56949559
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
+
+ @Override
+ public Collection<Partition<BlockWriter>> definePartitions(Collection<Partition<BlockWriter>> partitions,
+ PartitioningContext context)
+ {
+ if (context.getParallelPartitionCount() == 0) {
+ return partitions;
+ }
+
+ // if there is no change of count, return the same collection
+ if (context.getParallelPartitionCount() == partitions.size()) {
+ LOG.debug("no change is partition count: " + partitions.size());
+ return partitions;
+ }
+
+ List<BasicCounters<MutableLong>> deletedCounters = Lists.newArrayList();
+
+ LOG.debug("block writer parallel partition count {}", context.getParallelPartitionCount());
+ int morePartitionsToCreate = context.getParallelPartitionCount() - partitions.size();
+ if (morePartitionsToCreate < 0) {
+ //Delete partitions
+ Iterator<Partition<BlockWriter>> partitionIterator = partitions.iterator();
+
+ while (morePartitionsToCreate++ < 0) {
+ Partition<BlockWriter> toRemove = partitionIterator.next();
+ deletedCounters.add(toRemove.getPartitionedInstance().fileCounters);
+ partitionIterator.remove();
+ }
+ } else {
+ //Add more partitions
+ BlockWriter anOperator = partitions.iterator().next().getPartitionedInstance();
+
+ while (morePartitionsToCreate-- > 0) {
+ DefaultPartition<BlockWriter> partition = new DefaultPartition<BlockWriter>(anOperator);
+ partitions.add(partition);
+ }
+ }
+
+ //transfer the counters
+ BlockWriter targetWriter = partitions.iterator().next().getPartitionedInstance();
--- End diff --
Can we change baseclass too?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56628725
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileCopyModule.java ---
@@ -0,0 +1,123 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import javax.validation.constraints.NotNull;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Module;
+import com.datatorrent.lib.io.block.AbstractBlockReader.ReaderRecord;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * HDFS file copy module can be used in conjunction with file input modules to
+ * copy files from any file system to HDFS. This module supports parallel write
+ * to multiple blocks of the same file and then stitching those blocks in
+ * original sequence.
+ *
+ * Essential operators are wrapped into single component using Module API.
+ *
+ */
+public class HDFSFileCopyModule implements Module
+{
+
+ /**
+ * Path of the output directory. Relative path of the files copied will be
+ * maintained w.r.t. source directory and output directory
+ */
+ @NotNull
+ protected String outputDirectoryPath;
+
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict;
+
+ /**
+ * Input port for files metadata.
+ */
+ public final transient ProxyInputPort<FileMetadata> filesMetadataInput = new ProxyInputPort<FileMetadata>();
+
+ /**
+ * Input port for blocks metadata
+ */
+ public final transient ProxyInputPort<BlockMetadata.FileBlockMetadata> blocksMetadataInput = new ProxyInputPort<BlockMetadata.FileBlockMetadata>();
+
+ /**
+ * Input port for blocks data
+ */
+ public final transient ProxyInputPort<ReaderRecord<Slice>> blockData = new ProxyInputPort<ReaderRecord<Slice>>();
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+
+ //Defining DAG
+ BlockWriter blockWriter = dag.addOperator("BlockWriter", new BlockWriter());
+ Synchronizer synchronizer = dag.addOperator("BlockSynchronizer", new Synchronizer());
+
+ dag.setInputPortAttribute(blockWriter.input, PortContext.PARTITION_PARALLEL, true);
+ dag.setInputPortAttribute(blockWriter.blockMetadataInput, PortContext.PARTITION_PARALLEL, true);
+ dag.addStream("CompletedBlockmetadata", blockWriter.blockMetadataOutput, synchronizer.blocksMetadataInput);
+
+ HDFSFileMerger merger = new HDFSFileMerger();
+ merger = dag.addOperator("FileMerger", merger);
+ dag.addStream("MergeTrigger", synchronizer.trigger, merger.input);
+
+ //DevNull<OutputFileMetadata> devNull1 = dag.addOperator("devNull1", DevNull.class);
+ //dag.addStream("ignored", merger.completedFilesMetaOutput, devNull1.data);
+
+ //Setting operator properties
+ merger.setFilePath(outputDirectoryPath);
+ merger.setOverwriteOnConflict(overwriteOnConflict);
+
+ //Binding proxy ports
+ filesMetadataInput.set(synchronizer.filesMetadataInput);
+ blocksMetadataInput.set(blockWriter.blockMetadataInput);
+ blockData.set(blockWriter.input);
+
+ }
+
+ public String getOutputDirectoryPath()
--- End diff --
javadocs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56784713
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
--- End diff --
moved at the end.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r57190169
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
+
+ @Override
+ public Collection<Partition<BlockWriter>> definePartitions(Collection<Partition<BlockWriter>> partitions,
+ PartitioningContext context)
+ {
+ if (context.getParallelPartitionCount() == 0) {
+ return partitions;
+ }
+
+ // if there is no change of count, return the same collection
+ if (context.getParallelPartitionCount() == partitions.size()) {
+ LOG.debug("no change is partition count: " + partitions.size());
+ return partitions;
+ }
+
+ List<BasicCounters<MutableLong>> deletedCounters = Lists.newArrayList();
+
+ LOG.debug("block writer parallel partition count {}", context.getParallelPartitionCount());
+ int morePartitionsToCreate = context.getParallelPartitionCount() - partitions.size();
+ if (morePartitionsToCreate < 0) {
+ //Delete partitions
+ Iterator<Partition<BlockWriter>> partitionIterator = partitions.iterator();
+
+ while (morePartitionsToCreate++ < 0) {
+ Partition<BlockWriter> toRemove = partitionIterator.next();
+ deletedCounters.add(toRemove.getPartitionedInstance().fileCounters);
+ partitionIterator.remove();
+ }
+ } else {
+ //Add more partitions
+ BlockWriter anOperator = partitions.iterator().next().getPartitionedInstance();
+
+ while (morePartitionsToCreate-- > 0) {
+ DefaultPartition<BlockWriter> partition = new DefaultPartition<BlockWriter>(anOperator);
+ partitions.add(partition);
+ }
+ }
+
+ //transfer the counters
+ BlockWriter targetWriter = partitions.iterator().next().getPartitionedInstance();
--- End diff --
Created https://issues.apache.org/jira/browse/APEXMALHAR-2024 for this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56628702
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileCopyModule.java ---
@@ -0,0 +1,123 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import javax.validation.constraints.NotNull;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Module;
+import com.datatorrent.lib.io.block.AbstractBlockReader.ReaderRecord;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * HDFS file copy module can be used in conjunction with file input modules to
+ * copy files from any file system to HDFS. This module supports parallel write
+ * to multiple blocks of the same file and then stitching those blocks in
+ * original sequence.
+ *
+ * Essential operators are wrapped into single component using Module API.
+ *
+ */
+public class HDFSFileCopyModule implements Module
+{
+
+ /**
+ * Path of the output directory. Relative path of the files copied will be
+ * maintained w.r.t. source directory and output directory
+ */
+ @NotNull
+ protected String outputDirectoryPath;
+
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict;
+
+ /**
+ * Input port for files metadata.
+ */
+ public final transient ProxyInputPort<FileMetadata> filesMetadataInput = new ProxyInputPort<FileMetadata>();
+
+ /**
+ * Input port for blocks metadata
+ */
+ public final transient ProxyInputPort<BlockMetadata.FileBlockMetadata> blocksMetadataInput = new ProxyInputPort<BlockMetadata.FileBlockMetadata>();
+
+ /**
+ * Input port for blocks data
+ */
+ public final transient ProxyInputPort<ReaderRecord<Slice>> blockData = new ProxyInputPort<ReaderRecord<Slice>>();
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+
+ //Defining DAG
+ BlockWriter blockWriter = dag.addOperator("BlockWriter", new BlockWriter());
+ Synchronizer synchronizer = dag.addOperator("BlockSynchronizer", new Synchronizer());
+
+ dag.setInputPortAttribute(blockWriter.input, PortContext.PARTITION_PARALLEL, true);
+ dag.setInputPortAttribute(blockWriter.blockMetadataInput, PortContext.PARTITION_PARALLEL, true);
+ dag.addStream("CompletedBlockmetadata", blockWriter.blockMetadataOutput, synchronizer.blocksMetadataInput);
+
+ HDFSFileMerger merger = new HDFSFileMerger();
+ merger = dag.addOperator("FileMerger", merger);
+ dag.addStream("MergeTrigger", synchronizer.trigger, merger.input);
+
+ //DevNull<OutputFileMetadata> devNull1 = dag.addOperator("devNull1", DevNull.class);
+ //dag.addStream("ignored", merger.completedFilesMetaOutput, devNull1.data);
+
+ //Setting operator properties
+ merger.setFilePath(outputDirectoryPath);
+ merger.setOverwriteOnConflict(overwriteOnConflict);
--- End diff --
java default is false, and I guess we expect default in feature as true, so either use same default in module, or add some checks to see if user has provided this config.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56786263
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileCopyModule.java ---
@@ -0,0 +1,123 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import javax.validation.constraints.NotNull;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Module;
+import com.datatorrent.lib.io.block.AbstractBlockReader.ReaderRecord;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * HDFS file copy module can be used in conjunction with file input modules to
+ * copy files from any file system to HDFS. This module supports parallel write
+ * to multiple blocks of the same file and then stitching those blocks in
+ * original sequence.
+ *
+ * Essential operators are wrapped into single component using Module API.
+ *
+ */
+public class HDFSFileCopyModule implements Module
+{
+
+ /**
+ * Path of the output directory. Relative path of the files copied will be
+ * maintained w.r.t. source directory and output directory
+ */
+ @NotNull
+ protected String outputDirectoryPath;
+
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict;
+
+ /**
+ * Input port for files metadata.
+ */
+ public final transient ProxyInputPort<FileMetadata> filesMetadataInput = new ProxyInputPort<FileMetadata>();
+
+ /**
+ * Input port for blocks metadata
+ */
+ public final transient ProxyInputPort<BlockMetadata.FileBlockMetadata> blocksMetadataInput = new ProxyInputPort<BlockMetadata.FileBlockMetadata>();
+
+ /**
+ * Input port for blocks data
+ */
+ public final transient ProxyInputPort<ReaderRecord<Slice>> blockData = new ProxyInputPort<ReaderRecord<Slice>>();
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+
+ //Defining DAG
+ BlockWriter blockWriter = dag.addOperator("BlockWriter", new BlockWriter());
+ Synchronizer synchronizer = dag.addOperator("BlockSynchronizer", new Synchronizer());
+
+ dag.setInputPortAttribute(blockWriter.input, PortContext.PARTITION_PARALLEL, true);
+ dag.setInputPortAttribute(blockWriter.blockMetadataInput, PortContext.PARTITION_PARALLEL, true);
+ dag.addStream("CompletedBlockmetadata", blockWriter.blockMetadataOutput, synchronizer.blocksMetadataInput);
+
+ HDFSFileMerger merger = new HDFSFileMerger();
+ merger = dag.addOperator("FileMerger", merger);
+ dag.addStream("MergeTrigger", synchronizer.trigger, merger.input);
+
+ //DevNull<OutputFileMetadata> devNull1 = dag.addOperator("devNull1", DevNull.class);
+ //dag.addStream("ignored", merger.completedFilesMetaOutput, devNull1.data);
+
+ //Setting operator properties
+ merger.setFilePath(outputDirectoryPath);
+ merger.setOverwriteOnConflict(overwriteOnConflict);
--- End diff --
setting overwriteOnConflict default to true
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r58166321
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,235 @@
+/**
--- End diff --
Move this to block package.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56625412
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
+
+ private long bytesWritten;
+ private double windowTimeSec;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ windowTimeSec = (context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)
+ * context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) * 1.0) / 1000.0;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ bytesWrittenPerSec = 0;
+ bytesWritten = 0;
+ }
+
+ /*
--- End diff --
If you don't provide javadoc here, it would pick from parent class, that should be acceptable unless you want to write something different.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
GitHub user yogidevendra reopened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
commit 5b322b5e57cddd5ad2e3ec941d3daf65bf0b2144
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-03-04T07:12:17Z
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators BlockWriter, Synchronizer, FileStitcher,
FileMerger, HDFSFileMerger
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
4. Incorporating review comments
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56784591
--- Diff: apps/filecopy/src/main/java/com/datatorrent/apps/copy/HDFSFileCopySampleApp.java ---
@@ -0,0 +1,50 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.apps.copy;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DAG.Locality;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.lib.io.fs.HDFSFileCopyModule;
+import com.datatorrent.lib.io.fs.HDFSInputModule;
+
+/**
+ * Application for HDFS to HDFS file copy
+ */
+@ApplicationAnnotation(name = "HDFSFileCopySampleApp")
+public class HDFSFileCopySampleApp implements StreamingApplication
--- End diff --
Changed to HDFSFileCopyApp
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56628057
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileCopyModule.java ---
@@ -0,0 +1,123 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import javax.validation.constraints.NotNull;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Module;
+import com.datatorrent.lib.io.block.AbstractBlockReader.ReaderRecord;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * HDFS file copy module can be used in conjunction with file input modules to
+ * copy files from any file system to HDFS. This module supports parallel write
+ * to multiple blocks of the same file and then stitching those blocks in
+ * original sequence.
+ *
+ * Essential operators are wrapped into single component using Module API.
+ *
+ */
+public class HDFSFileCopyModule implements Module
+{
+
+ /**
+ * Path of the output directory. Relative path of the files copied will be
+ * maintained w.r.t. source directory and output directory
+ */
+ @NotNull
+ protected String outputDirectoryPath;
+
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict;
+
+ /**
+ * Input port for files metadata.
+ */
+ public final transient ProxyInputPort<FileMetadata> filesMetadataInput = new ProxyInputPort<FileMetadata>();
+
+ /**
+ * Input port for blocks metadata
+ */
+ public final transient ProxyInputPort<BlockMetadata.FileBlockMetadata> blocksMetadataInput = new ProxyInputPort<BlockMetadata.FileBlockMetadata>();
+
+ /**
+ * Input port for blocks data
+ */
+ public final transient ProxyInputPort<ReaderRecord<Slice>> blockData = new ProxyInputPort<ReaderRecord<Slice>>();
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+
+ //Defining DAG
--- End diff --
As per best practices, we should not write comments unless it's really required.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56786779
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileMerger.java ---
@@ -0,0 +1,218 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * HDFS file merger extends file merger to optimize for HDFS file copy usecase.
+ * This uses fast merge from HDFS if destination filesystem is same as
+ * application filesystem.
+ */
+public class HDFSFileMerger extends FileMerger
+{
+ /**
+ * Fast merge is possible if append is allowed for output file system.
+ */
+ private transient boolean fastMergeActive;
--- End diff --
Because fastMergeActive is quick check. FastMergeDecisionMaker.isFastMergePossible is expensive check. We do second check only if quick check is passed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56624654
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
--- End diff --
can we move this to top or end?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56627690
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileStitcher.java ---
@@ -0,0 +1,408 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import com.google.common.collect.Queues;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.DAGContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchBlock;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchedFileMetaData;
+
+/**
+ * This is generic File Stitcher which can be used to merge data from one or
+ * more files into single stitched file. StitchedFileMetaData defines
+ * constituents of the stitched file.
+ *
+ * This class uses Reconciler to
+ */
+public class FileStitcher<T extends StitchedFileMetaData> extends AbstractReconciler<T, T>
+{
+ /**
+ * Filesystem on which application is running
+ */
+ protected transient FileSystem appFS;
+
+ /**
+ * Destination file system
+ */
+ protected transient FileSystem outputFS;
+
+ /**
+ * Path for destination directory
+ */
+ @NotNull
+ protected String filePath;
+
+ /**
+ * Path for blocks directory
+ */
+ protected transient String blocksDir;
+
+ protected static final String PART_FILE_EXTENTION = "._COPYING_";
+
+ /**
+ * Queue maintaining successful files
+ */
+ protected Queue<T> successfulFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining skipped files
+ */
+ protected Queue<T> skippedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining failed files
+ */
+ protected Queue<T> failedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Output port for emitting completed stitched files metadata
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> completedFilesMetaOutput = new DefaultOutputPort<T>();
+
+ private boolean writeChecksum = true;
+ protected transient Path tempOutFilePath;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+
+ blocksDir = context.getValue(DAGContext.APPLICATION_PATH) + Path.SEPARATOR + BlockWriter.SUBDIR_BLOCKS;
+
+ try {
+ outputFS = getOutputFSInstance();
+ outputFS.setWriteChecksum(writeChecksum);
+ } catch (IOException ex) {
+ throw new RuntimeException("Exception in getting output file system.", ex);
+ }
+ try {
+ appFS = getAppFSInstance();
+ } catch (IOException ex) {
+ try {
+ outputFS.close();
+ } catch (IOException e) {
+ throw new RuntimeException("Exception in closing output file system.", e);
+ }
+ throw new RuntimeException("Exception in getting application file system.", ex);
+ }
+
+ super.setup(context); // Calling it at the end as the reconciler thread uses resources allocated above.
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ T stitchedFileMetaData;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ stitchedFileMetaData = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(stitchedFileMetaData)) {
+ successfulFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy successful: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(stitchedFileMetaData)) {
+ skippedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy skipped: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(stitchedFileMetaData)) {
+ failedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy failed: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in successfulFiles: "
+ + stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+ completedFilesMetaOutput.emit(stitchedFileMetaData);
+ committedTuples.remove(stitchedFileMetaData);
+ doneTuples.poll();
+ }
+ }
+
+ /**
+ *
+ * @return Application FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getAppFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(blocksDir)).toUri(), new Configuration());
+ }
+
+ /**
+ *
+ * @return Destination FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getOutputFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(filePath)).toUri(), new Configuration());
+ }
+
+ @Override
+ public void teardown()
+ {
+ super.teardown();
+
+ boolean gotException = false;
+ try {
+ if (appFS != null) {
+ appFS.close();
+ appFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+
+ try {
+ if (outputFS != null) {
+ outputFS.close();
+ outputFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+ if (gotException) {
+ throw new RuntimeException("Exception while closing file systems.");
+ }
+ }
+
+ /**
+ * Enques incoming data for for processing
+ */
+ @Override
+ protected void processTuple(T stitchedFileMetaData)
+ {
+ enqueueForProcessing(stitchedFileMetaData);
+ }
+
+ /**
+ * Stitches the output file when all blocks for that file are commited
+ */
+ @Override
+ protected void processCommittedData(T stitchedFileMetaData)
+ {
+ try {
+ mergeOutputFile(stitchedFileMetaData);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to merge file: " + stitchedFileMetaData.getStitchedFileRelativePath(), e);
+ }
+ }
+
+ /**
+ * Read data from block files and write to output file. Information about
+ * which block files should be read is specified in outFileMetadata
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+
+ protected void mergeOutputFile(T stitchedFileMetaData) throws IOException
+ {
+ mergeBlocks(stitchedFileMetaData);
+ successfulFiles.add(stitchedFileMetaData);
+ LOG.debug("Completed processing file: {} ", stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+
+ protected void mergeBlocks(T stitchedFileMetaData) throws IOException
+ {
+ //when writing to tmp files there can be vagrant tmp files which we have to clean
+ final Path dst = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ PathFilter tempFileFilter = new PathFilter()
+ {
+ @Override
+ public boolean accept(Path path)
+ {
+ return path.getName().startsWith(dst.getName()) && path.getName().endsWith(PART_FILE_EXTENTION);
+ }
+ };
+ if (outputFS.exists(dst.getParent())) {
+ FileStatus[] statuses = outputFS.listStatus(dst.getParent(), tempFileFilter);
+ for (FileStatus status : statuses) {
+ String statusName = status.getPath().getName();
+ LOG.debug("deleting vagrant file {}", statusName);
+ outputFS.delete(status.getPath(), true);
+ }
+ }
+ tempOutFilePath = new Path(filePath,
+ stitchedFileMetaData.getStitchedFileRelativePath() + '.' + System.currentTimeMillis() + PART_FILE_EXTENTION);
+ try {
+ writeTempOutputFile(stitchedFileMetaData);
+ moveToFinalFile(stitchedFileMetaData);
+ } catch (BlockNotFoundException e) {
+ LOG.info("Block file {} not found. Assuming recovery mode for file {}. ", e.getBlockPath(),
--- End diff --
shouldn't this message be at higher level than INFO?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56630912
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/Synchronizer.java ---
@@ -0,0 +1,373 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.block.BlockMetadata.FileBlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.lib.io.fs.FileStitcher.BlockNotFoundException;
+
+/**
+ * Synchronizer waits for all data blocks for a file to be written to disk. It
+ * sends trigger to merge the file only after all blocks are written to HDFS
+ */
+public class Synchronizer extends BaseOperator
+{
+ /**
+ * Map to FileMetadata for given file
+ */
+ private Map<String, FileMetadata> fileMetadataMap = Maps.newHashMap();
+
+ /**
+ * Map maintaining BlockId to BlockMetadata mapping for given file
+ */
+ private Map<String, Map<Long, BlockMetadata.FileBlockMetadata>> fileToReceivedBlocksMetadataMap = Maps.newHashMap();
+
+ public Synchronizer()
--- End diff --
No need of having this, java provides non parameterized constructor by default if you don't have any other constructor in your class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56625652
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
+
+ private long bytesWritten;
+ private double windowTimeSec;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ windowTimeSec = (context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)
+ * context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) * 1.0) / 1000.0;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ bytesWrittenPerSec = 0;
+ bytesWritten = 0;
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ OutputFileMetadata outputFileMetadata;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ outputFileMetadata = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(outputFileMetadata)) {
+ successfulFiles.remove(outputFileMetadata);
+ LOG.debug("File copy successful: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(outputFileMetadata)) {
+ skippedFiles.remove(outputFileMetadata);
+ LOG.debug("File copy skipped: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(outputFileMetadata)) {
+ failedFiles.remove(outputFileMetadata);
+ LOG.debug("File copy failed: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in successfulFiles: "
--- End diff --
minor: message cannot say "but not in successfulFiles" it can be sucess/skip/error.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56983061
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileStitcher.java ---
@@ -0,0 +1,408 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import com.google.common.collect.Queues;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.DAGContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchBlock;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchedFileMetaData;
+
+/**
+ * This is generic File Stitcher which can be used to merge data from one or
+ * more files into single stitched file. StitchedFileMetaData defines
+ * constituents of the stitched file.
+ *
+ * This class uses Reconciler to
+ */
+public class FileStitcher<T extends StitchedFileMetaData> extends AbstractReconciler<T, T>
+{
+ /**
+ * Filesystem on which application is running
+ */
+ protected transient FileSystem appFS;
+
+ /**
+ * Destination file system
+ */
+ protected transient FileSystem outputFS;
+
+ /**
+ * Path for destination directory
+ */
+ @NotNull
+ protected String filePath;
+
+ /**
+ * Path for blocks directory
+ */
+ protected transient String blocksDir;
+
+ protected static final String PART_FILE_EXTENTION = "._COPYING_";
+
+ /**
+ * Queue maintaining successful files
+ */
+ protected Queue<T> successfulFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining skipped files
+ */
+ protected Queue<T> skippedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining failed files
+ */
+ protected Queue<T> failedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Output port for emitting completed stitched files metadata
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> completedFilesMetaOutput = new DefaultOutputPort<T>();
+
+ private boolean writeChecksum = true;
+ protected transient Path tempOutFilePath;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+
+ blocksDir = context.getValue(DAGContext.APPLICATION_PATH) + Path.SEPARATOR + BlockWriter.SUBDIR_BLOCKS;
+
+ try {
+ outputFS = getOutputFSInstance();
+ outputFS.setWriteChecksum(writeChecksum);
+ } catch (IOException ex) {
+ throw new RuntimeException("Exception in getting output file system.", ex);
+ }
+ try {
+ appFS = getAppFSInstance();
+ } catch (IOException ex) {
+ try {
+ outputFS.close();
+ } catch (IOException e) {
+ throw new RuntimeException("Exception in closing output file system.", e);
+ }
+ throw new RuntimeException("Exception in getting application file system.", ex);
+ }
+
+ super.setup(context); // Calling it at the end as the reconciler thread uses resources allocated above.
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ T stitchedFileMetaData;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ stitchedFileMetaData = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(stitchedFileMetaData)) {
+ successfulFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy successful: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(stitchedFileMetaData)) {
+ skippedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy skipped: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(stitchedFileMetaData)) {
+ failedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy failed: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in successfulFiles: "
+ + stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+ completedFilesMetaOutput.emit(stitchedFileMetaData);
+ committedTuples.remove(stitchedFileMetaData);
+ doneTuples.poll();
+ }
+ }
+
+ /**
+ *
+ * @return Application FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getAppFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(blocksDir)).toUri(), new Configuration());
+ }
+
+ /**
+ *
+ * @return Destination FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getOutputFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(filePath)).toUri(), new Configuration());
+ }
+
+ @Override
+ public void teardown()
+ {
+ super.teardown();
+
+ boolean gotException = false;
+ try {
+ if (appFS != null) {
+ appFS.close();
+ appFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+
+ try {
+ if (outputFS != null) {
+ outputFS.close();
+ outputFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+ if (gotException) {
+ throw new RuntimeException("Exception while closing file systems.");
+ }
+ }
+
+ /**
+ * Enques incoming data for for processing
+ */
+ @Override
+ protected void processTuple(T stitchedFileMetaData)
+ {
+ enqueueForProcessing(stitchedFileMetaData);
+ }
+
+ /**
+ * Stitches the output file when all blocks for that file are commited
+ */
+ @Override
+ protected void processCommittedData(T stitchedFileMetaData)
+ {
+ try {
+ mergeOutputFile(stitchedFileMetaData);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to merge file: " + stitchedFileMetaData.getStitchedFileRelativePath(), e);
+ }
+ }
+
+ /**
+ * Read data from block files and write to output file. Information about
+ * which block files should be read is specified in outFileMetadata
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+
+ protected void mergeOutputFile(T stitchedFileMetaData) throws IOException
+ {
+ mergeBlocks(stitchedFileMetaData);
+ successfulFiles.add(stitchedFileMetaData);
+ LOG.debug("Completed processing file: {} ", stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+
+ protected void mergeBlocks(T stitchedFileMetaData) throws IOException
+ {
+ //when writing to tmp files there can be vagrant tmp files which we have to clean
+ final Path dst = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ PathFilter tempFileFilter = new PathFilter()
+ {
+ @Override
+ public boolean accept(Path path)
+ {
+ return path.getName().startsWith(dst.getName()) && path.getName().endsWith(PART_FILE_EXTENTION);
+ }
+ };
+ if (outputFS.exists(dst.getParent())) {
+ FileStatus[] statuses = outputFS.listStatus(dst.getParent(), tempFileFilter);
+ for (FileStatus status : statuses) {
+ String statusName = status.getPath().getName();
+ LOG.debug("deleting vagrant file {}", statusName);
+ outputFS.delete(status.getPath(), true);
+ }
+ }
+ tempOutFilePath = new Path(filePath,
+ stitchedFileMetaData.getStitchedFileRelativePath() + '.' + System.currentTimeMillis() + PART_FILE_EXTENTION);
+ try {
+ writeTempOutputFile(stitchedFileMetaData);
+ moveToFinalFile(stitchedFileMetaData);
+ } catch (BlockNotFoundException e) {
+ LOG.info("Block file {} not found. Assuming recovery mode for file {}. ", e.getBlockPath(),
--- End diff --
Using warn instead of info.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56627887
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileStitcher.java ---
@@ -0,0 +1,408 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import com.google.common.collect.Queues;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.DAGContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchBlock;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchedFileMetaData;
+
+/**
+ * This is generic File Stitcher which can be used to merge data from one or
+ * more files into single stitched file. StitchedFileMetaData defines
+ * constituents of the stitched file.
+ *
+ * This class uses Reconciler to
+ */
+public class FileStitcher<T extends StitchedFileMetaData> extends AbstractReconciler<T, T>
+{
+ /**
+ * Filesystem on which application is running
+ */
+ protected transient FileSystem appFS;
+
+ /**
+ * Destination file system
+ */
+ protected transient FileSystem outputFS;
+
+ /**
+ * Path for destination directory
+ */
+ @NotNull
+ protected String filePath;
+
+ /**
+ * Path for blocks directory
+ */
+ protected transient String blocksDir;
+
+ protected static final String PART_FILE_EXTENTION = "._COPYING_";
+
+ /**
+ * Queue maintaining successful files
+ */
+ protected Queue<T> successfulFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining skipped files
+ */
+ protected Queue<T> skippedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining failed files
+ */
+ protected Queue<T> failedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Output port for emitting completed stitched files metadata
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> completedFilesMetaOutput = new DefaultOutputPort<T>();
+
+ private boolean writeChecksum = true;
+ protected transient Path tempOutFilePath;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+
+ blocksDir = context.getValue(DAGContext.APPLICATION_PATH) + Path.SEPARATOR + BlockWriter.SUBDIR_BLOCKS;
+
+ try {
+ outputFS = getOutputFSInstance();
+ outputFS.setWriteChecksum(writeChecksum);
+ } catch (IOException ex) {
+ throw new RuntimeException("Exception in getting output file system.", ex);
+ }
+ try {
+ appFS = getAppFSInstance();
+ } catch (IOException ex) {
+ try {
+ outputFS.close();
+ } catch (IOException e) {
+ throw new RuntimeException("Exception in closing output file system.", e);
+ }
+ throw new RuntimeException("Exception in getting application file system.", ex);
+ }
+
+ super.setup(context); // Calling it at the end as the reconciler thread uses resources allocated above.
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ T stitchedFileMetaData;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ stitchedFileMetaData = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(stitchedFileMetaData)) {
+ successfulFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy successful: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(stitchedFileMetaData)) {
+ skippedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy skipped: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(stitchedFileMetaData)) {
+ failedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy failed: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in successfulFiles: "
+ + stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+ completedFilesMetaOutput.emit(stitchedFileMetaData);
+ committedTuples.remove(stitchedFileMetaData);
+ doneTuples.poll();
+ }
+ }
+
+ /**
+ *
+ * @return Application FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getAppFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(blocksDir)).toUri(), new Configuration());
+ }
+
+ /**
+ *
+ * @return Destination FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getOutputFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(filePath)).toUri(), new Configuration());
+ }
+
+ @Override
+ public void teardown()
+ {
+ super.teardown();
+
+ boolean gotException = false;
+ try {
+ if (appFS != null) {
+ appFS.close();
+ appFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+
+ try {
+ if (outputFS != null) {
+ outputFS.close();
+ outputFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+ if (gotException) {
+ throw new RuntimeException("Exception while closing file systems.");
+ }
+ }
+
+ /**
+ * Enques incoming data for for processing
+ */
+ @Override
+ protected void processTuple(T stitchedFileMetaData)
+ {
+ enqueueForProcessing(stitchedFileMetaData);
+ }
+
+ /**
+ * Stitches the output file when all blocks for that file are commited
+ */
+ @Override
+ protected void processCommittedData(T stitchedFileMetaData)
+ {
+ try {
+ mergeOutputFile(stitchedFileMetaData);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to merge file: " + stitchedFileMetaData.getStitchedFileRelativePath(), e);
+ }
+ }
+
+ /**
+ * Read data from block files and write to output file. Information about
+ * which block files should be read is specified in outFileMetadata
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+
+ protected void mergeOutputFile(T stitchedFileMetaData) throws IOException
+ {
+ mergeBlocks(stitchedFileMetaData);
+ successfulFiles.add(stitchedFileMetaData);
+ LOG.debug("Completed processing file: {} ", stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+
+ protected void mergeBlocks(T stitchedFileMetaData) throws IOException
+ {
+ //when writing to tmp files there can be vagrant tmp files which we have to clean
+ final Path dst = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ PathFilter tempFileFilter = new PathFilter()
+ {
+ @Override
+ public boolean accept(Path path)
+ {
+ return path.getName().startsWith(dst.getName()) && path.getName().endsWith(PART_FILE_EXTENTION);
+ }
+ };
+ if (outputFS.exists(dst.getParent())) {
+ FileStatus[] statuses = outputFS.listStatus(dst.getParent(), tempFileFilter);
+ for (FileStatus status : statuses) {
+ String statusName = status.getPath().getName();
+ LOG.debug("deleting vagrant file {}", statusName);
+ outputFS.delete(status.getPath(), true);
+ }
+ }
+ tempOutFilePath = new Path(filePath,
+ stitchedFileMetaData.getStitchedFileRelativePath() + '.' + System.currentTimeMillis() + PART_FILE_EXTENTION);
+ try {
+ writeTempOutputFile(stitchedFileMetaData);
+ moveToFinalFile(stitchedFileMetaData);
+ } catch (BlockNotFoundException e) {
+ LOG.info("Block file {} not found. Assuming recovery mode for file {}. ", e.getBlockPath(),
+ stitchedFileMetaData.getStitchedFileRelativePath());
+ //Remove temp output file
+ outputFS.delete(tempOutFilePath, false);
+ }
+ }
+
+ /**
+ * Writing all Stitch blocks to temporary file
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ * @throws BlockNotFoundException
+ */
+ protected OutputStream writeTempOutputFile(T stitchedFileMetaData) throws IOException, BlockNotFoundException
+ {
+ OutputStream outputStream = getOutputStream(tempOutFilePath);
+ try {
+ for (StitchBlock outputBlock : stitchedFileMetaData.getStitchBlocksList()) {
+ outputBlock.writeTo(appFS, blocksDir, outputStream);
+ }
+ } finally {
+ outputStream.close();
+ }
+ return outputStream;
+ }
+
+ protected OutputStream getOutputStream(Path partFilePath) throws IOException
+ {
+ return outputFS.create(partFilePath);
+ }
+
+ /**
+ * Moving temp output file to final file
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+ protected void moveToFinalFile(T stitchedFileMetaData) throws IOException
+ {
+ Path destination = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ moveToFinalFile(tempOutFilePath, destination);
+ }
+
+ /**
+ * Moving temp output file to final file
+ *
+ * @param tempOutFilePath
+ * Temporary output file
+ * @param destination
+ * Destination directory path
+ * @throws IOException
+ */
+ protected void moveToFinalFile(Path tempOutFilePath, Path destination) throws IOException
+ {
+ Path src = Path.getPathWithoutSchemeAndAuthority(tempOutFilePath);
+ Path dst = Path.getPathWithoutSchemeAndAuthority(destination);
+
+ boolean moveSuccessful = false;
+ if (!outputFS.exists(dst.getParent())) {
+ outputFS.mkdirs(dst.getParent());
+ }
+ if (outputFS.exists(dst)) {
+ outputFS.delete(dst, false);
+ }
+ moveSuccessful = outputFS.rename(src, dst);
+
+ if (moveSuccessful) {
+ LOG.debug("File {} moved successfully to destination folder.", dst);
+ } else {
+ throw new RuntimeException("Unable to move file from " + src + " to " + dst);
+ }
+ }
+
+ public String getBlocksDir()
+ {
+ return blocksDir;
+ }
+
+ public void setBlocksDir(String blocksDir)
--- End diff --
just put javadocs on all getter/setters
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56625987
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
+
+ private long bytesWritten;
+ private double windowTimeSec;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ windowTimeSec = (context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)
+ * context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) * 1.0) / 1000.0;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ bytesWrittenPerSec = 0;
+ bytesWritten = 0;
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ OutputFileMetadata outputFileMetadata;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ outputFileMetadata = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(outputFileMetadata)) {
+ successfulFiles.remove(outputFileMetadata);
+ LOG.debug("File copy successful: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(outputFileMetadata)) {
+ skippedFiles.remove(outputFileMetadata);
+ LOG.debug("File copy skipped: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(outputFileMetadata)) {
+ failedFiles.remove(outputFileMetadata);
+ LOG.debug("File copy failed: {}", outputFileMetadata.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in successfulFiles: "
+ + outputFileMetadata.getStitchedFileRelativePath());
+ }
+ completedFilesMetaOutput.emit(outputFileMetadata);
+ committedTuples.remove(outputFileMetadata);
+ doneTuples.poll();
+ }
+
+ bytesWrittenPerSec = (long)(bytesWritten / windowTimeSec);
+ }
+
+ @Override
+ protected void mergeOutputFile(OutputFileMetadata outputFileMetadata) throws IOException
+ {
+ LOG.debug("Processing file: {}", outputFileMetadata.getStitchedFileRelativePath());
+
+ Path outputFilePath = new Path(filePath, outputFileMetadata.getStitchedFileRelativePath());
+ if (outputFileMetadata.isDirectory()) {
+ createDir(outputFilePath);
+ successfulFiles.add(outputFileMetadata);
+ return;
+ }
+
+ if (outputFS.exists(outputFilePath) && !overwriteOnConflict) {
+ LOG.debug("Output file {} already exits and overwrite flag is off. Skipping.", outputFilePath);
+ skippedFiles.add(outputFileMetadata);
+ return;
+ }
+ //Call super method for serial merge of blocks
+ super.mergeOutputFile(outputFileMetadata);
+ }
+
+ @Override
+ protected OutputStream writeTempOutputFile(OutputFileMetadata outputFileMetadata)
+ throws IOException, BlockNotFoundException
+ {
+ OutputStream outputStream = super.writeTempOutputFile(outputFileMetadata);
+ bytesWritten += outputFileMetadata.getFileLength();
+ return outputStream;
+ }
+
+ private void createDir(Path outputFilePath) throws IOException
+ {
+ if (!outputFS.exists(outputFilePath)) {
+ outputFS.mkdirs(outputFilePath);
+ }
+ }
+
+ @Override
+ protected OutputStream getOutputStream(Path partFilePath) throws IOException
+ {
+ OutputStream outputStream = outputFS.create(partFilePath);
+ return outputStream;
+ }
+
+ public boolean isOverwriteOnConflict()
+ {
+ return overwriteOnConflict;
+ }
+
+ public void setOverwriteOnConflict(boolean overwriteOnConflict)
--- End diff --
javadoc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56794075
--- Diff: library/src/test/java/com/datatorrent/lib/io/fs/FastMergerDecisionMakerTest.java ---
@@ -0,0 +1,188 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.lib.io.fs.FileStitcher.BlockNotFoundException;
+import com.datatorrent.lib.io.fs.HDFSFileMerger.FastMergerDecisionMaker;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link FastMergerDecisionMaker}
+ */
+
+public class FastMergerDecisionMakerTest
+{
+ private static final long DEFAULT_BLOCK_SIZE = 100L;
+ private static final short[] DEFAULT_REPLICATION = new short[] {(short)3, (short)3, (short)3 };
+
+ FastMergerDecisionMaker fastMergerDecisionMaker;
+ @Mock
+ OutputFileMetadata fileMetadata;
+ @Mock
+ FileSystem appFS;
+ @Mock
+ FileStatus status0;
+ @Mock
+ FileStatus status1;
+ @Mock
+ FileStatus status2;
+
+ @Before
+ public void setup()
+ {
+ MockitoAnnotations.initMocks(this);
+
+ when(fileMetadata.getNumberOfBlocks()).thenReturn(3);
+ when(fileMetadata.getBlockIds()).thenReturn(new long[] {0L, 1L, 2L });
+
+ fastMergerDecisionMaker = new FastMergerDecisionMaker("", appFS, DEFAULT_BLOCK_SIZE);
+ }
+
+ /**
+ * If some block is missing then expect BlockNotFoundException.
+ *
+ * @throws IOException
+ * @throws BlockNotFoundException
+ */
+ @Test(expected = BlockNotFoundException.class)
+ public void testMissingBlock() throws IOException, BlockNotFoundException
+ {
+
+ when(appFS.exists(new Path("/0"))).thenReturn(true);
+ when(appFS.exists(new Path("/1"))).thenReturn(false);
+
+ when(status0.getReplication()).thenReturn(DEFAULT_REPLICATION[0]);
+ when(status0.getLen()).thenReturn(DEFAULT_BLOCK_SIZE);
+
+ when(appFS.getFileStatus(new Path("/0"))).thenReturn(status0);
+
+ fastMergerDecisionMaker.isFastMergePossible(fileMetadata);
+ fail("Failed when one block missing.");
+ }
+
+ private void initializeMocks(long[] blockSizes, short[] replication) throws IOException, BlockNotFoundException
+ {
+ when(appFS.exists(Matchers.any(Path.class))).thenReturn(true);
--- End diff --
moved to @ Before method
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56625510
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
+
+ private long bytesWritten;
+ private double windowTimeSec;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ windowTimeSec = (context.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)
+ * context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS) * 1.0) / 1000.0;
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ bytesWrittenPerSec = 0;
+ bytesWritten = 0;
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ OutputFileMetadata outputFileMetadata;
--- End diff --
there is no call to super, is it okay to skip the call?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
GitHub user yogidevendra reopened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
commit 5b322b5e57cddd5ad2e3ec941d3daf65bf0b2144
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-03-04T07:12:17Z
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators BlockWriter, Synchronizer, FileStitcher,
FileMerger, HDFSFileMerger
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
4. Incorporating review comments
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56786810
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/Synchronizer.java ---
@@ -0,0 +1,373 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.lib.io.block.BlockMetadata.FileBlockMetadata;
+import com.datatorrent.lib.io.fs.AbstractFileSplitter.FileMetadata;
+import com.datatorrent.lib.io.fs.FileStitcher.BlockNotFoundException;
+
+/**
+ * Synchronizer waits for all data blocks for a file to be written to disk. It
+ * sends trigger to merge the file only after all blocks are written to HDFS
+ */
+public class Synchronizer extends BaseOperator
+{
+ /**
+ * Map to FileMetadata for given file
+ */
+ private Map<String, FileMetadata> fileMetadataMap = Maps.newHashMap();
+
+ /**
+ * Map maintaining BlockId to BlockMetadata mapping for given file
+ */
+ private Map<String, Map<Long, BlockMetadata.FileBlockMetadata>> fileToReceivedBlocksMetadataMap = Maps.newHashMap();
+
+ public Synchronizer()
--- End diff --
removed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56630486
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/HDFSFileMerger.java ---
@@ -0,0 +1,218 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * HDFS file merger extends file merger to optimize for HDFS file copy usecase.
+ * This uses fast merge from HDFS if destination filesystem is same as
+ * application filesystem.
+ */
+public class HDFSFileMerger extends FileMerger
+{
+ /**
+ * Fast merge is possible if append is allowed for output file system.
+ */
+ private transient boolean fastMergeActive;
--- End diff --
why we need this flag if we have FastMergeDecisionMaker?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56624903
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
+
+ @Override
+ public Collection<Partition<BlockWriter>> definePartitions(Collection<Partition<BlockWriter>> partitions,
+ PartitioningContext context)
+ {
+ if (context.getParallelPartitionCount() == 0) {
+ return partitions;
+ }
+
+ // if there is no change of count, return the same collection
+ if (context.getParallelPartitionCount() == partitions.size()) {
+ LOG.debug("no change is partition count: " + partitions.size());
+ return partitions;
+ }
+
+ List<BasicCounters<MutableLong>> deletedCounters = Lists.newArrayList();
+
+ LOG.debug("block writer parallel partition count {}", context.getParallelPartitionCount());
+ int morePartitionsToCreate = context.getParallelPartitionCount() - partitions.size();
+ if (morePartitionsToCreate < 0) {
+ //Delete partitions
+ Iterator<Partition<BlockWriter>> partitionIterator = partitions.iterator();
+
+ while (morePartitionsToCreate++ < 0) {
+ Partition<BlockWriter> toRemove = partitionIterator.next();
+ deletedCounters.add(toRemove.getPartitionedInstance().fileCounters);
+ partitionIterator.remove();
+ }
+ } else {
+ //Add more partitions
+ BlockWriter anOperator = partitions.iterator().next().getPartitionedInstance();
+
+ while (morePartitionsToCreate-- > 0) {
+ DefaultPartition<BlockWriter> partition = new DefaultPartition<BlockWriter>(anOperator);
+ partitions.add(partition);
+ }
+ }
+
+ //transfer the counters
+ BlockWriter targetWriter = partitions.iterator().next().getPartitionedInstance();
--- End diff --
do we need this with AutoMetrics?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56624529
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
--- End diff --
can't we still accept filePath as parameter, and from your module code set the right value. This block write can be generic and need not be used to write only to "blocks" directory. in fact we can keep blockdir as well configurable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56637593
--- Diff: library/src/test/java/com/datatorrent/lib/io/fs/FastMergerDecisionMakerTest.java ---
@@ -0,0 +1,188 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.lib.io.fs.FileStitcher.BlockNotFoundException;
+import com.datatorrent.lib.io.fs.HDFSFileMerger.FastMergerDecisionMaker;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link FastMergerDecisionMaker}
+ */
+
+public class FastMergerDecisionMakerTest
+{
+ private static final long DEFAULT_BLOCK_SIZE = 100L;
+ private static final short[] DEFAULT_REPLICATION = new short[] {(short)3, (short)3, (short)3 };
+
+ FastMergerDecisionMaker fastMergerDecisionMaker;
+ @Mock
+ OutputFileMetadata fileMetadata;
+ @Mock
+ FileSystem appFS;
+ @Mock
+ FileStatus status0;
+ @Mock
+ FileStatus status1;
+ @Mock
+ FileStatus status2;
+
+ @Before
+ public void setup()
+ {
+ MockitoAnnotations.initMocks(this);
+
+ when(fileMetadata.getNumberOfBlocks()).thenReturn(3);
+ when(fileMetadata.getBlockIds()).thenReturn(new long[] {0L, 1L, 2L });
+
+ fastMergerDecisionMaker = new FastMergerDecisionMaker("", appFS, DEFAULT_BLOCK_SIZE);
+ }
+
+ /**
+ * If some block is missing then expect BlockNotFoundException.
+ *
+ * @throws IOException
+ * @throws BlockNotFoundException
+ */
+ @Test(expected = BlockNotFoundException.class)
+ public void testMissingBlock() throws IOException, BlockNotFoundException
+ {
+
+ when(appFS.exists(new Path("/0"))).thenReturn(true);
+ when(appFS.exists(new Path("/1"))).thenReturn(false);
+
+ when(status0.getReplication()).thenReturn(DEFAULT_REPLICATION[0]);
+ when(status0.getLen()).thenReturn(DEFAULT_BLOCK_SIZE);
+
+ when(appFS.getFileStatus(new Path("/0"))).thenReturn(status0);
+
+ fastMergerDecisionMaker.isFastMergePossible(fileMetadata);
+ fail("Failed when one block missing.");
+ }
+
+ private void initializeMocks(long[] blockSizes, short[] replication) throws IOException, BlockNotFoundException
+ {
+ when(appFS.exists(Matchers.any(Path.class))).thenReturn(true);
--- End diff --
you can do it in beforeTest as almost all tests need this. If one of the needs something different, you can override only that part for that particular test.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56784818
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/BlockWriter.java ---
@@ -0,0 +1,210 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.DefaultPartition;
+import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.io.block.AbstractBlockReader;
+import com.datatorrent.lib.io.block.BlockMetadata;
+import com.datatorrent.netlet.util.Slice;
+
+/**
+ * Writes a block to the appFS (HDFS on which app is running). This is temporary
+ * write to HDFS to handle large files.
+ */
+public class BlockWriter extends AbstractFileOutputOperator<AbstractBlockReader.ReaderRecord<Slice>>
+ implements Partitioner<BlockWriter>
+{
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ public static final String SUBDIR_BLOCKS = "blocks";
+ /**
+ * List of FileBlockMetadata received in the current window.
+ */
+ private transient List<BlockMetadata.FileBlockMetadata> blockMetadatas;
+
+ /**
+ * Input port to receive Block meta data
+ */
+ public final transient DefaultInputPort<BlockMetadata.FileBlockMetadata> blockMetadataInput = new DefaultInputPort<BlockMetadata.FileBlockMetadata>()
+ {
+
+ @Override
+ public void process(BlockMetadata.FileBlockMetadata blockMetadata)
+ {
+ blockMetadatas.add(blockMetadata);
+ LOG.debug("received blockId {} for file {} ", blockMetadata.getBlockId(), blockMetadata.getFilePath());
+ }
+ };
+
+ /**
+ * Output port to send Block meta data to downstream operator
+ */
+ public final transient DefaultOutputPort<BlockMetadata.FileBlockMetadata> blockMetadataOutput = new DefaultOutputPort<BlockMetadata.FileBlockMetadata>();
+
+ public BlockWriter()
+ {
+ super();
+ blockMetadatas = Lists.newArrayList();
+ //The base class puts a restriction that the file-path cannot be null. With this block writer it is
+ //being initialized in setup and not through configuration. So setting it to empty string.
+ filePath = "";
+ }
+
+ /**
+ * Also, initializes the filePath based on Application path
+ */
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ filePath = context.getValue(Context.DAGContext.APPLICATION_PATH) + Path.SEPARATOR + SUBDIR_BLOCKS;
+ super.setup(context);
+ }
+
+ /**
+ * Finalizes files for all the blockMetaDatas received during current window
+ */
+ @Override
+ public void endWindow()
+ {
+ super.endWindow();
+
+ streamsCache.asMap().clear();
+ endOffsets.clear();
+
+ for (BlockMetadata.FileBlockMetadata blockMetadata : blockMetadatas) {
+ try {
+ finalizeFile(Long.toString(blockMetadata.getBlockId()));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ blockMetadataOutput.emit(blockMetadata);
+ }
+ blockMetadatas.clear();
+ }
+
+ @Override
+ protected String getFileName(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return Long.toString(tuple.getBlockId());
+ }
+
+ @Override
+ protected byte[] getBytesForTuple(AbstractBlockReader.ReaderRecord<Slice> tuple)
+ {
+ return tuple.getRecord().buffer;
+ }
+
+ private static final Logger LOG = LoggerFactory.getLogger(BlockWriter.class);
+
+ @Override
+ public Collection<Partition<BlockWriter>> definePartitions(Collection<Partition<BlockWriter>> partitions,
+ PartitioningContext context)
+ {
+ if (context.getParallelPartitionCount() == 0) {
+ return partitions;
+ }
+
+ // if there is no change of count, return the same collection
+ if (context.getParallelPartitionCount() == partitions.size()) {
+ LOG.debug("no change is partition count: " + partitions.size());
+ return partitions;
+ }
+
+ List<BasicCounters<MutableLong>> deletedCounters = Lists.newArrayList();
+
+ LOG.debug("block writer parallel partition count {}", context.getParallelPartitionCount());
+ int morePartitionsToCreate = context.getParallelPartitionCount() - partitions.size();
+ if (morePartitionsToCreate < 0) {
+ //Delete partitions
+ Iterator<Partition<BlockWriter>> partitionIterator = partitions.iterator();
+
+ while (morePartitionsToCreate++ < 0) {
+ Partition<BlockWriter> toRemove = partitionIterator.next();
+ deletedCounters.add(toRemove.getPartitionedInstance().fileCounters);
+ partitionIterator.remove();
+ }
+ } else {
+ //Add more partitions
+ BlockWriter anOperator = partitions.iterator().next().getPartitionedInstance();
+
+ while (morePartitionsToCreate-- > 0) {
+ DefaultPartition<BlockWriter> partition = new DefaultPartition<BlockWriter>(anOperator);
+ partitions.add(partition);
+ }
+ }
+
+ //transfer the counters
+ BlockWriter targetWriter = partitions.iterator().next().getPartitionedInstance();
--- End diff --
Base class is currently using Counters and not AutoMetrics.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
Github user yogidevendra closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/216
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by yogidevendra <gi...@git.apache.org>.
GitHub user yogidevendra reopened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/216
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators required for file copy
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yogidevendra/incubator-apex-malhar APEXMALHAR-2013-hdfs-file-copy-module-PR2
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/216.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #216
----
commit 5b322b5e57cddd5ad2e3ec941d3daf65bf0b2144
Author: yogidevendra <de...@datatorrent.com>
Date: 2016-03-04T07:12:17Z
APEXMALHAR-2013 : HDFS output module for file copy
1. Added operators BlockWriter, Synchronizer, FileStitcher,
FileMerger, HDFSFileMerger
2. Added junit tests
3. Added sample app for HDFS to HDFS file copy app
4. Incorporating review comments
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by chaithu14 <gi...@git.apache.org>.
Github user chaithu14 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r58166301
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileStitcher.java ---
@@ -0,0 +1,438 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import com.google.common.collect.Queues;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.DAGContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchBlock;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchedFileMetaData;
+
+/**
+ * This is generic File Stitcher which can be used to merge data from one or
+ * more files into single stitched file. StitchedFileMetaData defines
+ * constituents of the stitched file.
+ *
+ * This class uses Reconciler to
+ */
+public class FileStitcher<T extends StitchedFileMetaData> extends AbstractReconciler<T, T>
+{
+ /**
+ * Filesystem on which application is running
+ */
+ protected transient FileSystem appFS;
+
+ /**
+ * Destination file system
+ */
+ protected transient FileSystem outputFS;
+
+ /**
+ * Path for destination directory
+ */
+ @NotNull
+ protected String filePath;
+
+ /**
+ * Path for blocks directory
+ */
+ protected transient String blocksDirectoryPath;
+
+ /**
+ * Directory under application directory where blocks gets stored
+ */
+ private String blocksDirectory = BlockWriter.DEFAULT_BLOCKS_DIR;
+
+ protected static final String PART_FILE_EXTENTION = "._COPYING_";
+
+ /**
+ * Queue maintaining successful files
+ */
+ protected Queue<T> successfulFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining skipped files
+ */
+ protected Queue<T> skippedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining failed files
+ */
+ protected Queue<T> failedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Output port for emitting completed stitched files metadata
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> completedFilesMetaOutput = new DefaultOutputPort<T>();
+
+ private boolean writeChecksum = true;
+ protected transient Path tempOutFilePath;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+
+ blocksDirectoryPath = context.getValue(DAGContext.APPLICATION_PATH) + Path.SEPARATOR + blocksDirectory;
+
+ try {
+ outputFS = getOutputFSInstance();
+ outputFS.setWriteChecksum(writeChecksum);
+ } catch (IOException ex) {
+ throw new RuntimeException("Exception in getting output file system.", ex);
+ }
+ try {
+ appFS = getAppFSInstance();
+ } catch (IOException ex) {
+ try {
+ outputFS.close();
+ } catch (IOException e) {
+ throw new RuntimeException("Exception in closing output file system.", e);
+ }
+ throw new RuntimeException("Exception in getting application file system.", ex);
+ }
+
+ super.setup(context); // Calling it at the end as the reconciler thread uses resources allocated above.
+ }
+
+ /*
+ * Calls super.endWindow() and sets counters
+ * @see com.datatorrent.api.BaseOperator#endWindow()
+ */
+ @Override
+ public void endWindow()
+ {
+ T stitchedFileMetaData;
+ int size = doneTuples.size();
+ for (int i = 0; i < size; i++) {
+ stitchedFileMetaData = doneTuples.peek();
+ // If a tuple is present in doneTuples, it has to be also present in successful/failed/skipped
+ // as processCommittedData adds tuple in successful/failed/skipped
+ // and then reconciler thread add that in doneTuples
+ if (successfulFiles.contains(stitchedFileMetaData)) {
+ successfulFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy successful: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (skippedFiles.contains(stitchedFileMetaData)) {
+ skippedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy skipped: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else if (failedFiles.contains(stitchedFileMetaData)) {
+ failedFiles.remove(stitchedFileMetaData);
+ LOG.debug("File copy failed: {}", stitchedFileMetaData.getStitchedFileRelativePath());
+ } else {
+ throw new RuntimeException("Tuple present in doneTuples but not in sucessful /skipped/ failed files: "
+ + stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+ completedFilesMetaOutput.emit(stitchedFileMetaData);
+ committedTuples.remove(stitchedFileMetaData);
+ doneTuples.poll();
+ }
+ }
+
+ /**
+ *
+ * @return Application FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getAppFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(blocksDirectoryPath)).toUri(), new Configuration());
+ }
+
+ /**
+ *
+ * @return Destination FileSystem instance
+ * @throws IOException
+ */
+ protected FileSystem getOutputFSInstance() throws IOException
+ {
+ return FileSystem.newInstance((new Path(filePath)).toUri(), new Configuration());
+ }
+
+ @Override
+ public void teardown()
+ {
+ super.teardown();
+
+ boolean gotException = false;
+ try {
+ if (appFS != null) {
+ appFS.close();
+ appFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+
+ try {
+ if (outputFS != null) {
+ outputFS.close();
+ outputFS = null;
+ }
+ } catch (IOException e) {
+ gotException = true;
+ }
+ if (gotException) {
+ throw new RuntimeException("Exception while closing file systems.");
+ }
+ }
+
+ /**
+ * Enques incoming data for for processing
+ */
+ @Override
+ protected void processTuple(T stitchedFileMetaData)
+ {
+ LOG.debug("stitchedFileMetaData: {}", stitchedFileMetaData);
+ enqueueForProcessing(stitchedFileMetaData);
+ }
+
+ /**
+ * Stitches the output file when all blocks for that file are commited
+ */
+ @Override
+ protected void processCommittedData(T stitchedFileMetaData)
+ {
+ try {
+ mergeOutputFile(stitchedFileMetaData);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to merge file: " + stitchedFileMetaData.getStitchedFileRelativePath(), e);
+ }
+ }
+
+ /**
+ * Read data from block files and write to output file. Information about
+ * which block files should be read is specified in outFileMetadata
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+
+ protected void mergeOutputFile(T stitchedFileMetaData) throws IOException
+ {
+ mergeBlocks(stitchedFileMetaData);
+ successfulFiles.add(stitchedFileMetaData);
+ LOG.debug("Completed processing file: {} ", stitchedFileMetaData.getStitchedFileRelativePath());
+ }
+
+ protected void mergeBlocks(T stitchedFileMetaData) throws IOException
+ {
+ //when writing to tmp files there can be vagrant tmp files which we have to clean
+ final Path dst = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ PathFilter tempFileFilter = new PathFilter()
+ {
+ @Override
+ public boolean accept(Path path)
+ {
+ return path.getName().startsWith(dst.getName()) && path.getName().endsWith(PART_FILE_EXTENTION);
+ }
+ };
+ if (outputFS.exists(dst.getParent())) {
+ FileStatus[] statuses = outputFS.listStatus(dst.getParent(), tempFileFilter);
+ for (FileStatus status : statuses) {
+ String statusName = status.getPath().getName();
+ LOG.debug("deleting vagrant file {}", statusName);
+ outputFS.delete(status.getPath(), true);
+ }
+ }
+ tempOutFilePath = new Path(filePath,
+ stitchedFileMetaData.getStitchedFileRelativePath() + '.' + System.currentTimeMillis() + PART_FILE_EXTENTION);
+ try {
+ writeTempOutputFile(stitchedFileMetaData);
+ moveToFinalFile(stitchedFileMetaData);
+ } catch (BlockNotFoundException e) {
+ LOG.warn("Block file {} not found. Assuming recovery mode for file {}. ", e.getBlockPath(),
+ stitchedFileMetaData.getStitchedFileRelativePath());
+ //Remove temp output file
+ outputFS.delete(tempOutFilePath, false);
+ }
+ }
+
+ /**
+ * Writing all Stitch blocks to temporary file
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ * @throws BlockNotFoundException
+ */
+ protected OutputStream writeTempOutputFile(T stitchedFileMetaData) throws IOException, BlockNotFoundException
+ {
+ OutputStream outputStream = getOutputStream(tempOutFilePath);
+ try {
+ for (StitchBlock outputBlock : stitchedFileMetaData.getStitchBlocksList()) {
+ outputBlock.writeTo(appFS, blocksDirectoryPath, outputStream);
+ }
+ } finally {
+ outputStream.close();
+ }
+ return outputStream;
+ }
+
+ protected OutputStream getOutputStream(Path partFilePath) throws IOException
+ {
+ return outputFS.create(partFilePath);
+ }
+
+ /**
+ * Moving temp output file to final file
+ *
+ * @param stitchedFileMetaData
+ * @throws IOException
+ */
+ protected void moveToFinalFile(T stitchedFileMetaData) throws IOException
+ {
+ Path destination = new Path(filePath, stitchedFileMetaData.getStitchedFileRelativePath());
+ moveToFinalFile(tempOutFilePath, destination);
+ }
+
+ /**
+ * Moving temp output file to final file
+ *
+ * @param tempOutFilePath
+ * Temporary output file
+ * @param destination
+ * Destination directory path
+ * @throws IOException
+ */
+ protected void moveToFinalFile(Path tempOutFilePath, Path destination) throws IOException
+ {
+ Path src = Path.getPathWithoutSchemeAndAuthority(tempOutFilePath);
+ Path dst = Path.getPathWithoutSchemeAndAuthority(destination);
+
+ boolean moveSuccessful = false;
+ if (!outputFS.exists(dst.getParent())) {
+ outputFS.mkdirs(dst.getParent());
+ }
+ if (outputFS.exists(dst)) {
+ outputFS.delete(dst, false);
+ }
+ moveSuccessful = outputFS.rename(src, dst);
+
+ if (moveSuccessful) {
+ LOG.debug("File {} moved successfully to destination folder.", dst);
+ } else {
+ throw new RuntimeException("Unable to move file from " + src + " to " + dst);
+ }
+ }
+
--- End diff --
Looks like you are not deleting the blocks after successfully moved the file.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56626208
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileStitcher.java ---
@@ -0,0 +1,408 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Queue;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+
+import com.google.common.collect.Queues;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.DAGContext;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchBlock;
+import com.datatorrent.lib.io.fs.Synchronizer.StitchedFileMetaData;
+
+/**
+ * This is generic File Stitcher which can be used to merge data from one or
+ * more files into single stitched file. StitchedFileMetaData defines
+ * constituents of the stitched file.
+ *
+ * This class uses Reconciler to
+ */
+public class FileStitcher<T extends StitchedFileMetaData> extends AbstractReconciler<T, T>
+{
+ /**
+ * Filesystem on which application is running
+ */
+ protected transient FileSystem appFS;
+
+ /**
+ * Destination file system
+ */
+ protected transient FileSystem outputFS;
+
+ /**
+ * Path for destination directory
+ */
+ @NotNull
+ protected String filePath;
+
+ /**
+ * Path for blocks directory
+ */
+ protected transient String blocksDir;
+
+ protected static final String PART_FILE_EXTENTION = "._COPYING_";
+
+ /**
+ * Queue maintaining successful files
+ */
+ protected Queue<T> successfulFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining skipped files
+ */
+ protected Queue<T> skippedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Queue maintaining failed files
+ */
+ protected Queue<T> failedFiles = Queues.newLinkedBlockingQueue();
+
+ /**
+ * Output port for emitting completed stitched files metadata
+ */
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<T> completedFilesMetaOutput = new DefaultOutputPort<T>();
+
+ private boolean writeChecksum = true;
+ protected transient Path tempOutFilePath;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+
+ blocksDir = context.getValue(DAGContext.APPLICATION_PATH) + Path.SEPARATOR + BlockWriter.SUBDIR_BLOCKS;
--- End diff --
same as above, can we have this blocksDir parameter configurable?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56637510
--- Diff: library/src/test/java/com/datatorrent/lib/io/fs/FastMergerDecisionMakerTest.java ---
@@ -0,0 +1,188 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Matchers;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.lib.io.fs.FileStitcher.BlockNotFoundException;
+import com.datatorrent.lib.io.fs.HDFSFileMerger.FastMergerDecisionMaker;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link FastMergerDecisionMaker}
+ */
+
+public class FastMergerDecisionMakerTest
+{
+ private static final long DEFAULT_BLOCK_SIZE = 100L;
+ private static final short[] DEFAULT_REPLICATION = new short[] {(short)3, (short)3, (short)3 };
+
+ FastMergerDecisionMaker fastMergerDecisionMaker;
+ @Mock
+ OutputFileMetadata fileMetadata;
+ @Mock
+ FileSystem appFS;
+ @Mock
+ FileStatus status0;
+ @Mock
+ FileStatus status1;
+ @Mock
+ FileStatus status2;
+
+ @Before
+ public void setup()
+ {
+ MockitoAnnotations.initMocks(this);
+
+ when(fileMetadata.getNumberOfBlocks()).thenReturn(3);
+ when(fileMetadata.getBlockIds()).thenReturn(new long[] {0L, 1L, 2L });
+
+ fastMergerDecisionMaker = new FastMergerDecisionMaker("", appFS, DEFAULT_BLOCK_SIZE);
+ }
+
+ /**
+ * If some block is missing then expect BlockNotFoundException.
+ *
+ * @throws IOException
+ * @throws BlockNotFoundException
+ */
+ @Test(expected = BlockNotFoundException.class)
+ public void testMissingBlock() throws IOException, BlockNotFoundException
--- End diff --
:+1:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-2013 : HDFS output ...
Posted by DT-Priyanka <gi...@git.apache.org>.
Github user DT-Priyanka commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/216#discussion_r56625332
--- Diff: library/src/main/java/com/datatorrent/lib/io/fs/FileMerger.java ---
@@ -0,0 +1,161 @@
+/**
+ * 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.
+ */
+
+package com.datatorrent.lib.io.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.fs.Path;
+
+import com.datatorrent.api.AutoMetric;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.io.fs.Synchronizer.OutputFileMetadata;
+
+/**
+ * This operator merges the blocks into a file. The list of blocks is obtained
+ * from the OutputFileMetadata. The implementation extends FileStitcher (which
+ * uses reconciler), hence the file merging operation is carried out in a
+ * separate thread.
+ *
+ */
+public class FileMerger extends FileStitcher<OutputFileMetadata>
+{
+ /**
+ * Flag to control if existing file with same name should be overwritten
+ */
+ private boolean overwriteOnConflict = true;
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileMerger.class);
+
+ @AutoMetric
+ private long bytesWrittenPerSec;
--- End diff --
Just emit bytesWritten in a window, no need to emit bytesPerSec
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---