You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@falcon.apache.org by sr...@apache.org on 2013/10/17 17:45:29 UTC
[4/8] git commit: FALCON-93 Replication to handle hive table
replication. Contributed by Venkatesh Seetharam
FALCON-93 Replication to handle hive table replication. Contributed by Venkatesh Seetharam
Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/353574c6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/353574c6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/353574c6
Branch: refs/heads/FALCON-85
Commit: 353574c6daffd8de5d01119b4d52ce5b5f580c07
Parents: 5a8fa13
Author: srikanth.sundarrajan <sr...@inmobi.com>
Authored: Thu Oct 17 20:10:26 2013 +0530
Committer: srikanth.sundarrajan <sr...@inmobi.com>
Committed: Thu Oct 17 20:10:26 2013 +0530
----------------------------------------------------------------------
.../falcon/cleanup/AbstractCleanupHandler.java | 9 +-
.../falcon/cleanup/FeedCleanupHandler.java | 41 +-
.../apache/falcon/entity/CatalogStorage.java | 54 +-
.../org/apache/falcon/entity/FeedHelper.java | 24 +-
.../falcon/cleanup/LogCleanupServiceTest.java | 65 ++-
.../falcon/entity/CatalogStorageTest.java | 24 +-
.../falcon/converter/OozieFeedMapper.java | 519 ++++++++++++-------
.../config/workflow/falcon-table-export.hql | 18 +
.../config/workflow/falcon-table-import.hql | 18 +
.../config/workflow/replication-workflow.xml | 78 ++-
.../falcon/converter/OozieFeedMapperTest.java | 89 +++-
feed/src/test/resources/src-cluster.xml | 2 +-
.../test/resources/table-replication-feed.xml | 42 ++
feed/src/test/resources/trg-cluster.xml | 2 +-
.../converter/AbstractOozieEntityMapper.java | 17 +
oozie/src/main/resources/oozie-workflow-0.3.xsd | 4 +-
pom.xml | 3 +
replication/pom.xml | 5 +
.../falcon/replication/FeedReplicator.java | 107 ++--
.../falcon/replication/FeedReplicatorTest.java | 16 +-
.../falcon/cluster/util/EmbeddedCluster.java | 15 +-
webapp/pom.xml | 10 +
.../catalog/TableStorageFeedReplicationIT.java | 174 +++++++
.../org/apache/falcon/resource/TestContext.java | 45 --
.../org/apache/falcon/util/HiveTestUtils.java | 21 +-
.../org/apache/falcon/util/OozieTestUtils.java | 84 +++
webapp/src/test/resources/table/bcp-cluster.xml | 52 ++
.../table/customer-table-replicating-feed.xml | 46 ++
.../test/resources/table/primary-cluster.xml | 52 ++
29 files changed, 1321 insertions(+), 315 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/common/src/main/java/org/apache/falcon/cleanup/AbstractCleanupHandler.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/cleanup/AbstractCleanupHandler.java b/common/src/main/java/org/apache/falcon/cleanup/AbstractCleanupHandler.java
index 49662d1..e2943e1 100644
--- a/common/src/main/java/org/apache/falcon/cleanup/AbstractCleanupHandler.java
+++ b/common/src/main/java/org/apache/falcon/cleanup/AbstractCleanupHandler.java
@@ -84,7 +84,7 @@ public abstract class AbstractCleanupHandler {
return paths;
}
- private FileSystem getFileSystem(org.apache.falcon.entity.v0.cluster.Cluster cluster)
+ protected FileSystem getFileSystem(org.apache.falcon.entity.v0.cluster.Cluster cluster)
throws FalconException {
FileSystem fs;
@@ -101,6 +101,12 @@ public abstract class AbstractCleanupHandler {
throws FalconException {
FileStatus[] logs = getAllLogs(cluster, entity);
+ delete(cluster, entity, retention, logs);
+ }
+
+ protected void delete(Cluster cluster, Entity entity, long retention, FileStatus[] logs)
+ throws FalconException {
+
long now = System.currentTimeMillis();
for (FileStatus log : logs) {
@@ -126,7 +132,6 @@ public abstract class AbstractCleanupHandler {
+ log.getPath());
}
}
-
}
private void deleteParentIfEmpty(FileSystem fs, Path parent) throws IOException {
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/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 0c8cf82..7e35c88 100644
--- a/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
+++ b/common/src/main/java/org/apache/falcon/cleanup/FeedCleanupHandler.java
@@ -18,12 +18,19 @@
package org.apache.falcon.cleanup;
import org.apache.falcon.FalconException;
+import org.apache.falcon.Tag;
+import org.apache.falcon.entity.CatalogStorage;
+import org.apache.falcon.entity.FeedHelper;
+import org.apache.falcon.entity.Storage;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.feed.Feed;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import java.io.IOException;
import java.util.Collection;
/**
@@ -44,9 +51,10 @@ public class FeedCleanupHandler extends AbstractCleanupHandler {
Cluster currentCluster = STORE.get(EntityType.CLUSTER,
cluster.getName());
if (currentCluster.getColo().equals(getCurrentColo())) {
- LOG.info("Cleaning up logs for process:" + feedName
+ LOG.info("Cleaning up logs & staged data for feed:" + feedName
+ " in cluster: " + cluster.getName() + " with retention: " + retention);
delete(currentCluster, feed, retention);
+ deleteStagedData(currentCluster, feed, retention);
} else {
LOG.info("Ignoring cleanup for process:" + feedName
+ " in cluster: " + cluster.getName() + " as this does not belong to current colo");
@@ -56,11 +64,36 @@ public class FeedCleanupHandler extends AbstractCleanupHandler {
}
}
+ /**
+ * Delete the staging area used for replicating tables.
+ *
+ * @param cluster cluster hosting the staged data
+ * @param feed feed entity
+ * @param retention retention limit
+ * @throws FalconException
+ */
+ private void deleteStagedData(Cluster cluster, Feed feed, long retention)
+ throws FalconException {
+ Storage storage = FeedHelper.createStorage(cluster, feed);
+ if (storage.getType() == Storage.TYPE.FILESYSTEM) { // FS does NOT use staging dirs
+ return;
+ }
+
+ final CatalogStorage tableStorage = (CatalogStorage) storage;
+ String stagingDir = FeedHelper.getStagingDir(cluster, feed, tableStorage, Tag.REPLICATION);
+ Path stagingPath = new Path(stagingDir + "/*/*");
+ FileSystem fs = getFileSystem(cluster);
+ try {
+ FileStatus[] paths = fs.globStatus(stagingPath);
+ delete(cluster, feed, retention, paths);
+ } catch (IOException e) {
+ throw new FalconException(e);
+ }
+ }
+
@Override
protected Path getLogPath(Entity entity, String stagingPath) {
- Path logPath = new Path(stagingPath, "falcon/workflows/feed/"
+ return new Path(stagingPath, "falcon/workflows/feed/"
+ entity.getName() + "/logs/job-*/*/*");
- return logPath;
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/common/src/main/java/org/apache/falcon/entity/CatalogStorage.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/entity/CatalogStorage.java b/common/src/main/java/org/apache/falcon/entity/CatalogStorage.java
index 8e49e38..32f7605 100644
--- a/common/src/main/java/org/apache/falcon/entity/CatalogStorage.java
+++ b/common/src/main/java/org/apache/falcon/entity/CatalogStorage.java
@@ -29,7 +29,7 @@ import org.apache.falcon.entity.v0.feed.LocationType;
import java.net.URI;
import java.net.URISyntaxException;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.regex.Matcher;
@@ -42,6 +42,7 @@ public class CatalogStorage implements Storage {
public static final String PARTITION_KEYVAL_SEPARATOR = "=";
public static final String INPUT_PATH_SEPARATOR = ":";
public static final String OUTPUT_PATH_SEPARATOR = "/";
+ public static final String PARTITION_VALUE_QUOTE = "'";
public static final String CATALOG_URL = "${hcatNode}";
@@ -54,8 +55,8 @@ public class CatalogStorage implements Storage {
this(CATALOG_URL, feed.getTable());
}
- protected CatalogStorage(Cluster cluster, Feed feed) throws URISyntaxException {
- this(ClusterHelper.getInterface(cluster, Interfacetype.REGISTRY).getEndpoint(), feed.getTable());
+ protected CatalogStorage(Cluster cluster, CatalogTable table) throws URISyntaxException {
+ this(ClusterHelper.getInterface(cluster, Interfacetype.REGISTRY).getEndpoint(), table);
}
protected CatalogStorage(String catalogUrl, CatalogTable table) throws URISyntaxException {
@@ -118,7 +119,7 @@ public class CatalogStorage implements Storage {
final String rawPartition = partRaw.replaceAll(DOLLAR_EXPR_START_NORMALIZED, DOLLAR_EXPR_START_REGEX)
.replaceAll(EXPR_CLOSE_NORMALIZED, EXPR_CLOSE_REGEX);
- partitions = new HashMap<String, String>();
+ partitions = new LinkedHashMap<String, String>(); // preserve insertion order
String[] parts = rawPartition.split(PARTITION_SEPARATOR);
for (String part : parts) {
if (part == null || part.length() == 0) {
@@ -180,7 +181,7 @@ public class CatalogStorage implements Storage {
String rawPartition = partRaw.replaceAll(DOLLAR_EXPR_START_NORMALIZED, DOLLAR_EXPR_START_REGEX)
.replaceAll(EXPR_CLOSE_NORMALIZED, EXPR_CLOSE_REGEX);
- partitions = new HashMap<String, String>();
+ partitions = new LinkedHashMap<String, String>();
String[] parts = rawPartition.split(PARTITION_SEPARATOR);
for (String part : parts) {
if (part == null || part.length() == 0) {
@@ -244,6 +245,49 @@ public class CatalogStorage implements Storage {
return datedPartitionKey;
}
+ /**
+ * Convert the partition map to filter string.
+ * Each key value pair is separated by ';'.
+ *
+ * @return filter string
+ */
+ public String toPartitionFilter() {
+ StringBuilder filter = new StringBuilder();
+ filter.append("(");
+ for (Map.Entry<String, String> entry : partitions.entrySet()) {
+ if (filter.length() > 1) {
+ filter.append(PARTITION_SEPARATOR);
+ }
+ filter.append(entry.getKey());
+ filter.append(PARTITION_KEYVAL_SEPARATOR);
+ filter.append(PARTITION_VALUE_QUOTE);
+ filter.append(entry.getValue());
+ filter.append(PARTITION_VALUE_QUOTE);
+ }
+ filter.append(")");
+ return filter.toString();
+ }
+
+ /**
+ * Convert the partition map to path string.
+ * Each key value pair is separated by '/'.
+ *
+ * @return path string
+ */
+ public String toPartitionAsPath() {
+ StringBuilder partitionFilter = new StringBuilder();
+
+ for (Map.Entry<String, String> entry : getPartitions().entrySet()) {
+ partitionFilter.append(entry.getKey())
+ .append(PARTITION_KEYVAL_SEPARATOR)
+ .append(entry.getValue())
+ .append(OUTPUT_PATH_SEPARATOR);
+ }
+
+ partitionFilter.setLength(partitionFilter.length() - 1);
+ return partitionFilter.toString();
+ }
+
@Override
public TYPE getType() {
return TYPE.TABLE;
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/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 a35ee68..e76ed2f 100644
--- a/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
+++ b/common/src/main/java/org/apache/falcon/entity/FeedHelper.java
@@ -20,6 +20,7 @@ package org.apache.falcon.entity;
import org.apache.commons.lang.StringUtils;
import org.apache.falcon.FalconException;
+import org.apache.falcon.Tag;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.cluster.Property;
import org.apache.falcon.entity.v0.feed.CatalogTable;
@@ -30,6 +31,7 @@ import org.apache.falcon.entity.v0.feed.Locations;
import org.apache.falcon.expression.ExpressionHelper;
import java.net.URISyntaxException;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -103,7 +105,7 @@ public final class FeedHelper {
try {
final CatalogTable table = getTable(cluster, feed);
if (table != null) {
- return new CatalogStorage(clusterEntity, feed);
+ return new CatalogStorage(clusterEntity, table);
}
} catch (URISyntaxException e) {
throw new FalconException(e);
@@ -112,6 +114,14 @@ public final class FeedHelper {
throw new FalconException("Both catalog and locations are not defined.");
}
+ /**
+ * Factory method to dole out a storage instance used for replication source.
+ *
+ * @param clusterEntity cluster entity
+ * @param feed feed entity
+ * @return an implementation of Storage
+ * @throws FalconException
+ */
public static Storage createReadOnlyStorage(org.apache.falcon.entity.v0.cluster.Cluster clusterEntity,
Feed feed) throws FalconException {
Cluster feedCluster = getCluster(feed, clusterEntity.getName());
@@ -123,7 +133,7 @@ public final class FeedHelper {
try {
final CatalogTable table = getTable(feedCluster, feed);
if (table != null) {
- return new CatalogStorage(clusterEntity, feed);
+ return new CatalogStorage(clusterEntity, table);
}
} catch (URISyntaxException e) {
throw new FalconException(e);
@@ -200,4 +210,14 @@ public final class FeedHelper {
expHelp.setPropertiesForVariable(properties);
return expHelp.evaluateFullExpression(exp, String.class);
}
+
+ 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 + "/"
+ + storage.getDatabase() + "/"
+ + storage.getTable();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/common/src/test/java/org/apache/falcon/cleanup/LogCleanupServiceTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/cleanup/LogCleanupServiceTest.java b/common/src/test/java/org/apache/falcon/cleanup/LogCleanupServiceTest.java
index fa21a90..10a9cc0 100644
--- a/common/src/test/java/org/apache/falcon/cleanup/LogCleanupServiceTest.java
+++ b/common/src/test/java/org/apache/falcon/cleanup/LogCleanupServiceTest.java
@@ -17,14 +17,17 @@
*/
package org.apache.falcon.cleanup;
-import java.io.IOException;
-
import org.apache.falcon.FalconException;
+import org.apache.falcon.Tag;
import org.apache.falcon.cluster.util.EmbeddedCluster;
import org.apache.falcon.entity.AbstractTestBase;
+import org.apache.falcon.entity.CatalogStorage;
+import org.apache.falcon.entity.FeedHelper;
import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency;
+import org.apache.falcon.entity.v0.cluster.Cluster;
+import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -33,6 +36,9 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import java.io.IOException;
+import java.io.InputStream;
+
/**
* Test for log cleanup service.
*/
@@ -41,6 +47,11 @@ public class LogCleanupServiceTest extends AbstractTestBase {
private FileSystem fs;
private FileSystem tfs;
private EmbeddedCluster targetDfsCluster;
+ private Path sourceStagingPath1;
+ private Path sourceStagingPath2;
+ private Path targetStagingPath1;
+ private Path targetStagingPath2;
+
private final Path instanceLogPath = new Path("/projects/falcon/staging/falcon/workflows/process/"
+ "sample" + "/logs/job-2010-01-01-01-00/000");
private final Path instanceLogPath1 = new Path("/projects/falcon/staging/falcon/workflows/process/"
@@ -107,7 +118,43 @@ public class LogCleanupServiceTest extends AbstractTestBase {
fs.createNewFile(new Path(feedInstanceLogPath, "oozie.log"));
tfs.createNewFile(new Path(feedInstanceLogPath, "oozie.log"));
+ // table feed staging dir setup
+ initializeStagingDirs();
+ createStageData(sourceStagingPath1, targetStagingPath1);
+
Thread.sleep(61000);
+
+ createStageData(sourceStagingPath2, targetStagingPath2);
+ }
+
+ private void initializeStagingDirs() throws Exception {
+ final InputStream inputStream = getClass().getResourceAsStream("/config/feed/hive-table-feed.xml");
+ Feed tableFeed = (Feed) EntityType.FEED.getUnmarshaller().unmarshal(inputStream);
+ getStore().publish(EntityType.FEED, tableFeed);
+
+ final Cluster srcCluster = dfsCluster.getCluster();
+ final CatalogStorage sourceStorage = (CatalogStorage) FeedHelper.createStorage(srcCluster, tableFeed);
+ String sourceStagingDir = FeedHelper.getStagingDir(srcCluster, tableFeed, sourceStorage, Tag.REPLICATION);
+
+ sourceStagingPath1 = new Path(sourceStagingDir + "/ds=2012092400/" + System.currentTimeMillis());
+ sourceStagingPath2 = new Path(sourceStagingDir + "/ds=2012092500/" + System.currentTimeMillis());
+
+ final Cluster targetCluster = targetDfsCluster.getCluster();
+ final CatalogStorage targetStorage = (CatalogStorage) FeedHelper.createStorage(targetCluster, tableFeed);
+ String targetStagingDir = FeedHelper.getStagingDir(targetCluster, tableFeed, targetStorage, Tag.REPLICATION);
+
+ targetStagingPath1 = new Path(targetStagingDir + "/ds=2012092400/" + System.currentTimeMillis());
+ targetStagingPath2 = new Path(targetStagingDir + "/ds=2012092500/" + System.currentTimeMillis());
+ }
+
+ private void createStageData(Path sourcePath, Path targetPath) throws Exception {
+ fs.mkdirs(sourcePath);
+ fs.createNewFile(new Path(sourcePath, "_metadata.xml"));
+ fs.createNewFile(new Path(sourcePath, "data.txt"));
+
+ tfs.mkdirs(targetPath);
+ tfs.createNewFile(new Path(targetPath, "_metadata.xml"));
+ tfs.createNewFile(new Path(targetPath, "data.txt"));
}
@Test
@@ -120,7 +167,6 @@ public class LogCleanupServiceTest extends AbstractTestBase {
Assert.assertFalse(fs.exists(instanceLogPath1));
Assert.assertFalse(fs.exists(instanceLogPath2));
Assert.assertTrue(fs.exists(instanceLogPath3));
-
}
@Test
@@ -134,5 +180,18 @@ public class LogCleanupServiceTest extends AbstractTestBase {
Assert.assertTrue(fs.exists(feedInstanceLogPath1));
Assert.assertTrue(tfs.exists(feedInstanceLogPath1));
+ // source table replication staging dirs
+ Assert.assertFalse(fs.exists(new Path(sourceStagingPath1, "_metadata.xml")));
+ Assert.assertFalse(fs.exists(new Path(sourceStagingPath1, "data.txt")));
+
+ Assert.assertTrue(fs.exists(new Path(sourceStagingPath2, "_metadata.xml")));
+ Assert.assertTrue(fs.exists(new Path(sourceStagingPath2, "data.txt")));
+
+ // target table replication staging dirs
+ Assert.assertFalse(tfs.exists(new Path(targetStagingPath1, "_metadata.xml")));
+ Assert.assertFalse(tfs.exists(new Path(targetStagingPath1, "data.txt")));
+
+ Assert.assertTrue(tfs.exists(new Path(targetStagingPath2, "_metadata.xml")));
+ Assert.assertTrue(tfs.exists(new Path(targetStagingPath2, "data.txt")));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/common/src/test/java/org/apache/falcon/entity/CatalogStorageTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/entity/CatalogStorageTest.java b/common/src/test/java/org/apache/falcon/entity/CatalogStorageTest.java
index 5c1878b..ced3026 100644
--- a/common/src/test/java/org/apache/falcon/entity/CatalogStorageTest.java
+++ b/common/src/test/java/org/apache/falcon/entity/CatalogStorageTest.java
@@ -143,7 +143,7 @@ public class CatalogStorageTest {
public void testGetUriTemplateWithCatalogUrl() throws Exception {
final String catalogUrl = "thrift://localhost:49083";
String tableUri = "catalog:clicksdb:clicks#ds=${YEAR}-${MONTH}-${DAY};region=us";
- String uriTemplate = "thrift://localhost:49083/clicksdb/clicks/region=us;ds=${YEAR}-${MONTH}-${DAY}";
+ String uriTemplate = "thrift://localhost:49083/clicksdb/clicks/ds=${YEAR}-${MONTH}-${DAY};region=us";
CatalogStorage table = new CatalogStorage(catalogUrl, tableUri);
@@ -155,7 +155,7 @@ public class CatalogStorageTest {
@Test
public void testGetUriTemplateWithOutCatalogUrl() throws Exception {
String tableUri = "catalog:clicksdb:clicks#ds=${YEAR}-${MONTH}-${DAY};region=us";
- String uriTemplate = "${hcatNode}/clicksdb/clicks/region=us;ds=${YEAR}-${MONTH}-${DAY}";
+ String uriTemplate = "${hcatNode}/clicksdb/clicks/ds=${YEAR}-${MONTH}-${DAY};region=us";
CatalogStorage table = new CatalogStorage(CatalogStorage.CATALOG_URL, tableUri);
@@ -163,4 +163,24 @@ public class CatalogStorageTest {
Assert.assertEquals(uriTemplate, table.getUriTemplate(LocationType.DATA));
Assert.assertEquals(table.getUriTemplate(), table.getUriTemplate(LocationType.DATA));
}
+
+ @Test
+ public void testToPartitionFilter() throws Exception {
+ final String catalogUrl = "thrift://localhost:49083";
+ String tableUri = "catalog:clicksdb:clicks#ds=20130918;region=us";
+ String partitionFilter = "(ds='20130918';region='us')";
+
+ CatalogStorage table = new CatalogStorage(catalogUrl, tableUri);
+ Assert.assertEquals(table.toPartitionFilter(), partitionFilter);
+ }
+
+ @Test
+ public void testToPartitionAsPath() throws Exception {
+ final String catalogUrl = "thrift://localhost:49083";
+ String tableUri = "catalog:clicksdb:clicks#ds=20130918;region=us";
+ String partitionPath = "ds=20130918/region=us";
+
+ CatalogStorage table = new CatalogStorage(catalogUrl, tableUri);
+ Assert.assertEquals(table.toPartitionAsPath(), partitionPath);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
----------------------------------------------------------------------
diff --git a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
index bf5367c..4e11a98 100644
--- a/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
+++ b/feed/src/main/java/org/apache/falcon/converter/OozieFeedMapper.java
@@ -18,9 +18,12 @@
package org.apache.falcon.converter;
+import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.falcon.FalconException;
import org.apache.falcon.Tag;
+import org.apache.falcon.entity.CatalogStorage;
+import org.apache.falcon.entity.ClusterHelper;
import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.FeedHelper;
import org.apache.falcon.entity.Storage;
@@ -42,10 +45,13 @@ import org.apache.falcon.oozie.coordinator.COORDINATORAPP;
import org.apache.falcon.oozie.coordinator.SYNCDATASET;
import org.apache.falcon.oozie.coordinator.WORKFLOW;
import org.apache.falcon.oozie.workflow.WORKFLOWAPP;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.util.*;
/**
@@ -56,15 +62,8 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
private static final Logger LOG = Logger.getLogger(OozieFeedMapper.class);
- private static final int THIRTY_MINUTES = 30 * 60 * 1000;
-
- private static final String REPLICATION_COORD_TEMPLATE = "/config/coordinator/replication-coordinator.xml";
- private static final String REPLICATION_WF_TEMPLATE = "/config/workflow/replication-workflow.xml";
-
- private static final String TIMEOUT = "timeout";
- private static final String PARALLEL = "parallel";
-
private final RetentionOozieWorkflowMapper retentionMapper = new RetentionOozieWorkflowMapper();
+ private final ReplicationOozieWorkflowMapper replicationMapper = new ReplicationOozieWorkflowMapper();
public OozieFeedMapper(Feed feed) {
super(feed);
@@ -105,181 +104,22 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
if (FeedHelper.getCluster(feed, targetCluster.getName()).getType() == ClusterType.TARGET) {
String coordName = EntityUtil.getWorkflowName(Tag.REPLICATION, feed).toString();
Path basePath = getCoordPath(bundlePath, coordName);
- createReplicatonWorkflow(targetCluster, basePath, coordName);
+ replicationMapper.createReplicatonWorkflow(targetCluster, basePath, coordName);
for (org.apache.falcon.entity.v0.feed.Cluster feedCluster : feed.getClusters().getClusters()) {
if (feedCluster.getType() == ClusterType.SOURCE) {
- COORDINATORAPP coord = createAndGetCoord(feed,
+ COORDINATORAPP coord = replicationMapper.createAndGetCoord(feed,
(Cluster) ConfigurationStore.get().get(EntityType.CLUSTER, feedCluster.getName()),
- targetCluster,
- bundlePath);
+ targetCluster, bundlePath);
+
if (coord != null) {
replicationCoords.add(coord);
}
}
}
-
- }
- return replicationCoords;
- }
-
- private COORDINATORAPP createAndGetCoord(Feed feed, Cluster srcCluster, Cluster trgCluster, Path bundlePath)
- throws FalconException {
-
- COORDINATORAPP replicationCoord;
- String coordName;
- try {
- replicationCoord = getCoordinatorTemplate(REPLICATION_COORD_TEMPLATE);
- coordName = EntityUtil.getWorkflowName(Tag.REPLICATION, Arrays.asList(srcCluster.getName()),
- feed).toString();
- replicationCoord.setName(coordName);
- replicationCoord.setFrequency("${coord:" + feed.getFrequency().toString() + "}");
-
- long frequencyInMillis = ExpressionHelper.get().
- evaluate(feed.getFrequency().toString(), Long.class);
- long timeoutInMillis = frequencyInMillis * 6;
- if (timeoutInMillis < THIRTY_MINUTES) {
- timeoutInMillis = THIRTY_MINUTES;
- }
- Map<String, String> props = getEntityProperties();
- String timeout = props.get(TIMEOUT);
- if (timeout!=null) {
- try{
- timeoutInMillis= ExpressionHelper.get().
- evaluate(timeout, Long.class);
- } catch (Exception ignore) {
- LOG.error("Unable to evaluate timeout:", ignore);
- }
- }
- String parallelProp = props.get(PARALLEL);
- int parallel = 1;
- if (parallelProp != null) {
- try {
- parallel = Integer.parseInt(parallelProp);
- } catch (NumberFormatException ignore) {
- LOG.error("Unable to parse parallel:", ignore);
- }
- }
-
- replicationCoord.getControls().setTimeout(String.valueOf(timeoutInMillis / (1000 * 60)));
- replicationCoord.getControls().setThrottle(String.valueOf(timeoutInMillis / frequencyInMillis * 2));
- replicationCoord.getControls().setConcurrency(String.valueOf(parallel));
-
- Frequency replicationDelay = FeedHelper.getCluster(feed,
- srcCluster.getName()).getDelay();
- long delayInMillis=0;
- if (replicationDelay != null) {
- delayInMillis = ExpressionHelper.get().evaluate(
- replicationDelay.toString(), Long.class);
- long delayInMins = -1 * delayInMillis / (1000 * 60);
- String elExp = "${now(0," + delayInMins + ")}";
- replicationCoord.getInputEvents().getDataIn().get(0)
- .getInstance().set(0, elExp);
- replicationCoord.getOutputEvents().getDataOut().get(0)
- .setInstance(elExp);
- }
- Date srcStartDate = FeedHelper.getCluster(feed, srcCluster.getName()).getValidity().getStart();
- srcStartDate=new Date(srcStartDate.getTime()+delayInMillis);
- Date srcEndDate = FeedHelper.getCluster(feed, srcCluster.getName()).getValidity().getEnd();
- Date trgStartDate = FeedHelper.getCluster(feed, trgCluster.getName()).getValidity().getStart();
- Date trgEndDate = FeedHelper.getCluster(feed, trgCluster.getName()).getValidity().getEnd();
- trgStartDate=new Date(trgStartDate.getTime()+delayInMillis);
- if (srcStartDate.after(trgEndDate)
- || trgStartDate.after(srcEndDate)) {
- LOG.warn("Not creating replication coordinator, as the source cluster:"
- + srcCluster.getName()
- + " and target cluster: "
- + trgCluster.getName()
- + " do not have overlapping dates");
- return null;
- }
- replicationCoord.setStart(
- srcStartDate.after(trgStartDate) ? SchemaHelper.formatDateUTC(srcStartDate) : SchemaHelper
- .formatDateUTC(trgStartDate));
- replicationCoord.setEnd(
- srcEndDate.before(trgEndDate) ? SchemaHelper.formatDateUTC(srcEndDate) : SchemaHelper
- .formatDateUTC(trgEndDate));
- replicationCoord.setTimezone(feed.getTimezone().getID());
- SYNCDATASET inputDataset = (SYNCDATASET) replicationCoord.getDatasets().getDatasetOrAsyncDataset().get(0);
- SYNCDATASET outputDataset = (SYNCDATASET) replicationCoord.getDatasets().getDatasetOrAsyncDataset().get(1);
-
- final Storage srcStorage = FeedHelper.createReadOnlyStorage(srcCluster, feed);
- inputDataset.setUriTemplate(srcStorage.getUriTemplate(LocationType.DATA));
- outputDataset.setUriTemplate(FeedHelper.createStorage(feed).getUriTemplate(LocationType.DATA));
- setDatasetValues(inputDataset, feed, srcCluster);
- setDatasetValues(outputDataset, feed, srcCluster);
- if (feed.getAvailabilityFlag() == null) {
- inputDataset.setDoneFlag("");
- } else {
- inputDataset.setDoneFlag(feed.getAvailabilityFlag());
- }
-
- } catch (FalconException e) {
- throw new FalconException("Cannot unmarshall replication coordinator template", e);
- }
-
- Path wfPath = getCoordPath(bundlePath, coordName);
- replicationCoord.setAction(getReplicationWorkflowAction(srcCluster, trgCluster, wfPath, coordName));
- return replicationCoord;
- }
-
- private void setDatasetValues(SYNCDATASET dataset, Feed feed, Cluster cluster) {
- dataset.setInitialInstance(
- SchemaHelper.formatDateUTC(FeedHelper.getCluster(feed, cluster.getName()).getValidity().getStart()));
- dataset.setTimezone(feed.getTimezone().getID());
- dataset.setFrequency("${coord:" + feed.getFrequency().toString() + "}");
- }
-
- private ACTION getReplicationWorkflowAction(Cluster srcCluster, Cluster trgCluster, Path wfPath, String wfName)
- throws FalconException {
-
- ACTION replicationAction = new ACTION();
- WORKFLOW replicationWF = new WORKFLOW();
- try {
- replicationWF.setAppPath(getStoragePath(wfPath.toString()));
- Feed feed = getEntity();
-
- String srcPart = FeedHelper.normalizePartitionExpression(
- FeedHelper.getCluster(feed, srcCluster.getName()).getPartition());
- srcPart = FeedHelper.evaluateClusterExp(srcCluster, srcPart);
- String targetPart = FeedHelper.normalizePartitionExpression(
- FeedHelper.getCluster(feed, trgCluster.getName()).getPartition());
- targetPart = FeedHelper.evaluateClusterExp(trgCluster, targetPart);
-
- StringBuilder pathsWithPartitions = new StringBuilder();
- pathsWithPartitions.append("${coord:dataIn('input')}/").append(
- FeedHelper.normalizePartitionExpression(srcPart, targetPart));
-
- Map<String, String> props = createCoordDefaultConfiguration(trgCluster, wfPath, wfName);
- props.put("srcClusterName", srcCluster.getName());
- props.put("srcClusterColo", srcCluster.getColo());
- props.put(ARG.feedNames.getPropName(), feed.getName());
- props.put(ARG.feedInstancePaths.getPropName(), pathsWithPartitions.toString());
- String parts = pathsWithPartitions.toString().replaceAll("//+", "/");
- parts = StringUtils.stripEnd(parts, "/");
- props.put("sourceRelativePaths", parts);
- props.put("distcpSourcePaths", "${coord:dataIn('input')}");
- props.put("distcpTargetPaths", "${coord:dataOut('output')}");
- props.put("falconInPaths", pathsWithPartitions.toString());
- props.put("falconInputFeeds", feed.getName());
- replicationWF.setConfiguration(getCoordConfig(props));
- replicationAction.setWorkflow(replicationWF);
- } catch (Exception e) {
- throw new FalconException("Unable to create replication workflow", e);
}
- return replicationAction;
-
- }
- private void createReplicatonWorkflow(Cluster cluster, Path wfPath, String wfName) throws FalconException {
- try {
- WORKFLOWAPP repWFapp = getWorkflowTemplate(REPLICATION_WF_TEMPLATE);
- repWFapp.setName(wfName);
- addLibExtensionsToWorkflow(cluster, repWFapp, EntityType.FEED, "replication");
- marshal(cluster, repWFapp, wfPath);
- } catch(IOException e) {
- throw new FalconException("Unable to create replication workflow", e);
- }
+ return replicationCoords;
}
@Override
@@ -329,7 +169,6 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
retentionWorkflow.setAppPath(getStoragePath(wfPath.toString()));
Map<String, String> props = createCoordDefaultConfiguration(cluster, wfPath, wfName);
-
props.put("timeZone", feed.getTimezone().getID());
props.put("frequency", feed.getFrequency().getTimeUnit().name());
@@ -365,4 +204,338 @@ public class OozieFeedMapper extends AbstractOozieEntityMapper<Feed> {
}
}
}
+
+ private class ReplicationOozieWorkflowMapper {
+ private static final int THIRTY_MINUTES = 30 * 60 * 1000;
+
+ private static final String REPLICATION_COORD_TEMPLATE = "/config/coordinator/replication-coordinator.xml";
+ private static final String REPLICATION_WF_TEMPLATE = "/config/workflow/replication-workflow.xml";
+
+ private static final String TIMEOUT = "timeout";
+ private static final String PARALLEL = "parallel";
+
+ private void createReplicatonWorkflow(Cluster cluster, Path wfPath, String wfName)
+ throws FalconException {
+ try {
+ WORKFLOWAPP repWFapp = getWorkflowTemplate(REPLICATION_WF_TEMPLATE);
+ repWFapp.setName(wfName);
+ addLibExtensionsToWorkflow(cluster, repWFapp, EntityType.FEED, "replication");
+ marshal(cluster, repWFapp, wfPath);
+ } catch(IOException e) {
+ throw new FalconException("Unable to create replication workflow", e);
+ }
+ }
+
+ private COORDINATORAPP createAndGetCoord(Feed feed, Cluster srcCluster, Cluster trgCluster,
+ Path bundlePath) throws FalconException {
+ long replicationDelayInMillis = getReplicationDelayInMillis(feed, srcCluster);
+ Date sourceStartDate = getStartDate(feed, srcCluster, replicationDelayInMillis);
+ Date sourceEndDate = getEndDate(feed, srcCluster);
+
+ Date targetStartDate = getStartDate(feed, srcCluster, replicationDelayInMillis);
+ Date targetEndDate = getEndDate(feed, srcCluster);
+
+ if (noOverlapExists(sourceStartDate, sourceEndDate,
+ targetStartDate, targetEndDate)) {
+ LOG.warn("Not creating replication coordinator, as the source cluster:"
+ + srcCluster.getName()
+ + " and target cluster: "
+ + trgCluster.getName()
+ + " do not have overlapping dates");
+ return null;
+ }
+
+ COORDINATORAPP replicationCoord;
+ try {
+ replicationCoord = getCoordinatorTemplate(REPLICATION_COORD_TEMPLATE);
+ } catch (FalconException e) {
+ throw new FalconException("Cannot unmarshall replication coordinator template", e);
+ }
+
+ String coordName = EntityUtil.getWorkflowName(
+ Tag.REPLICATION, Arrays.asList(srcCluster.getName()), feed).toString();
+ String start = sourceStartDate.after(targetStartDate)
+ ? SchemaHelper.formatDateUTC(sourceStartDate) : SchemaHelper.formatDateUTC(targetStartDate);
+ String end = sourceEndDate.before(targetEndDate)
+ ? SchemaHelper.formatDateUTC(sourceEndDate) : SchemaHelper.formatDateUTC(targetEndDate);
+
+ initializeCoordAttributes(replicationCoord, coordName, feed, start, end, replicationDelayInMillis);
+ setCoordControls(feed, replicationCoord);
+
+ final Storage sourceStorage = FeedHelper.createReadOnlyStorage(srcCluster, feed);
+ initializeInputDataSet(feed, srcCluster, replicationCoord, sourceStorage);
+
+ final Storage targetStorage = FeedHelper.createStorage(trgCluster, feed);
+ initializeOutputDataSet(feed, trgCluster, replicationCoord, targetStorage);
+
+ Path wfPath = getCoordPath(bundlePath, coordName);
+ ACTION replicationWorkflowAction = getReplicationWorkflowAction(
+ srcCluster, trgCluster, wfPath, coordName, sourceStorage, targetStorage);
+ replicationCoord.setAction(replicationWorkflowAction);
+
+ return replicationCoord;
+ }
+
+ private Date getStartDate(Feed feed, Cluster cluster, long replicationDelayInMillis) {
+ Date startDate = FeedHelper.getCluster(feed, cluster.getName()).getValidity().getStart();
+ return replicationDelayInMillis == 0 ? startDate : new Date(startDate.getTime() + replicationDelayInMillis);
+ }
+
+ private Date getEndDate(Feed feed, Cluster cluster) {
+ return FeedHelper.getCluster(feed, cluster.getName()).getValidity().getEnd();
+ }
+
+ private boolean noOverlapExists(Date sourceStartDate, Date sourceEndDate,
+ Date targetStartDate, Date targetEndDate) {
+ return sourceStartDate.after(targetEndDate) || targetStartDate.after(sourceEndDate);
+ }
+
+ private void initializeCoordAttributes(COORDINATORAPP replicationCoord, String coordName,
+ Feed feed, String start, String end, long delayInMillis) {
+ replicationCoord.setName(coordName);
+ replicationCoord.setFrequency("${coord:" + feed.getFrequency().toString() + "}");
+
+ if (delayInMillis > 0) {
+ long delayInMins = -1 * delayInMillis / (1000 * 60);
+ String elExp = "${now(0," + delayInMins + ")}";
+
+ replicationCoord.getInputEvents().getDataIn().get(0).getInstance().set(0, elExp);
+ replicationCoord.getOutputEvents().getDataOut().get(0).setInstance(elExp);
+ }
+
+ replicationCoord.setStart(start);
+ replicationCoord.setEnd(end);
+ replicationCoord.setTimezone(feed.getTimezone().getID());
+ }
+
+ private long getReplicationDelayInMillis(Feed feed, Cluster srcCluster) throws FalconException {
+ Frequency replicationDelay = FeedHelper.getCluster(feed, srcCluster.getName()).getDelay();
+ long delayInMillis=0;
+ if (replicationDelay != null) {
+ delayInMillis = ExpressionHelper.get().evaluate(
+ replicationDelay.toString(), Long.class);
+ }
+
+ return delayInMillis;
+ }
+
+ private void setCoordControls(Feed feed, COORDINATORAPP replicationCoord) throws FalconException {
+ long frequencyInMillis = ExpressionHelper.get().evaluate(
+ feed.getFrequency().toString(), Long.class);
+ long timeoutInMillis = frequencyInMillis * 6;
+ if (timeoutInMillis < THIRTY_MINUTES) {
+ timeoutInMillis = THIRTY_MINUTES;
+ }
+
+ Map<String, String> props = getEntityProperties();
+ String timeout = props.get(TIMEOUT);
+ if (timeout!=null) {
+ try{
+ timeoutInMillis= ExpressionHelper.get().evaluate(timeout, Long.class);
+ } catch (Exception ignore) {
+ LOG.error("Unable to evaluate timeout:", ignore);
+ }
+ }
+ replicationCoord.getControls().setTimeout(String.valueOf(timeoutInMillis / (1000 * 60)));
+ replicationCoord.getControls().setThrottle(String.valueOf(timeoutInMillis / frequencyInMillis * 2));
+
+ String parallelProp = props.get(PARALLEL);
+ int parallel = 1;
+ if (parallelProp != null) {
+ try {
+ parallel = Integer.parseInt(parallelProp);
+ } catch (NumberFormatException ignore) {
+ LOG.error("Unable to parse parallel:", ignore);
+ }
+ }
+ replicationCoord.getControls().setConcurrency(String.valueOf(parallel));
+ }
+
+ private void initializeInputDataSet(Feed feed, Cluster srcCluster, COORDINATORAPP replicationCoord,
+ Storage sourceStorage) throws FalconException {
+ SYNCDATASET inputDataset = (SYNCDATASET)
+ replicationCoord.getDatasets().getDatasetOrAsyncDataset().get(0);
+
+ String uriTemplate = sourceStorage.getUriTemplate(LocationType.DATA);
+ if (sourceStorage.getType() == Storage.TYPE.TABLE) {
+ uriTemplate = uriTemplate.replace("thrift", "hcat"); // Oozie requires this!!!
+ }
+ inputDataset.setUriTemplate(uriTemplate);
+
+ setDatasetValues(inputDataset, feed, srcCluster);
+
+ if (feed.getAvailabilityFlag() == null) {
+ inputDataset.setDoneFlag("");
+ } else {
+ inputDataset.setDoneFlag(feed.getAvailabilityFlag());
+ }
+ }
+
+ private void initializeOutputDataSet(Feed feed, Cluster targetCluster, COORDINATORAPP replicationCoord,
+ Storage targetStorage) throws FalconException {
+ SYNCDATASET outputDataset = (SYNCDATASET)
+ replicationCoord.getDatasets().getDatasetOrAsyncDataset().get(1);
+
+ String uriTemplate = targetStorage.getUriTemplate(LocationType.DATA);
+ if (targetStorage.getType() == Storage.TYPE.TABLE) {
+ uriTemplate = uriTemplate.replace("thrift", "hcat"); // Oozie requires this!!!
+ }
+ outputDataset.setUriTemplate(uriTemplate);
+
+ setDatasetValues(outputDataset, feed, targetCluster);
+ }
+
+ private void setDatasetValues(SYNCDATASET dataset, Feed feed, Cluster cluster) {
+ dataset.setInitialInstance(SchemaHelper.formatDateUTC(
+ FeedHelper.getCluster(feed, cluster.getName()).getValidity().getStart()));
+ dataset.setTimezone(feed.getTimezone().getID());
+ dataset.setFrequency("${coord:" + feed.getFrequency().toString() + "}");
+ }
+
+ private ACTION getReplicationWorkflowAction(Cluster srcCluster, Cluster trgCluster, Path wfPath,
+ String wfName, Storage sourceStorage,
+ Storage targetStorage) throws FalconException {
+ ACTION replicationAction = new ACTION();
+ WORKFLOW replicationWF = new WORKFLOW();
+ try {
+ replicationWF.setAppPath(getStoragePath(wfPath.toString()));
+ Feed feed = getEntity();
+
+ Map<String, String> props = createCoordDefaultConfiguration(trgCluster, wfPath, wfName);
+ props.put("srcClusterName", srcCluster.getName());
+ props.put("srcClusterColo", srcCluster.getColo());
+
+ props.put("falconFeedStorageType", sourceStorage.getType().name());
+
+ String instancePaths = null;
+ if (sourceStorage.getType() == Storage.TYPE.FILESYSTEM) {
+ String pathsWithPartitions = getPathsWithPartitions(srcCluster, trgCluster, feed);
+ instancePaths = pathsWithPartitions;
+
+ propagateFileSystemCopyProperties(pathsWithPartitions, props);
+ } else if (sourceStorage.getType() == Storage.TYPE.TABLE) {
+ instancePaths = "${coord:dataIn('input')}";
+
+ props.put("shouldRecord", "false"); // todo - override until late data is handled
+ final CatalogStorage sourceTableStorage = (CatalogStorage) sourceStorage;
+ propagateTableStorageProperties(srcCluster, sourceTableStorage, props, "falconSource");
+ final CatalogStorage targetTableStorage = (CatalogStorage) targetStorage;
+ propagateTableStorageProperties(trgCluster, targetTableStorage, props, "falconTarget");
+ propagateTableCopyProperties(srcCluster, sourceTableStorage,
+ trgCluster, targetTableStorage, props);
+ setupHiveConfiguration(trgCluster, sourceTableStorage, targetTableStorage, wfPath);
+ }
+
+ propagateLateDataProperties(feed, instancePaths, props);
+
+ replicationWF.setConfiguration(getCoordConfig(props));
+ replicationAction.setWorkflow(replicationWF);
+
+ } catch (Exception e) {
+ throw new FalconException("Unable to create replication workflow", e);
+ }
+
+ return replicationAction;
+ }
+
+ private String getPathsWithPartitions(Cluster srcCluster, Cluster trgCluster,
+ Feed feed) throws FalconException {
+ String srcPart = FeedHelper.normalizePartitionExpression(
+ FeedHelper.getCluster(feed, srcCluster.getName()).getPartition());
+ srcPart = FeedHelper.evaluateClusterExp(srcCluster, srcPart);
+ String targetPart = FeedHelper.normalizePartitionExpression(
+ FeedHelper.getCluster(feed, trgCluster.getName()).getPartition());
+ targetPart = FeedHelper.evaluateClusterExp(trgCluster, targetPart);
+
+ StringBuilder pathsWithPartitions = new StringBuilder();
+ pathsWithPartitions.append("${coord:dataIn('input')}/")
+ .append(FeedHelper.normalizePartitionExpression(srcPart, targetPart));
+ return pathsWithPartitions.toString();
+ }
+
+ private void propagateFileSystemCopyProperties(String pathsWithPartitions,
+ Map<String, String> props) throws FalconException {
+ String parts = pathsWithPartitions.replaceAll("//+", "/");
+ parts = StringUtils.stripEnd(parts, "/");
+ props.put("sourceRelativePaths", parts);
+
+ props.put("distcpSourcePaths", "${coord:dataIn('input')}");
+ props.put("distcpTargetPaths", "${coord:dataOut('output')}");
+ }
+
+ private void propagateTableStorageProperties(Cluster cluster, CatalogStorage tableStorage,
+ Map<String, String> props, String prefix) {
+ props.put(prefix + "NameNode", ClusterHelper.getStorageUrl(cluster));
+ props.put(prefix + "JobTracker", ClusterHelper.getMREndPoint(cluster));
+ props.put(prefix + "HcatNode", tableStorage.getCatalogUrl());
+
+ props.put(prefix + "Database", tableStorage.getDatabase());
+ props.put(prefix + "Table", tableStorage.getTable());
+ props.put(prefix + "Partition", "${coord:dataInPartitionFilter('input', 'hive')}");
+ }
+
+ private void setupHiveConfiguration(Cluster trgCluster, CatalogStorage sourceStorage,
+ CatalogStorage targetStorage, Path wfPath) throws IOException {
+ FileSystem fs = FileSystem.get(ClusterHelper.getConfiguration(trgCluster));
+
+ // copy import export scripts to stagingDir
+ Path scriptPath = new Path(wfPath, "scripts");
+ copyHiveScript(fs, scriptPath, "/config/workflow/", "falcon-table-export.hql");
+ copyHiveScript(fs, scriptPath, "/config/workflow/", "falcon-table-import.hql");
+
+ // create hive conf to stagingDir
+ Path confPath = new Path(wfPath + "/conf");
+ createHiveConf(fs, confPath, sourceStorage.getCatalogUrl(), "falcon-source-");
+ createHiveConf(fs, confPath, targetStorage.getCatalogUrl(), "falcon-target-");
+ }
+
+ private void copyHiveScript(FileSystem fs, Path scriptPath,
+ String localScriptPath, String scriptName) throws IOException {
+ OutputStream out = null;
+ InputStream in = null;
+ try {
+ out = fs.create(new Path(scriptPath, scriptName));
+ in = OozieFeedMapper.class.getResourceAsStream(localScriptPath + scriptName);
+ IOUtils.copy(in, out);
+ } finally {
+ IOUtils.closeQuietly(in);
+ IOUtils.closeQuietly(out);
+ }
+ }
+
+ private void propagateTableCopyProperties(Cluster srcCluster, CatalogStorage sourceStorage,
+ 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, getEntity(), sourceStorage, Tag.REPLICATION)
+ + "/" + sourceDatedPartitionKey
+ + "=${coord:dataOutPartitionValue('output', '" + sourceDatedPartitionKey + "')}";
+ props.put("falconSourceStagingDir", sourceStagingDir);
+ props.put("distcpSourcePaths", sourceStagingDir + "/" + NOMINAL_TIME_EL);
+
+ // create staging dirs for import at target & set it as distcpTargetPaths
+ String targetDatedPartitionKey = targetStorage.getDatedPartitionKey();
+ String targetStagingDir =
+ FeedHelper.getStagingDir(trgCluster, getEntity(), targetStorage, Tag.REPLICATION)
+ + "/" + targetDatedPartitionKey
+ + "=${coord:dataOutPartitionValue('output', '" + targetDatedPartitionKey + "')}";
+ props.put("falconTargetStagingDir", targetStagingDir);
+ props.put("distcpTargetPaths", targetStagingDir + "/" + NOMINAL_TIME_EL);
+
+ props.put("sourceRelativePaths", "IGNORE"); // this will bot be used for Table storage.
+ }
+
+ private void propagateLateDataProperties(Feed feed, String instancePaths, Map<String, String> props) {
+ // todo these pairs are the same but used in different context
+ // late data handler - should-record action
+ props.put("falconInputFeeds", feed.getName());
+ props.put("falconInPaths", instancePaths);
+
+ // late data consumer - falcon post processing
+ props.put(ARG.feedNames.getPropName(), feed.getName());
+ props.put(ARG.feedInstancePaths.getPropName(), instancePaths);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/main/resources/config/workflow/falcon-table-export.hql
----------------------------------------------------------------------
diff --git a/feed/src/main/resources/config/workflow/falcon-table-export.hql b/feed/src/main/resources/config/workflow/falcon-table-export.hql
new file mode 100644
index 0000000..37fd1b7
--- /dev/null
+++ b/feed/src/main/resources/config/workflow/falcon-table-export.hql
@@ -0,0 +1,18 @@
+--
+-- 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.
+--
+export table ${falconSourceDatabase}.${falconSourceTable} partition ${falconSourcePartition} to '${falconSourceStagingDir}';
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/main/resources/config/workflow/falcon-table-import.hql
----------------------------------------------------------------------
diff --git a/feed/src/main/resources/config/workflow/falcon-table-import.hql b/feed/src/main/resources/config/workflow/falcon-table-import.hql
new file mode 100644
index 0000000..917b5b9
--- /dev/null
+++ b/feed/src/main/resources/config/workflow/falcon-table-import.hql
@@ -0,0 +1,18 @@
+--
+-- 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.
+--
+import table ${falconTargetDatabase}.${falconTargetTable} partition ${falconTargetPartition} from '${falconTargetStagingDir}';
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/main/resources/config/workflow/replication-workflow.xml
----------------------------------------------------------------------
diff --git a/feed/src/main/resources/config/workflow/replication-workflow.xml b/feed/src/main/resources/config/workflow/replication-workflow.xml
index d141476..306a249 100644
--- a/feed/src/main/resources/config/workflow/replication-workflow.xml
+++ b/feed/src/main/resources/config/workflow/replication-workflow.xml
@@ -22,7 +22,7 @@
<case to="recordsize">
${shouldRecord=="true"}
</case>
- <default to="replication"/>
+ <default to="replication-decision"/>
</switch>
</decision>
<action name='recordsize'>
@@ -48,9 +48,46 @@
<arg>${falconInputFeeds}</arg>
<capture-output/>
</java>
+ <ok to="replication-decision"/>
+ <error to="fail"/>
+ </action>
+ <decision name="replication-decision">
+ <switch>
+ <case to="table-export">
+ ${falconFeedStorageType == "TABLE"}
+ </case>
+ <default to="replication"/>
+ </switch>
+ </decision>
+ <!-- Table Replication - Export data and metadata to HDFS Staging from Source Hive -->
+ <action name="table-export">
+ <hive xmlns="uri:oozie:hive-action:0.2">
+ <job-tracker>${falconSourceJobTracker}</job-tracker>
+ <name-node>${falconSourceNameNode}</name-node>
+ <prepare>
+ <delete path="${falconSourceStagingDir}/${nominalTime}"/>
+ </prepare>
+ <job-xml>${wf:appPath()}/conf/falcon-source-hive-site.xml</job-xml>
+ <configuration>
+ <property>
+ <name>mapred.job.queue.name</name>
+ <value>${queueName}</value>
+ </property>
+ <property>
+ <name>oozie.launcher.mapred.job.priority</name>
+ <value>${jobPriority}</value>
+ </property>
+ </configuration>
+ <script>${wf:appPath()}/scripts/falcon-table-export.hql</script>
+ <param>falconSourceDatabase=${falconSourceDatabase}</param>
+ <param>falconSourceTable=${falconSourceTable}</param>
+ <param>falconSourcePartition=${falconSourcePartition}</param>
+ <param>falconSourceStagingDir=${falconSourceStagingDir}/${nominalTime}</param>
+ </hive>
<ok to="replication"/>
<error to="fail"/>
</action>
+ <!-- Replication action -->
<action name="replication">
<java>
<job-tracker>${jobTracker}</job-tracker>
@@ -75,8 +112,43 @@
<arg>${distcpSourcePaths}</arg>
<arg>-targetPath</arg>
<arg>${distcpTargetPaths}</arg>
+ <arg>-falconFeedStorageType</arg>
+ <arg>${falconFeedStorageType}</arg>
<file>${wf:conf("falcon.libpath")}/hadoop-distcp.jar</file>
</java>
+ <ok to="post-replication-decision"/>
+ <error to="failed-post-processing"/>
+ </action>
+ <decision name="post-replication-decision">
+ <switch>
+ <case to="table-import">
+ ${falconFeedStorageType == "TABLE"}
+ </case>
+ <default to="succeeded-post-processing"/>
+ </switch>
+ </decision>
+ <!-- Table Replication - Import data and metadata from HDFS Staging into Target Hive -->
+ <action name="table-import">
+ <hive xmlns="uri:oozie:hive-action:0.2">
+ <job-tracker>${falconTargetJobTracker}</job-tracker>
+ <name-node>${falconTargetNameNode}</name-node>
+ <job-xml>${wf:appPath()}/conf/falcon-target-hive-site.xml</job-xml>
+ <configuration>
+ <property>
+ <name>mapred.job.queue.name</name>
+ <value>${queueName}</value>
+ </property>
+ <property>
+ <name>oozie.launcher.mapred.job.priority</name>
+ <value>${jobPriority}</value>
+ </property>
+ </configuration>
+ <script>${wf:appPath()}/scripts/falcon-table-import.hql</script>
+ <param>falconTargetDatabase=${falconTargetDatabase}</param>
+ <param>falconTargetTable=${falconTargetTable}</param>
+ <param>falconTargetPartition=${falconTargetPartition}</param>
+ <param>falconTargetStagingDir=${falconTargetStagingDir}/${nominalTime}</param>
+ </hive>
<ok to="succeeded-post-processing"/>
<error to="failed-post-processing"/>
</action>
@@ -211,8 +283,8 @@
<error to="fail"/>
</action>
<kill name="fail">
- <message>Workflow failed, error
- message[${wf:errorMessage(wf:lastErrorNode())}]
+ <message>
+ Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]
</message>
</kill>
<end name='end'/>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
----------------------------------------------------------------------
diff --git a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
index 330cc63..efa923e 100644
--- a/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
+++ b/feed/src/test/java/org/apache/falcon/converter/OozieFeedMapperTest.java
@@ -17,14 +17,6 @@
*/
package org.apache.falcon.converter;
-import java.util.Calendar;
-import java.util.Collection;
-import java.util.List;
-
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBElement;
-import javax.xml.bind.Unmarshaller;
-
import org.apache.falcon.FalconException;
import org.apache.falcon.cluster.util.EmbeddedCluster;
import org.apache.falcon.entity.ClusterHelper;
@@ -49,6 +41,13 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.JAXBElement;
+import javax.xml.bind.Unmarshaller;
+import java.util.Calendar;
+import java.util.Collection;
+import java.util.List;
+
/**
* Tests for Oozie workflow definition for feed replication & retention.
*/
@@ -59,10 +58,12 @@ public class OozieFeedMapperTest {
private Cluster srcCluster;
private Cluster trgCluster;
private Feed feed;
+ private Feed tableFeed;
private static final String SRC_CLUSTER_PATH = "/src-cluster.xml";
private static final String TRG_CLUSTER_PATH = "/trg-cluster.xml";
private static final String FEED = "/feed.xml";
+ private static final String TABLE_FEED = "/table-replication-feed.xml";
@BeforeClass
public void setUpDFS() throws Exception {
@@ -79,6 +80,8 @@ public class OozieFeedMapperTest {
trgCluster = (Cluster) storeEntity(EntityType.CLUSTER, TRG_CLUSTER_PATH, trgHdfsUrl);
feed = (Feed) storeEntity(EntityType.FEED, FEED, null);
+
+ tableFeed = (Feed) storeEntity(EntityType.FEED, TABLE_FEED, null);
}
protected Entity storeEntity(EntityType type, String template, String writeEndpoint) throws Exception {
@@ -114,7 +117,7 @@ public class OozieFeedMapperTest {
}
@Test
- public void testFeedCoords() throws Exception {
+ public void testReplicationCoordsForFSStorage() throws Exception {
OozieFeedMapper feedMapper = new OozieFeedMapper(feed);
List<COORDINATORAPP> coords = feedMapper.getCoordinators(trgCluster,
new Path("/projects/falcon/"));
@@ -145,9 +148,8 @@ public class OozieFeedMapperTest {
Assert.assertEquals("${coord:minutes(20)}",
outputDataset.getFrequency());
Assert.assertEquals("output-dataset", outputDataset.getName());
- Assert.assertEquals(
- "${nameNode}"
- + "/examples/input-data/rawLogs/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}",
+ Assert.assertEquals(ClusterHelper.getStorageUrl(trgCluster)
+ + "/examples/input-data/rawLogs/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}",
outputDataset.getUriTemplate());
String inEventName =coord.getInputEvents().getDataIn().get(0).getName();
String inEventDataset =coord.getInputEvents().getDataIn().get(0).getDataset();
@@ -195,6 +197,69 @@ public class OozieFeedMapperTest {
}
@Test
+ public void testReplicationCoordsForTableStorage() throws Exception {
+ OozieFeedMapper feedMapper = new OozieFeedMapper(tableFeed);
+ List<COORDINATORAPP> coords = feedMapper.getCoordinators(
+ trgCluster, new Path("/projects/falcon/"));
+ COORDINATORAPP coord = coords.get(0);
+
+ Assert.assertEquals("2010-01-01T00:40Z", coord.getStart());
+ Assert.assertEquals("${nameNode}/projects/falcon/REPLICATION",
+ coord.getAction().getWorkflow().getAppPath());
+ Assert.assertEquals("FALCON_FEED_REPLICATION_" + tableFeed.getName() + "_"
+ + srcCluster.getName(), coord.getName());
+ Assert.assertEquals("${coord:minutes(20)}", coord.getFrequency());
+
+ SYNCDATASET inputDataset = (SYNCDATASET) coord.getDatasets()
+ .getDatasetOrAsyncDataset().get(0);
+ Assert.assertEquals("${coord:minutes(20)}", inputDataset.getFrequency());
+ Assert.assertEquals("input-dataset", inputDataset.getName());
+
+ String sourceRegistry = ClusterHelper.getInterface(srcCluster, Interfacetype.REGISTRY).getEndpoint();
+ sourceRegistry = sourceRegistry.replace("thrift", "hcat");
+ Assert.assertEquals(inputDataset.getUriTemplate(),
+ sourceRegistry + "/source_db/source_clicks_table/ds=${YEAR}${MONTH}${DAY};region=${region}");
+
+ SYNCDATASET outputDataset = (SYNCDATASET) coord.getDatasets()
+ .getDatasetOrAsyncDataset().get(1);
+ Assert.assertEquals(outputDataset.getFrequency(), "${coord:minutes(20)}");
+ Assert.assertEquals("output-dataset", outputDataset.getName());
+
+ String targetRegistry = ClusterHelper.getInterface(trgCluster, Interfacetype.REGISTRY).getEndpoint();
+ targetRegistry = targetRegistry.replace("thrift", "hcat");
+ Assert.assertEquals(outputDataset.getUriTemplate(),
+ targetRegistry + "/target_db/target_clicks_table/ds=${YEAR}${MONTH}${DAY};region=${region}");
+
+ String inEventName =coord.getInputEvents().getDataIn().get(0).getName();
+ String inEventDataset =coord.getInputEvents().getDataIn().get(0).getDataset();
+ String inEventInstance = coord.getInputEvents().getDataIn().get(0).getInstance().get(0);
+ Assert.assertEquals("input", inEventName);
+ Assert.assertEquals("input-dataset", inEventDataset);
+ Assert.assertEquals("${now(0,-40)}", inEventInstance);
+
+ String outEventInstance = coord.getOutputEvents().getDataOut().get(0).getInstance();
+ Assert.assertEquals("${now(0,-40)}", outEventInstance);
+
+ // assert FS staging area
+ String wfPath = coord.getAction().getWorkflow().getAppPath().replace("${nameNode}", "");
+ final FileSystem fs = trgMiniDFS.getFileSystem();
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts")));
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts/falcon-table-export.hql")));
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts/falcon-table-import.hql")));
+
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/conf")));
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/conf/falcon-source-hive-site.xml")));
+ Assert.assertTrue(fs.exists(new Path(wfPath + "/conf/falcon-target-hive-site.xml")));
+
+ for (Property prop : coord.getAction().getWorkflow().getConfiguration().getProperty()) {
+ if (prop.getName().equals("shouldRecord")) {
+ Assert.assertEquals(prop.getValue(), "false");
+ break;
+ }
+ }
+ }
+
+ @Test
public void testRetentionCoords() throws FalconException {
org.apache.falcon.entity.v0.feed.Cluster cluster = FeedHelper.getCluster(feed, srcCluster.getName());
final Calendar instance = Calendar.getInstance();
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/test/resources/src-cluster.xml
----------------------------------------------------------------------
diff --git a/feed/src/test/resources/src-cluster.xml b/feed/src/test/resources/src-cluster.xml
index 75d8ed0..730f8d2 100644
--- a/feed/src/test/resources/src-cluster.xml
+++ b/feed/src/test/resources/src-cluster.xml
@@ -27,7 +27,7 @@
version="3.1"/>
<interface type="messaging" endpoint="tcp://localhost:61616?daemon=true"
version="5.1.6"/>
- <interface type="registry" endpoint="Hcat" version="1"/>
+ <interface type="registry" endpoint="thrift://localhost:49093" version="1"/>
</interfaces>
<locations>
<location name="temp" path="/tmp"/>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/test/resources/table-replication-feed.xml
----------------------------------------------------------------------
diff --git a/feed/src/test/resources/table-replication-feed.xml b/feed/src/test/resources/table-replication-feed.xml
new file mode 100644
index 0000000..4c610f6
--- /dev/null
+++ b/feed/src/test/resources/table-replication-feed.xml
@@ -0,0 +1,42 @@
+<!--
+ 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.
+ -->
+<feed description="clicks log" name="raw-logs-table" xmlns="uri:falcon:feed:0.1">
+
+ <frequency>minutes(20)</frequency>
+ <timezone>UTC</timezone>
+
+ <clusters>
+ <cluster name="corp1" type="source" delay="minutes(40)">
+ <validity start="2010-01-01T00:00Z" end="2010-01-01T02:00Z"/>
+ <retention limit="minutes(5)" action="delete"/>
+ <!-- Limit can be in Time or Instances 100, Action ENUM DELETE,ARCHIVE -->
+ </cluster>
+ <cluster name="corp2" type="target">
+ <validity start="2010-01-01T00:00Z" end="2010-01-01T02:00Z"/>
+ <retention limit="minutes(7)" action="delete"/>
+ <!-- Limit can be in Time or Instances 100, Action ENUM DELETE,ARCHIVE -->
+ <table uri="catalog:target_db:target_clicks_table#ds=${YEAR}${MONTH}${DAY};region=${region}" />
+ </cluster>
+ </clusters>
+
+ <table uri="catalog:source_db:source_clicks_table#ds=${YEAR}${MONTH}${DAY};region=${region}" />
+
+ <ACL owner="testuser" group="group" permission="0x755"/>
+ <schema location="/schema/clicks" provider="protobuf"/>
+
+</feed>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/feed/src/test/resources/trg-cluster.xml
----------------------------------------------------------------------
diff --git a/feed/src/test/resources/trg-cluster.xml b/feed/src/test/resources/trg-cluster.xml
index 9a99b62..8260fda 100644
--- a/feed/src/test/resources/trg-cluster.xml
+++ b/feed/src/test/resources/trg-cluster.xml
@@ -27,7 +27,7 @@
version="3.1"/>
<interface type="messaging" endpoint="tcp://localhost:61616?daemon=true"
version="5.1.6"/>
- <interface type="registry" endpoint="Hcat" version="1"/>
+ <interface type="registry" endpoint="thrift://localhost:59093" version="1"/>
</interfaces>
<locations>
<location name="temp" path="/tmp"/>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
----------------------------------------------------------------------
diff --git a/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java b/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
index 5ee55c2..5438b6f 100644
--- a/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
+++ b/oozie/src/main/java/org/apache/falcon/converter/AbstractOozieEntityMapper.java
@@ -40,10 +40,12 @@ import org.apache.falcon.oozie.workflow.WORKFLOWAPP;
import org.apache.falcon.service.FalconPathFilter;
import org.apache.falcon.service.SharedLibraryHostingService;
import org.apache.falcon.util.StartupProperties;
+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.permission.FsPermission;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.log4j.Logger;
import org.apache.oozie.client.OozieClient;
@@ -380,4 +382,19 @@ public abstract class AbstractOozieEntityMapper<T extends Entity> {
IOUtils.closeQuietly(resourceAsStream);
}
}
+
+ protected void createHiveConf(FileSystem fs, Path confPath, String metastoreUrl,
+ String prefix) throws IOException {
+ Configuration hiveConf = new Configuration(false);
+ hiveConf.set(HiveConf.ConfVars.METASTOREURIS.varname, metastoreUrl);
+ hiveConf.set("hive.metastore.local", "false");
+
+ OutputStream out = null;
+ try {
+ out = fs.create(new Path(confPath, prefix + "hive-site.xml"));
+ hiveConf.writeXml(out);
+ } finally {
+ IOUtils.closeQuietly(out);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/oozie/src/main/resources/oozie-workflow-0.3.xsd
----------------------------------------------------------------------
diff --git a/oozie/src/main/resources/oozie-workflow-0.3.xsd b/oozie/src/main/resources/oozie-workflow-0.3.xsd
index 858e34c..012d9f7 100644
--- a/oozie/src/main/resources/oozie-workflow-0.3.xsd
+++ b/oozie/src/main/resources/oozie-workflow-0.3.xsd
@@ -126,11 +126,11 @@
<xs:element name="sub-workflow" type="workflow:SUB-WORKFLOW" minOccurs="1" maxOccurs="1"/>
<xs:element name="fs" type="workflow:FS" minOccurs="1" maxOccurs="1"/>
<xs:element name="java" type="workflow:JAVA" minOccurs="1" maxOccurs="1"/>
- <!-- <xs:any namespace="##other" minOccurs="1" maxOccurs="1"/>-->
+ <xs:any namespace="##other" minOccurs="1" maxOccurs="1"/>
</xs:choice>
<xs:element name="ok" type="workflow:ACTION_TRANSITION" minOccurs="1" maxOccurs="1"/>
<xs:element name="error" type="workflow:ACTION_TRANSITION" minOccurs="1" maxOccurs="1"/>
- <xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>
+ <!--<xs:any namespace="uri:oozie:sla:0.1" minOccurs="0" maxOccurs="1"/>-->
</xs:sequence>
<xs:attribute name="name" type="workflow:IDENTIFIER" use="required"/>
</xs:complexType>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b04a1be..d0c3cfd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -979,6 +979,9 @@
</goals>
</execution>
</executions>
+ <configuration>
+ <skip>${skipCheck}</skip>
+ </configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/replication/pom.xml
----------------------------------------------------------------------
diff --git a/replication/pom.xml b/replication/pom.xml
index 200cb2f..07175e6 100644
--- a/replication/pom.xml
+++ b/replication/pom.xml
@@ -68,6 +68,11 @@
<dependencies>
<dependency>
+ <groupId>org.apache.falcon</groupId>
+ <artifactId>falcon-common</artifactId>
+ </dependency>
+
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
----------------------------------------------------------------------
diff --git a/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java b/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
index 5c51b5f..17227bf 100644
--- a/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
+++ b/replication/src/main/java/org/apache/falcon/replication/FeedReplicator.java
@@ -18,6 +18,7 @@
package org.apache.falcon.replication;
import org.apache.commons.cli.*;
+import org.apache.falcon.entity.Storage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
@@ -48,8 +49,8 @@ public class FeedReplicator extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
-
- DistCpOptions options = getDistCpOptions(args);
+ CommandLine cmd = getCommand(args);
+ DistCpOptions options = getDistCpOptions(cmd);
Configuration conf = this.getConf();
// inject wf configs
@@ -60,10 +61,68 @@ public class FeedReplicator extends Configured implements Tool {
+ confPath.getFileSystem(conf).exists(confPath));
conf.addResource(confPath);
- DistCp distCp = new CustomReplicator(conf, options);
+ String falconFeedStorageType = cmd.getOptionValue("falconFeedStorageType").trim();
+ Storage.TYPE feedStorageType = Storage.TYPE.valueOf(falconFeedStorageType);
+
+ DistCp distCp = (feedStorageType == Storage.TYPE.FILESYSTEM)
+ ? new CustomReplicator(conf, options)
+ : new DistCp(conf, options);
LOG.info("Started DistCp");
distCp.execute();
+ if (feedStorageType == Storage.TYPE.FILESYSTEM) {
+ executePostProcessing(options); // this only applies for FileSystem Storage.
+ }
+
+ LOG.info("Completed DistCp");
+ return 0;
+ }
+
+ protected CommandLine getCommand(String[] args) throws ParseException {
+ Options options = new Options();
+ Option opt = new Option("maxMaps", true,
+ "max number of maps to use for this copy");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ opt = new Option("sourcePaths", true,
+ "comma separtated list of source paths to be copied");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ opt = new Option("targetPath", true, "target path");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ opt = new Option("falconFeedStorageType", true, "feed storage type");
+ opt.setRequired(true);
+ options.addOption(opt);
+
+ return new GnuParser().parse(options, args);
+ }
+
+ protected DistCpOptions getDistCpOptions(CommandLine cmd) {
+ String[] paths = cmd.getOptionValue("sourcePaths").trim().split(",");
+ List<Path> srcPaths = getPaths(paths);
+ String trgPath = cmd.getOptionValue("targetPath").trim();
+
+ DistCpOptions distcpOptions = new DistCpOptions(srcPaths, new Path(trgPath));
+ distcpOptions.setSyncFolder(true);
+ distcpOptions.setBlocking(true);
+ distcpOptions.setMaxMaps(Integer.valueOf(cmd.getOptionValue("maxMaps")));
+
+ return distcpOptions;
+ }
+
+ private List<Path> getPaths(String[] paths) {
+ List<Path> listPaths = new ArrayList<Path>();
+ for (String path : paths) {
+ listPaths.add(new Path(path));
+ }
+ return listPaths;
+ }
+
+ private void executePostProcessing(DistCpOptions options) throws IOException {
Path targetPath = options.getTargetPath();
FileSystem fs = targetPath.getFileSystem(getConf());
List<Path> inPaths = options.getSourcePaths();
@@ -87,8 +146,6 @@ public class FeedReplicator extends Configured implements Tool {
LOG.info("No files present in path: "
+ new Path(targetPath.toString() + "/" + fixedPath).toString());
}
- LOG.info("Completed DistCp");
- return 0;
}
private String getFixedPath(String relativePath) throws IOException {
@@ -113,44 +170,4 @@ public class FeedReplicator extends Configured implements Tool {
String result = resultBuffer.toString();
return result.substring(0, result.lastIndexOf('/'));
}
-
- public DistCpOptions getDistCpOptions(String[] args) throws ParseException {
- Options options = new Options();
- Option opt;
- opt = new Option("maxMaps", true,
- "max number of maps to use for this copy");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("sourcePaths", true,
- "comma separtated list of source paths to be copied");
- opt.setRequired(true);
- options.addOption(opt);
-
- opt = new Option("targetPath", true, "target path");
- opt.setRequired(true);
- options.addOption(opt);
-
- CommandLine cmd = new GnuParser().parse(options, args);
- String[] paths = cmd.getOptionValue("sourcePaths").trim().split(",");
- List<Path> srcPaths = getPaths(paths);
- String trgPath = cmd.getOptionValue("targetPath").trim();
-
- DistCpOptions distcpOptions = new DistCpOptions(srcPaths, new Path(
- trgPath));
- distcpOptions.setSyncFolder(true);
- distcpOptions.setBlocking(true);
- distcpOptions
- .setMaxMaps(Integer.valueOf(cmd.getOptionValue("maxMaps")));
-
- return distcpOptions;
- }
-
- private List<Path> getPaths(String[] paths) {
- List<Path> listPaths = new ArrayList<Path>();
- for (String path : paths) {
- listPaths.add(new Path(path));
- }
- return listPaths;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/replication/src/test/java/org/apache/falcon/replication/FeedReplicatorTest.java
----------------------------------------------------------------------
diff --git a/replication/src/test/java/org/apache/falcon/replication/FeedReplicatorTest.java b/replication/src/test/java/org/apache/falcon/replication/FeedReplicatorTest.java
index 67795b5..b8b39ad 100644
--- a/replication/src/test/java/org/apache/falcon/replication/FeedReplicatorTest.java
+++ b/replication/src/test/java/org/apache/falcon/replication/FeedReplicatorTest.java
@@ -17,6 +17,8 @@
*/
package org.apache.falcon.replication;
+import org.apache.commons.cli.CommandLine;
+import org.apache.falcon.entity.Storage;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.tools.DistCpOptions;
import org.testng.Assert;
@@ -38,11 +40,17 @@ public class FeedReplicatorTest {
* <arg>-sourcePaths</arg><arg>${distcpSourcePaths}</arg>
* <arg>-targetPath</arg><arg>${distcpTargetPaths}</arg>
*/
+ final String[] args = {
+ "true",
+ "-maxMaps", "5",
+ "-sourcePaths", "hdfs://localhost:8020/tmp/",
+ "-targetPath", "hdfs://localhost1:8020/tmp/",
+ "-falconFeedStorageType", Storage.TYPE.FILESYSTEM.name(),
+ };
+
FeedReplicator replicator = new FeedReplicator();
- DistCpOptions options = replicator.getDistCpOptions(new String[] {
- "true", "-maxMaps", "5", "-sourcePaths",
- "hdfs://localhost:8020/tmp/", "-targetPath",
- "hdfs://localhost1:8020/tmp/", });
+ CommandLine cmd = replicator.getCommand(args);
+ DistCpOptions options = replicator.getDistCpOptions(cmd);
List<Path> srcPaths = new ArrayList<Path>();
srcPaths.add(new Path("hdfs://localhost:8020/tmp/"));
http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/353574c6/test-util/src/main/java/org/apache/falcon/cluster/util/EmbeddedCluster.java
----------------------------------------------------------------------
diff --git a/test-util/src/main/java/org/apache/falcon/cluster/util/EmbeddedCluster.java b/test-util/src/main/java/org/apache/falcon/cluster/util/EmbeddedCluster.java
index 36365c2..a40a030 100644
--- a/test-util/src/main/java/org/apache/falcon/cluster/util/EmbeddedCluster.java
+++ b/test-util/src/main/java/org/apache/falcon/cluster/util/EmbeddedCluster.java
@@ -111,22 +111,21 @@ public class EmbeddedCluster {
interfaces.getInterfaces().add(newInterface(Interfacetype.WRITE, fsUrl, "0.1"));
interfaces.getInterfaces().add(newInterface(Interfacetype.EXECUTE,
conf.get("mapred.job.tracker"), "0.1"));
- interfaces
- .getInterfaces()
- .add(newInterface(
- Interfacetype.MESSAGING,
- "vm://localhost",
- "0.1"));
+ interfaces.getInterfaces().add(
+ newInterface(Interfacetype.REGISTRY, "localhost", "0.1"));
+ interfaces.getInterfaces().add(
+ newInterface(Interfacetype.MESSAGING, "vm://localhost", "0.1"));
+
clusterEntity.setInterfaces(interfaces);
Location location = new Location();
location.setName("staging");
- location.setPath("/workflow/staging");
+ location.setPath("/projects/falcon/staging");
Locations locs = new Locations();
locs.getLocations().add(location);
location = new Location();
location.setName("working");
- location.setPath("/workflow/work");
+ location.setPath("/projects/falcon/working");
locs.getLocations().add(location);
clusterEntity.setLocations(locs);
}