falcon-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From venkat...@apache.org
Subject [1/3] git commit: FALCON-401 replicate hcat table to two targets can fail with export dir exists on source. Contributed by Sowmya Ramesh
Date Thu, 01 May 2014 20:42:59 GMT
Repository: incubator-falcon
Updated Branches:
  refs/heads/master c6bfca909 -> f17ed681b


FALCON-401 replicate hcat table to two targets can fail with export dir exists on source.
Contributed by Sowmya Ramesh


Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/d41a7354
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/d41a7354
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/d41a7354

Branch: refs/heads/master
Commit: d41a7354440de27f3cf0b9d9aa782ee658a46804
Parents: c6bfca9
Author: Venkatesh Seetharam <venkatesh@apache.org>
Authored: Thu May 1 13:02:35 2014 -0700
Committer: Venkatesh Seetharam <venkatesh@apache.org>
Committed: Thu May 1 13:02:35 2014 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |  3 +++
 .../falcon/cleanup/FeedCleanupHandler.java      |  3 ++-
 .../org/apache/falcon/entity/FeedHelper.java    | 17 ++++++++++++++++-
 .../workflow/OozieFeedWorkflowBuilder.java      | 20 ++++++++------------
 .../converter/OozieFeedWorkflowBuilderTest.java | 10 ++++------
 5 files changed, 33 insertions(+), 20 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d41a7354/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7de6a1e..db3633c 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -118,6 +118,9 @@ Trunk (Unreleased)
     FALCON-123 Improve build speeds in falcon. (Srikanth Sundarrajan via Shwetha GS)
 
   BUG FIXES
+    FALCON-401 replicate hcat table to two targets can fail with export dir
+    exists on source (Sowmya Ramesh via Venkatesh Seetharam)
+
     FALCON-364 Process specification contains a mistake in the parallel section
     (Jean-Baptiste Onofré via Venkatesh Seetharam)
 

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d41a7354/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java b/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
index 58d2199..ce96eb3 100644
--- a/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
+++ b/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
@@ -81,7 +81,8 @@ public class FeedCleanupHandler extends AbstractCleanupHandler {
 
         final CatalogStorage tableStorage = (CatalogStorage) storage;
         String stagingDir = FeedHelper.getStagingDir(cluster, feed, tableStorage, Tag.REPLICATION);
-        Path stagingPath = new Path(stagingDir + "/*/*/*");  // stagingDir/dataOutPartitionValue/nominal-time/data
+        //stagingDir/dataOutPartitionValue/nominal-time/clusterName/data
+        Path stagingPath = new Path(stagingDir + "/*/*/*/*");
         FileSystem fs = getFileSystem(cluster);
         try {
             FileStatus[] paths = fs.globStatus(stagingPath);

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d41a7354/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
index 67257e3..9b66363 100644
--- a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
@@ -248,13 +248,28 @@ public final class FeedHelper {
         return expHelp.evaluateFullExpression(exp, String.class);
     }
 
+    public static String getStagingPath(org.apache.falcon.entity.v0.cluster.Cluster clusterEntity,
+                                       Feed feed, CatalogStorage storage, Tag tag, String
suffix) {
+        String stagingDirPath = getStagingDir(clusterEntity, feed, storage, tag);
+
+        String datedPartitionKey = storage.getDatedPartitionKey();
+        String datedPartitionKeySuffix = datedPartitionKey + "=${coord:dataOutPartitionValue('output',"
+                + "'" + datedPartitionKey + "')}";
+        return stagingDirPath + "/"
+                + datedPartitionKeySuffix + "/"
+                + suffix + "/"
+                + "data";
+    }
+
     public static String getStagingDir(org.apache.falcon.entity.v0.cluster.Cluster clusterEntity,
                                        Feed feed, CatalogStorage storage, Tag tag) {
         String workflowName = EntityUtil.getWorkflowName(
                 tag, Arrays.asList(clusterEntity.getName()), feed).toString();
+
         return ClusterHelper.getCompleteLocation(clusterEntity, "staging") + "/"
-                + workflowName  + "/"
+                + workflowName + "/"
                 + storage.getDatabase() + "/"
                 + storage.getTable();
     }
+
 }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d41a7354/feed/src/main/java/org/apache/falcon/workflow/OozieFeedWorkflowBuilder.java
----------------------------------------------------------------------
diff --git a/feed/src/main/java/org/apache/falcon/workflow/OozieFeedWorkflowBuilder.java b/feed/src/main/java/org/apache/falcon/workflow/OozieFeedWorkflowBuilder.java
index f83f90c..8d5df88 100644
--- a/feed/src/main/java/org/apache/falcon/workflow/OozieFeedWorkflowBuilder.java
+++ b/feed/src/main/java/org/apache/falcon/workflow/OozieFeedWorkflowBuilder.java
@@ -662,20 +662,16 @@ public class OozieFeedWorkflowBuilder extends OozieWorkflowBuilder<Feed>
{
             Cluster trgCluster, CatalogStorage targetStorage,
             Map<String, String> props) {
             // create staging dirs for export at source & set it as distcpSourcePaths
-            String sourceDatedPartitionKey = sourceStorage.getDatedPartitionKey();
-            String sourceStagingDir =
-                FeedHelper.getStagingDir(srcCluster, entity, sourceStorage, Tag.REPLICATION)
-                    + "/" + sourceDatedPartitionKey
-                    + "=${coord:dataOutPartitionValue('output', '" + sourceDatedPartitionKey
+ "')}";
-            props.put("distcpSourcePaths", sourceStagingDir + "/" + NOMINAL_TIME_EL + "/data");
+            String sourceStagingPath =
+                FeedHelper.getStagingPath(srcCluster, entity, sourceStorage, Tag.REPLICATION,
+                        NOMINAL_TIME_EL + "/" + trgCluster.getName());
+            props.put("distcpSourcePaths", sourceStagingPath);
 
             // create staging dirs for import at target & set it as distcpTargetPaths
-            String targetDatedPartitionKey = targetStorage.getDatedPartitionKey();
-            String targetStagingDir =
-                FeedHelper.getStagingDir(trgCluster, entity, targetStorage, Tag.REPLICATION)
-                    + "/" + targetDatedPartitionKey
-                    + "=${coord:dataOutPartitionValue('output', '" + targetDatedPartitionKey
+ "')}";
-            props.put("distcpTargetPaths", targetStagingDir + "/" + NOMINAL_TIME_EL + "/data");
+            String targetStagingPath =
+                FeedHelper.getStagingPath(trgCluster, entity, targetStorage, Tag.REPLICATION,
+                        NOMINAL_TIME_EL + "/" + trgCluster.getName());
+            props.put("distcpTargetPaths", targetStagingPath);
 
             props.put("sourceRelativePaths", IGNORE); // this will bot be used for Table
storage.
         }

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/d41a7354/feed/src/test/java/org/apache/falcon/converter/OozieFeedWorkflowBuilderTest.java
----------------------------------------------------------------------
diff --git a/feed/src/test/java/org/apache/falcon/converter/OozieFeedWorkflowBuilderTest.java
b/feed/src/test/java/org/apache/falcon/converter/OozieFeedWorkflowBuilderTest.java
index 77a89c0..7cb055a 100644
--- a/feed/src/test/java/org/apache/falcon/converter/OozieFeedWorkflowBuilderTest.java
+++ b/feed/src/test/java/org/apache/falcon/converter/OozieFeedWorkflowBuilderTest.java
@@ -454,15 +454,13 @@ public class OozieFeedWorkflowBuilderTest {
 
         Assert.assertTrue(props.containsKey("distcpSourcePaths"));
         Assert.assertEquals(props.get("distcpSourcePaths"),
-                FeedHelper.getStagingDir(srcCluster, tableFeed, srcStorage, Tag.REPLICATION)
-                        + "/ds=${coord:dataOutPartitionValue('output', 'ds')}/"
-                        + "${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}/data");
+                FeedHelper.getStagingPath(srcCluster, tableFeed, srcStorage, Tag.REPLICATION,
+                        "${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}" +
"/" + trgCluster.getName()));
 
         Assert.assertTrue(props.containsKey("distcpTargetPaths"));
         Assert.assertEquals(props.get("distcpTargetPaths"),
-                FeedHelper.getStagingDir(trgCluster, tableFeed, trgStorage, Tag.REPLICATION)
-                        + "/ds=${coord:dataOutPartitionValue('output', 'ds')}/"
-                        + "${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}/data");
+                FeedHelper.getStagingPath(trgCluster, tableFeed, trgStorage, Tag.REPLICATION,
+                        "${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}" +
"/" + trgCluster.getName()));
 
         Assert.assertEquals(props.get("falconFeedStorageType"), Storage.TYPE.TABLE.name());
 


Mime
View raw message