code cleanup for adaptors/factories
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
index 355382a..d57d424 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
@@ -445,8 +445,7 @@
datasetDetails = new FeedDatasetDetails(InternalDatasetDetails.FileStructure.BTREE,
InternalDatasetDetails.PartitioningStrategy.HASH, partitioningExprs, partitioningExprs,
- ngName, adapter, configuration, signature,
- FeedDatasetDetails.FeedState.INACTIVE.toString(), null, null);
+ ngName, adapter, configuration, signature);
break;
}
}
@@ -1355,7 +1354,7 @@
try {
ControlFeedStatement cfs = (ControlFeedStatement) stmt;
String dataverseName = getActiveDataverseName(cfs.getDataverseName());
-
+
String datasetName = cfs.getDatasetName().getValue();
Dataset dataset = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(),
dataverseName, cfs.getDatasetName().getValue());
diff --git a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_01/feeds_01.1.adm b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_01/feeds_01.1.adm
index 4e3714c..fb4e84e 100644
--- a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_01/feeds_01.1.adm
+++ b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_01/feeds_01.1.adm
@@ -1 +1 @@
-{ "DataverseName": "feeds", "DatasetName": "TweetFeed", "DataTypeName": "TweetType", "DatasetType": "FEED", "InternalDetails": null, "ExternalDetails": null, "FeedDetails": { "FileStructure": "BTREE", "PartitioningStrategy": "HASH", "PartitioningKey": [ "id" ], "PrimaryKey": [ "id" ], "GroupName": "DEFAULT_NG_ALL_NODES", "DatasourceAdapter": "edu.uci.ics.asterix.tools.external.data.RateControlledFileSystemBasedAdapterFactory", "Properties": [ { "Name": "output-type-name", "Value": "TweetType" }, { "Name": "fs", "Value": "localfs" }, { "Name": "path", "Value": "nc1://data/twitter/obamatweets.adm" }, { "Name": "format", "Value": "adm" }, { "Name": "tuple-interval", "Value": "10" } ], "Function": null, "Status": "INACTIVE" }, "Hints": {{ }}, "Timestamp": "Tue Jan 29 19:07:24 PST 2013" }
+{ "DataverseName": "feeds", "DatasetName": "TweetFeed", "DataTypeName": "TweetType", "DatasetType": "FEED", "InternalDetails": null, "ExternalDetails": null, "FeedDetails": { "FileStructure": "BTREE", "PartitioningStrategy": "HASH", "PartitioningKey": [ "id" ], "PrimaryKey": [ "id" ], "GroupName": "DEFAULT_NG_ALL_NODES", "DatasourceAdapter": "edu.uci.ics.asterix.tools.external.data.RateControlledFileSystemBasedAdapterFactory", "Properties": [ { "Name": "output-type-name", "Value": "TweetType" }, { "Name": "fs", "Value": "localfs" }, { "Name": "path", "Value": "nc1://data/twitter/obamatweets.adm" }, { "Name": "format", "Value": "adm" }, { "Name": "tuple-interval", "Value": "10" } ], "Function": null }, "Hints": {{ }}, "Timestamp": "Tue Jun 11 13:56:43 PDT 2013", "DatasetId": 696, "PendingOp": 0 }
diff --git a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_02/feeds_02.1.adm b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_02/feeds_02.1.adm
index 9720960..e69de29 100644
--- a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_02/feeds_02.1.adm
+++ b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_02/feeds_02.1.adm
@@ -1,12 +0,0 @@
-{ "id": "nc1:1", "username": "BronsonMike", "location": "", "text": "@GottaLaff @reutersus Christie and obama just foul weather friends", "timestamp": "Thu Dec 06 16:53:06 PST 2012" }
-{ "id": "nc1:100", "username": "KidrauhlProuds", "location": "", "text": "RT @01Direclieber: A filha do Michael Jackson uma Belieber,a filha do Eminem e uma Belieber,as filhas de Obama sao Beliebers, e a filha do meu pai e Belieber", "timestamp": "Thu Dec 06 16:53:16 PST 2012" }
-{ "id": "nc1:102", "username": "jaysauce82", "location": "", "text": "Not voting for President Obama #BadDecision", "timestamp": "Thu Dec 06 16:53:16 PST 2012" }
-{ "id": "nc1:104", "username": "princeofsupras", "location": "", "text": "RT @01Direclieber: A filha do Michael Jackson e uma Belieber,a filha do Eminem e uma Belieber,as filhas de Obama sao Beliebers, e a filha do meu pai e Belieber", "timestamp": "Thu Dec 06 16:53:15 PST 2012" }
-{ "id": "nc1:106", "username": "GulfDogs", "location": "", "text": "Obama Admin Knew Libyan Terrorists Had US-Provided Weaponsteaparty #tcot #ccot #NewGuards #BreitbartArmy #patriotwttp://t.co/vJxzrQUE", "timestamp": "Thu Dec 06 16:53:14 PST 2012" }
-{ "id": "nc1:108", "username": "Laugzpz", "location": "", "text": "@AlfredoJalife Maestro Obama se hace de la vista gorda, es un acuerdo de siempre creo yo.", "timestamp": "Thu Dec 06 16:53:14 PST 2012" }
-{ "id": "nc1:11", "username": "magarika", "location": "", "text": "RT @ken24xavier: Obama tells SOROS - our plan is ALMOST finished http://t.co/WvzK0GtU", "timestamp": "Thu Dec 06 16:53:05 PST 2012" }
-{ "id": "nc1:111", "username": "ToucanMall", "location": "", "text": "RT @WorldWar3Watch: Michelle Obama Gets More Grammy Nominations Than Justin ... #Obama #WW3 http://t.co/0Wv2GKij", "timestamp": "Thu Dec 06 16:53:13 PST 2012" }
-{ "id": "nc1:113", "username": "ToucanMall", "location": "", "text": "RT @ObamaPalooza: Tiffany Shared What $2,000 Meant to Her ... and the President Stopped by to Talk About It http://t.co/sgT7lsNV #Obama", "timestamp": "Thu Dec 06 16:53:12 PST 2012" }
-{ "id": "nc1:115", "username": "thewildpitch", "location": "", "text": "RT @RevkahJC: Dennis Miller: Obama Should Just Say He Wants To Tax Successful People http://t.co/Ihlemy9Y", "timestamp": "Thu Dec 06 16:53:11 PST 2012" }
-{ "id": "nc1:117", "username": "Rnugent24", "location": "", "text": "RT @ConservativeQuo: unemployment is above 8% again. I wonder how long it will take for Obama to start blaming Bush? 3-2-1 #tcot #antiobama", "timestamp": "Thu Dec 06 16:53:10 PST 2012" }
-{ "id": "nc1:119", "username": "ToucanMall", "location": "", "text": "RT @Newitrsdotcom: I hope #Obama will win re-election... Other four years without meaningless #wars", "timestamp": "Thu Dec 06 16:53:09 PST 2012" }
diff --git a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
index 8011e4b..d8a0c3f 100644
--- a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
+++ b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
@@ -1 +1 @@
-{ "DataverseName": "feeds", "DatasetName": "TweetFeed", "DataTypeName": "TweetType", "DatasetType": "FEED", "InternalDetails": null, "ExternalDetails": null, "FeedDetails": { "FileStructure": "BTREE", "PartitioningStrategy": "HASH", "PartitioningKey": [ "id" ], "PrimaryKey": [ "id" ], "GroupName": "DEFAULT_NG_ALL_NODES", "DatasourceAdapter": "edu.uci.ics.asterix.tools.external.data.RateControlledFileSystemBasedAdapterFactory", "Properties": [ { "Name": "output-type-name", "Value": "TweetType" }, { "Name": "fs", "Value": "localfs" }, { "Name": "path", "Value": "nc1://data/twitter/obamatweets.adm" }, { "Name": "format", "Value": "adm" }, { "Name": "tuple-interval", "Value": "10" } ], "Function": "feeds.feed_processor@1", "Status": "INACTIVE" }, "Hints": {{ }}, "Timestamp": "Tue Jan 29 19:08:49 PST 2013" }
+{ "DataverseName": "feeds", "DatasetName": "TweetFeed", "DataTypeName": "TweetType", "DatasetType": "FEED", "InternalDetails": null, "ExternalDetails": null, "FeedDetails": { "FileStructure": "BTREE", "PartitioningStrategy": "HASH", "PartitioningKey": [ "id" ], "PrimaryKey": [ "id" ], "GroupName": "DEFAULT_NG_ALL_NODES", "DatasourceAdapter": "edu.uci.ics.asterix.tools.external.data.RateControlledFileSystemBasedAdapterFactory", "Properties": [ { "Name": "output-type-name", "Value": "TweetType" }, { "Name": "fs", "Value": "localfs" }, { "Name": "path", "Value": "nc1://data/twitter/obamatweets.adm" }, { "Name": "format", "Value": "adm" }, { "Name": "tuple-interval", "Value": "10" } ], "Function": "feeds.feed_processor@1" }, "Hints": {{ }}, "Timestamp": "Tue Jun 11 12:28:32 PDT 2013", "DatasetId": 698, "PendingOp": 0 }
diff --git a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_04/feeds_04.1.adm b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_04/feeds_04.1.adm
index 2567483..e69de29 100644
--- a/asterix-app/src/test/resources/runtimets/results/feeds/feeds_04/feeds_04.1.adm
+++ b/asterix-app/src/test/resources/runtimets/results/feeds/feeds_04/feeds_04.1.adm
@@ -1,11 +0,0 @@
-{ "id": "nc1:1", "username": "BronsonMike", "location": "", "text": "@GottaLaff @reutersus Christie and obama just foul weather friends", "timestamp": "Thu Dec 06 16:53:06 PST 2012" }
-{ "id": "nc1:100", "username": "KidrauhlProuds", "location": "", "text": "RT @01Direclieber: A filha do Michael Jackson uma Belieber,a filha do Eminem e uma Belieber,as filhas de Obama sao Beliebers, e a filha do meu pai e Belieber", "timestamp": "Thu Dec 06 16:53:16 PST 2012" }
-{ "id": "nc1:102", "username": "jaysauce82", "location": "", "text": "Not voting for President Obama #BadDecision", "timestamp": "Thu Dec 06 16:53:16 PST 2012" }
-{ "id": "nc1:104", "username": "princeofsupras", "location": "", "text": "RT @01Direclieber: A filha do Michael Jackson e uma Belieber,a filha do Eminem e uma Belieber,as filhas de Obama sao Beliebers, e a filha do meu pai e Belieber", "timestamp": "Thu Dec 06 16:53:15 PST 2012" }
-{ "id": "nc1:106", "username": "GulfDogs", "location": "", "text": "Obama Admin Knew Libyan Terrorists Had US-Provided Weaponsteaparty #tcot #ccot #NewGuards #BreitbartArmy #patriotwttp://t.co/vJxzrQUE", "timestamp": "Thu Dec 06 16:53:14 PST 2012" }
-{ "id": "nc1:108", "username": "Laugzpz", "location": "", "text": "@AlfredoJalife Maestro Obama se hace de la vista gorda, es un acuerdo de siempre creo yo.", "timestamp": "Thu Dec 06 16:53:14 PST 2012" }
-{ "id": "nc1:11", "username": "magarika", "location": "", "text": "RT @ken24xavier: Obama tells SOROS - our plan is ALMOST finished http://t.co/WvzK0GtU", "timestamp": "Thu Dec 06 16:53:05 PST 2012" }
-{ "id": "nc1:111", "username": "ToucanMall", "location": "", "text": "RT @WorldWar3Watch: Michelle Obama Gets More Grammy Nominations Than Justin ... #Obama #WW3 http://t.co/0Wv2GKij", "timestamp": "Thu Dec 06 16:53:13 PST 2012" }
-{ "id": "nc1:113", "username": "ToucanMall", "location": "", "text": "RT @ObamaPalooza: Tiffany Shared What $2,000 Meant to Her ... and the President Stopped by to Talk About It http://t.co/sgT7lsNV #Obama", "timestamp": "Thu Dec 06 16:53:12 PST 2012" }
-{ "id": "nc1:115", "username": "thewildpitch", "location": "", "text": "RT @RevkahJC: Dennis Miller: Obama Should Just Say He Wants To Tax Successful People http://t.co/Ihlemy9Y", "timestamp": "Thu Dec 06 16:53:11 PST 2012" }
-{ "id": "nc1:117", "username": "Rnugent24", "location": "", "text": "RT @ConservativeQuo: unemployment is above 8% again. I wonder how long it will take for Obama to start blaming Bush? 3-2-1 #tcot #antiobama", "timestamp": "Thu Dec 06 16:53:10 PST 2012" }
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/CNNFeedAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/CNNFeedAdapterFactory.java
index 68eac48..465a226 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/CNNFeedAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/CNNFeedAdapterFactory.java
@@ -14,22 +14,70 @@
*/
package edu.uci.ics.asterix.external.adapter.factory;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import edu.uci.ics.asterix.external.dataset.adapter.CNNFeedAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasetAdapterFactory;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
- * A factory class for creating the @see {CNNFeedAdapter}.
+ * A factory class for creating the @see {CNNFeedAdapter}.
*/
-public class CNNFeedAdapterFactory implements ITypedDatasetAdapterFactory {
+public class CNNFeedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
+ private Map<String, Object> configuration;
+
+ private List<String> feedURLs = new ArrayList<String>();
+ private static Map<String, String> topicFeeds = new HashMap<String, String>();
+
+ public static final String KEY_RSS_URL = "topic";
+ public static final String KEY_INTERVAL = "interval";
+ public static final String TOP_STORIES = "topstories";
+ public static final String WORLD = "world";
+ public static final String US = "us";
+ public static final String SPORTS = "sports";
+ public static final String BUSINESS = "business";
+ public static final String POLITICS = "politics";
+ public static final String CRIME = "crime";
+ public static final String TECHNOLOGY = "technology";
+ public static final String HEALTH = "health";
+ public static final String ENTERNTAINMENT = "entertainemnt";
+ public static final String TRAVEL = "travel";
+ public static final String LIVING = "living";
+ public static final String VIDEO = "video";
+ public static final String STUDENT = "student";
+ public static final String POPULAR = "popular";
+ public static final String RECENT = "recent";
+
+ private void initTopics() {
+ topicFeeds.put(TOP_STORIES, "http://rss.cnn.com/rss/cnn_topstories.rss");
+ topicFeeds.put(WORLD, "http://rss.cnn.com/rss/cnn_world.rss");
+ topicFeeds.put(US, "http://rss.cnn.com/rss/cnn_us.rss");
+ topicFeeds.put(SPORTS, "http://rss.cnn.com/rss/si_topstories.rss");
+ topicFeeds.put(BUSINESS, "http://rss.cnn.com/rss/money_latest.rss");
+ topicFeeds.put(POLITICS, "http://rss.cnn.com/rss/cnn_allpolitics.rss");
+ topicFeeds.put(CRIME, "http://rss.cnn.com/rss/cnn_crime.rss");
+ topicFeeds.put(TECHNOLOGY, "http://rss.cnn.com/rss/cnn_tech.rss");
+ topicFeeds.put(HEALTH, "http://rss.cnn.com/rss/cnn_health.rss");
+ topicFeeds.put(ENTERNTAINMENT, "http://rss.cnn.com/rss/cnn_showbiz.rss");
+ topicFeeds.put(LIVING, "http://rss.cnn.com/rss/cnn_living.rss");
+ topicFeeds.put(VIDEO, "http://rss.cnn.com/rss/cnn_freevideo.rss");
+ topicFeeds.put(TRAVEL, "http://rss.cnn.com/rss/cnn_travel.rss");
+ topicFeeds.put(STUDENT, "http://rss.cnn.com/rss/cnn_studentnews.rss");
+ topicFeeds.put(POPULAR, "http://rss.cnn.com/rss/cnn_mostpopular.rss");
+ topicFeeds.put(RECENT, "http://rss.cnn.com/rss/cnn_latest.rss");
+ }
+
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration) throws Exception {
- CNNFeedAdapter cnnFeedAdapter = new CNNFeedAdapter();
- cnnFeedAdapter.configure(configuration);
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ CNNFeedAdapter cnnFeedAdapter = new CNNFeedAdapter(configuration, ctx);
return cnnFeedAdapter;
}
@@ -38,4 +86,53 @@
return "cnn_feed";
}
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.TYPED;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ String rssURLProperty = (String) configuration.get(KEY_RSS_URL);
+ if (rssURLProperty == null) {
+ throw new IllegalArgumentException("no rss url provided");
+ }
+ initializeFeedURLs(rssURLProperty);
+ }
+
+ private void initializeFeedURLs(String rssURLProperty) {
+ feedURLs.clear();
+ String[] rssTopics = rssURLProperty.split(",");
+ initTopics();
+ for (String topic : rssTopics) {
+ String feedURL = topicFeeds.get(topic);
+ if (feedURL == null) {
+ throw new IllegalArgumentException(" unknown topic :" + topic + " please choose from the following "
+ + getValidTopics());
+ }
+ feedURLs.add(feedURL);
+ }
+ }
+
+ private static String getValidTopics() {
+ StringBuilder builder = new StringBuilder();
+ for (String key : topicFeeds.keySet()) {
+ builder.append(key);
+ builder.append(" ");
+ }
+ return new String(builder);
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ AlgebricksPartitionConstraint partitionConstraint = new AlgebricksCountPartitionConstraint(feedURLs.size());
+ return new AlgebricksCountPartitionConstraint(1);
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSAdapterFactory.java
index 15f702b..5618ef8 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSAdapterFactory.java
@@ -14,6 +14,7 @@
*/
package edu.uci.ics.asterix.external.adapter.factory;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
@@ -23,10 +24,14 @@
import edu.uci.ics.asterix.external.dataset.adapter.HDFSAdapter;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericDatasetAdapterFactory;
import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.util.AsterixAppContextInfo;
+import edu.uci.ics.asterix.om.util.AsterixClusterProperties;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.ICCContext;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksException;
import edu.uci.ics.hyracks.hdfs.dataflow.ConfFactory;
import edu.uci.ics.hyracks.hdfs.dataflow.InputSplitsFactory;
import edu.uci.ics.hyracks.hdfs.scheduler.Scheduler;
@@ -35,7 +40,7 @@
* A factory class for creating an instance of HDFSAdapter
*/
@SuppressWarnings("deprecation")
-public class HDFSAdapterFactory implements IGenericDatasetAdapterFactory {
+public class HDFSAdapterFactory extends FileSystemAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String HDFS_ADAPTER_NAME = "hdfs";
@@ -53,7 +58,21 @@
private boolean executed[];
private InputSplitsFactory inputSplitsFactory;
private ConfFactory confFactory;
- private boolean setup = false;
+ private IAType atype;
+ private boolean configured = false;
+ public static Scheduler hdfsScheduler = initializeHDFSScheduler();
+
+ private static Scheduler initializeHDFSScheduler() {
+ ICCContext ccContext = AsterixAppContextInfo.getInstance().getCCApplicationContext().getCCContext();
+ Scheduler scheduler = null;
+ try {
+ scheduler = new Scheduler(ccContext.getClusterControllerInfo().getClientNetAddress(), ccContext
+ .getClusterControllerInfo().getClientNetPort());
+ } catch (HyracksException e) {
+ throw new IllegalStateException("Cannot obtain hdfs scheduler");
+ }
+ return scheduler;
+ }
private static final Map<String, String> formatClassNames = initInputFormatMap();
@@ -65,30 +84,12 @@
}
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration, IAType atype) throws Exception {
- if (!setup) {
- /** set up the factory --serializable stuff --- this if-block should be called only once for each factory instance */
- configureJobConf(configuration);
- JobConf conf = configureJobConf(configuration);
- confFactory = new ConfFactory(conf);
-
- clusterLocations = (AlgebricksPartitionConstraint) configuration.get(CLUSTER_LOCATIONS);
- int numPartitions = ((AlgebricksAbsolutePartitionConstraint) clusterLocations).getLocations().length;
-
- InputSplit[] inputSplits = conf.getInputFormat().getSplits(conf, numPartitions);
- inputSplitsFactory = new InputSplitsFactory(inputSplits);
-
- Scheduler scheduler = (Scheduler) configuration.get(SCHEDULER);
- readSchedule = scheduler.getLocationConstraints(inputSplits);
- executed = new boolean[readSchedule.length];
- Arrays.fill(executed, false);
-
- setup = true;
- }
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
JobConf conf = confFactory.getConf();
InputSplit[] inputSplits = inputSplitsFactory.getSplits();
- HDFSAdapter hdfsAdapter = new HDFSAdapter(atype, readSchedule, executed, inputSplits, conf, clusterLocations);
- hdfsAdapter.configure(configuration);
+ String nodeName = ctx.getJobletContext().getApplicationContext().getNodeId();
+ HDFSAdapter hdfsAdapter = new HDFSAdapter(atype, readSchedule, executed, inputSplits, conf, nodeName,
+ parserFactory, ctx);
return hdfsAdapter;
}
@@ -108,4 +109,60 @@
return conf;
}
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.GENERIC;
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ if (!configured) {
+ throw new IllegalStateException("Adapter factory has not been configured yet");
+ }
+ return (AlgebricksPartitionConstraint) clusterLocations;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ JobConf conf = configureJobConf(configuration);
+ confFactory = new ConfFactory(conf);
+
+ clusterLocations = getClusterLocations();
+ int numPartitions = ((AlgebricksAbsolutePartitionConstraint) clusterLocations).getLocations().length;
+
+ InputSplit[] inputSplits = conf.getInputFormat().getSplits(conf, numPartitions);
+ inputSplitsFactory = new InputSplitsFactory(inputSplits);
+
+ readSchedule = hdfsScheduler.getLocationConstraints(inputSplits);
+ executed = new boolean[readSchedule.length];
+ Arrays.fill(executed, false);
+ configured = true;
+
+ atype = (IAType) configuration.get(KEY_SOURCE_DATATYPE);
+ configureFormat(atype);
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ private static AlgebricksPartitionConstraint getClusterLocations() {
+ ArrayList<String> locs = new ArrayList<String>();
+ Map<String, String[]> stores = AsterixAppContextInfo.getInstance().getMetadataProperties().getStores();
+ for (String i : stores.keySet()) {
+ String[] nodeStores = stores.get(i);
+ int numIODevices = AsterixClusterProperties.INSTANCE.getNumberOfIODevices(i);
+ for (int j = 0; j < nodeStores.length; j++) {
+ for (int k = 0; k < numIODevices; k++) {
+ locs.add(i);
+ }
+ }
+ }
+ String[] cluster = new String[locs.size()];
+ cluster = locs.toArray(cluster);
+ return new AlgebricksAbsolutePartitionConstraint(cluster);
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HiveAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HiveAdapterFactory.java
index 4fd9411..16e6ef7 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HiveAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HiveAdapterFactory.java
@@ -23,11 +23,13 @@
import edu.uci.ics.asterix.external.dataset.adapter.HDFSAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.HiveAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory.AdapterType;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericDatasetAdapterFactory;
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.hdfs.dataflow.ConfFactory;
import edu.uci.ics.hyracks.hdfs.dataflow.InputSplitsFactory;
import edu.uci.ics.hyracks.hdfs.scheduler.Scheduler;
@@ -36,7 +38,7 @@
* A factory class for creating an instance of HiveAdapter
*/
@SuppressWarnings("deprecation")
-public class HiveAdapterFactory implements IGenericDatasetAdapterFactory {
+public class HiveAdapterFactory extends FileSystemAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String HDFS_ADAPTER_NAME = "hdfs";
@@ -66,7 +68,8 @@
private InputSplitsFactory inputSplitsFactory;
private ConfFactory confFactory;
private transient AlgebricksPartitionConstraint clusterLocations;
- private boolean setup = false;
+ private boolean configured = false;
+ private IAType atype;
private static final Map<String, String> formatClassNames = initInputFormatMap();
@@ -78,8 +81,33 @@
}
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration, IAType atype) throws Exception {
- if (!setup) {
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ JobConf conf = confFactory.getConf();
+ InputSplit[] inputSplits = inputSplitsFactory.getSplits();
+ String nodeName = ctx.getJobletContext().getApplicationContext().getNodeId();
+ HiveAdapter hiveAdapter = new HiveAdapter(atype, readSchedule, executed, inputSplits, conf, clusterLocations,
+ nodeName, parserFactory, ctx);
+ return hiveAdapter;
+ }
+
+ @Override
+ public String getName() {
+ return "hive";
+ }
+
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.GENERIC;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ if (!configured) {
/** set up the factory --serializable stuff --- this if-block should be called only once for each factory instance */
configureJobConf(configuration);
JobConf conf = configureJobConf(configuration);
@@ -96,18 +124,11 @@
executed = new boolean[readSchedule.length];
Arrays.fill(executed, false);
- setup = true;
+ atype = (IAType) configuration.get(KEY_SOURCE_DATATYPE);
+ configureFormat(atype);
+ configured = true;
}
- JobConf conf = confFactory.getConf();
- InputSplit[] inputSplits = inputSplitsFactory.getSplits();
- HiveAdapter hiveAdapter = new HiveAdapter(atype, readSchedule, executed, inputSplits, conf, clusterLocations);
- hiveAdapter.configure(configuration);
- return hiveAdapter;
- }
- @Override
- public String getName() {
- return "hive";
}
private JobConf configureJobConf(Map<String, Object> configuration) throws Exception {
@@ -122,7 +143,7 @@
tablePath = configuration.get(HIVE_WAREHOUSE_DIR) + "/" + tablePath + ".db" + "/"
+ configuration.get(HIVE_TABLE);
}
- configuration.put(HDFSAdapter.KEY_PATH, tablePath);
+ configuration.put(HDFSAdapterFactory.KEY_PATH, tablePath);
if (!configuration.get(KEY_FORMAT).equals(FORMAT_DELIMITED_TEXT)) {
throw new IllegalArgumentException("format" + configuration.get(KEY_FORMAT) + " is not supported");
}
@@ -142,4 +163,11 @@
(String) formatClassNames.get(((String) configuration.get(KEY_INPUT_FORMAT)).trim()));
return conf;
}
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/NCFileSystemAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/NCFileSystemAdapterFactory.java
index 02f04e8..3d8bedf 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/NCFileSystemAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/NCFileSystemAdapterFactory.java
@@ -14,26 +14,35 @@
*/
package edu.uci.ics.asterix.external.adapter.factory;
+import java.io.File;
import java.util.Map;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.external.dataset.adapter.NCFileSystemAdapter;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericDatasetAdapterFactory;
import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
/**
* Factory class for creating an instance of NCFileSystemAdapter. An
* NCFileSystemAdapter reads external data residing on the local file system of
* an NC.
*/
-public class NCFileSystemAdapterFactory implements IGenericDatasetAdapterFactory {
+public class NCFileSystemAdapterFactory extends FileSystemAdapterFactory {
private static final long serialVersionUID = 1L;
+
public static final String NC_FILE_SYSTEM_ADAPTER_NAME = "localfs";
+ private IAType sourceDatatype;
+ private FileSplit[] fileSplits;
+
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration, IAType atype) throws Exception {
- NCFileSystemAdapter fsAdapter = new NCFileSystemAdapter(atype);
- fsAdapter.configure(configuration);
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ NCFileSystemAdapter fsAdapter = new NCFileSystemAdapter(fileSplits, parserFactory, sourceDatatype, ctx);
return fsAdapter;
}
@@ -41,4 +50,60 @@
public String getName() {
return NC_FILE_SYSTEM_ADAPTER_NAME;
}
+
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.GENERIC;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ String[] splits = ((String) configuration.get(KEY_PATH)).split(",");
+ IAType sourceDatatype = (IAType) configuration.get(KEY_SOURCE_DATATYPE);
+ configureFileSplits(splits);
+ configureFormat(sourceDatatype);
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return configurePartitionConstraint();
+ }
+
+ private void configureFileSplits(String[] splits) throws AsterixException {
+ if (fileSplits == null) {
+ fileSplits = new FileSplit[splits.length];
+ String nodeName;
+ String nodeLocalPath;
+ int count = 0;
+ String trimmedValue;
+ for (String splitPath : splits) {
+ trimmedValue = splitPath.trim();
+ if (!trimmedValue.contains("://")) {
+ throw new AsterixException("Invalid path: " + splitPath
+ + "\nUsage- path=\"Host://Absolute File Path\"");
+ }
+ nodeName = trimmedValue.split(":")[0];
+ nodeLocalPath = trimmedValue.split("://")[1];
+ FileSplit fileSplit = new FileSplit(nodeName, new FileReference(new File(nodeLocalPath)));
+ fileSplits[count++] = fileSplit;
+ }
+ }
+ }
+
+ private AlgebricksPartitionConstraint configurePartitionConstraint() throws AsterixException {
+ String[] locs = new String[fileSplits.length];
+ String location;
+ for (int i = 0; i < fileSplits.length; i++) {
+ location = getNodeResolver().resolveNode(fileSplits[i].getNodeName());
+ locs[i] = location;
+ }
+ return new AlgebricksAbsolutePartitionConstraint(locs);
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedTwitterAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedTwitterAdapterFactory.java
index bd181e5..5937013 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedTwitterAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedTwitterAdapterFactory.java
@@ -17,23 +17,26 @@
import java.util.Map;
import edu.uci.ics.asterix.external.dataset.adapter.PullBasedTwitterAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasetAdapterFactory;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* Factory class for creating an instance of PullBasedTwitterAdapter.
* This adapter provides the functionality of fetching tweets from Twitter service
* via pull-based Twitter API.
*/
-public class PullBasedTwitterAdapterFactory implements ITypedDatasetAdapterFactory {
+public class PullBasedTwitterAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String PULL_BASED_TWITTER_ADAPTER_NAME = "pull_twitter";
+ private Map<String, Object> configuration;
+
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration) throws Exception {
- PullBasedTwitterAdapter twitterAdapter = new PullBasedTwitterAdapter();
- twitterAdapter.configure(configuration);
- return twitterAdapter;
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ return new PullBasedTwitterAdapter(configuration, ctx);
}
@Override
@@ -41,4 +44,24 @@
return PULL_BASED_TWITTER_ADAPTER_NAME;
}
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.TYPED;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return new AlgebricksCountPartitionConstraint(1);
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/RSSFeedAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/RSSFeedAdapterFactory.java
index 6e732ba..d904af9 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/RSSFeedAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/RSSFeedAdapterFactory.java
@@ -14,24 +14,38 @@
*/
package edu.uci.ics.asterix.external.adapter.factory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import edu.uci.ics.asterix.external.dataset.adapter.RSSFeedAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasetAdapterFactory;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.asterix.om.types.BuiltinType;
+import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* Factory class for creating an instance of @see {RSSFeedAdapter}.
* RSSFeedAdapter provides the functionality of fetching an RSS based feed.
*/
-public class RSSFeedAdapterFactory implements ITypedDatasetAdapterFactory {
+public class RSSFeedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String RSS_FEED_ADAPTER_NAME = "rss_feed";
+ public static final String KEY_RSS_URL = "url";
+ public static final String KEY_INTERVAL = "interval";
+
+ private Map<String, Object> configuration;
+ private ARecordType recordType;
+ private List<String> feedURLs = new ArrayList<String>();
+
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration) throws Exception {
- RSSFeedAdapter rssFeedAdapter = new RSSFeedAdapter();
- rssFeedAdapter.configure(configuration);
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ RSSFeedAdapter rssFeedAdapter = new RSSFeedAdapter(configuration, ctx);
return rssFeedAdapter;
}
@@ -40,4 +54,46 @@
return "rss_feed";
}
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.TYPED;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ String rssURLProperty = (String) configuration.get(KEY_RSS_URL);
+ if (rssURLProperty == null) {
+ throw new IllegalArgumentException("no rss url provided");
+ }
+ initializeFeedURLs(rssURLProperty);
+ configurePartitionConstraints();
+ recordType = new ARecordType("FeedRecordType", new String[] { "id", "title", "description", "link" },
+ new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING },
+ false);
+
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return new AlgebricksCountPartitionConstraint(feedURLs.size());
+ }
+
+ private void initializeFeedURLs(String rssURLProperty) {
+ feedURLs.clear();
+ String[] feedURLProperty = rssURLProperty.split(",");
+ for (String feedURL : feedURLProperty) {
+ feedURLs.add(feedURL);
+ }
+ }
+
+ protected void configurePartitionConstraints() {
+
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/CNNFeedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/CNNFeedAdapter.java
index cfcfd74..5debc53 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/CNNFeedAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/CNNFeedAdapter.java
@@ -14,95 +14,23 @@
*/
package edu.uci.ics.asterix.external.dataset.adapter;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IManagedFeedAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* An Adapter that provides the functionality of fetching news feed from CNN service
* The Adapter provides news feed as ADM records.
*/
-public class CNNFeedAdapter extends RSSFeedAdapter implements IDatasourceAdapter, IManagedFeedAdapter {
+public class CNNFeedAdapter extends RSSFeedAdapter implements IDatasourceAdapter, IFeedAdapter {
private static final long serialVersionUID = 2523303758114582251L;
- private List<String> feedURLs = new ArrayList<String>();
- private static Map<String, String> topicFeeds = new HashMap<String, String>();
- public static final String KEY_RSS_URL = "topic";
- public static final String KEY_INTERVAL = "interval";
- public static final String TOP_STORIES = "topstories";
- public static final String WORLD = "world";
- public static final String US = "us";
- public static final String SPORTS = "sports";
- public static final String BUSINESS = "business";
- public static final String POLITICS = "politics";
- public static final String CRIME = "crime";
- public static final String TECHNOLOGY = "technology";
- public static final String HEALTH = "health";
- public static final String ENTERNTAINMENT = "entertainemnt";
- public static final String TRAVEL = "travel";
- public static final String LIVING = "living";
- public static final String VIDEO = "video";
- public static final String STUDENT = "student";
- public static final String POPULAR = "popular";
- public static final String RECENT = "recent";
-
- private void initTopics() {
- topicFeeds.put(TOP_STORIES, "http://rss.cnn.com/rss/cnn_topstories.rss");
- topicFeeds.put(WORLD, "http://rss.cnn.com/rss/cnn_world.rss");
- topicFeeds.put(US, "http://rss.cnn.com/rss/cnn_us.rss");
- topicFeeds.put(SPORTS, "http://rss.cnn.com/rss/si_topstories.rss");
- topicFeeds.put(BUSINESS, "http://rss.cnn.com/rss/money_latest.rss");
- topicFeeds.put(POLITICS, "http://rss.cnn.com/rss/cnn_allpolitics.rss");
- topicFeeds.put(CRIME, "http://rss.cnn.com/rss/cnn_crime.rss");
- topicFeeds.put(TECHNOLOGY, "http://rss.cnn.com/rss/cnn_tech.rss");
- topicFeeds.put(HEALTH, "http://rss.cnn.com/rss/cnn_health.rss");
- topicFeeds.put(ENTERNTAINMENT, "http://rss.cnn.com/rss/cnn_showbiz.rss");
- topicFeeds.put(LIVING, "http://rss.cnn.com/rss/cnn_living.rss");
- topicFeeds.put(VIDEO, "http://rss.cnn.com/rss/cnn_freevideo.rss");
- topicFeeds.put(TRAVEL, "http://rss.cnn.com/rss/cnn_travel.rss");
- topicFeeds.put(STUDENT, "http://rss.cnn.com/rss/cnn_studentnews.rss");
- topicFeeds.put(POPULAR, "http://rss.cnn.com/rss/cnn_mostpopular.rss");
- topicFeeds.put(RECENT, "http://rss.cnn.com/rss/cnn_latest.rss");
- }
-
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- configuration = arguments;
- String rssURLProperty = (String) configuration.get(KEY_RSS_URL);
- if (rssURLProperty == null) {
- throw new IllegalArgumentException("no rss url provided");
- }
- initializeFeedURLs(rssURLProperty);
- configurePartitionConstraints();
-
- }
-
- private void initializeFeedURLs(String rssURLProperty) {
- feedURLs.clear();
- String[] rssTopics = rssURLProperty.split(",");
- initTopics();
- for (String topic : rssTopics) {
- String feedURL = topicFeeds.get(topic);
- if (feedURL == null) {
- throw new IllegalArgumentException(" unknown topic :" + topic + " please choose from the following "
- + getValidTopics());
- }
- feedURLs.add(feedURL);
- }
- }
-
- private static String getValidTopics() {
- StringBuilder builder = new StringBuilder();
- for (String key : topicFeeds.keySet()) {
- builder.append(key);
- builder.append(" ");
- }
- return new String(builder);
+ public CNNFeedAdapter(Map<String, Object> configuration, IHyracksTaskContext ctx) throws AsterixException {
+ super(configuration, ctx);
}
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
index 6ce18fa..1897272 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FileSystemBasedAdapter.java
@@ -16,145 +16,36 @@
import java.io.IOException;
import java.io.InputStream;
-import java.util.Map;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
-import edu.uci.ics.asterix.external.util.DNSResolverFactory;
-import edu.uci.ics.asterix.external.util.INodeResolver;
-import edu.uci.ics.asterix.external.util.INodeResolverFactory;
-import edu.uci.ics.asterix.metadata.feeds.AbstractDatasourceAdapter;
-import edu.uci.ics.asterix.om.types.ARecordType;
-import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.asterix.runtime.operators.file.AdmSchemafullRecordParserFactory;
-import edu.uci.ics.asterix.runtime.operators.file.NtDelimitedDataTupleParserFactory;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
-import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
-public abstract class FileSystemBasedAdapter extends AbstractDatasourceAdapter {
+public abstract class FileSystemBasedAdapter implements IDatasourceAdapter {
private static final long serialVersionUID = 1L;
public static final String NODE_RESOLVER_FACTORY_PROPERTY = "node.Resolver";
- public static final String KEY_DELIMITER = "delimiter";
- public static final String KEY_PATH = "path";
-
- protected ITupleParserFactory parserFactory;
- protected ITupleParser parser;
- protected static INodeResolver nodeResolver;
-
- private static final INodeResolver DEFAULT_NODE_RESOLVER = new DNSResolverFactory().createNodeResolver();
- private static final Logger LOGGER = Logger.getLogger(FileSystemBasedAdapter.class.getName());
public abstract InputStream getInputStream(int partition) throws IOException;
- public FileSystemBasedAdapter(IAType atype) {
- this.atype = atype;
+ protected final ITupleParser tupleParser;
+ protected final IAType sourceDatatype;
+ protected IHyracksTaskContext ctx;
+
+ public FileSystemBasedAdapter(ITupleParserFactory parserFactory, IAType sourceDatatype, IHyracksTaskContext ctx) {
+ this.tupleParser = parserFactory.createTupleParser(ctx);
+ this.sourceDatatype = sourceDatatype;
+ this.ctx = ctx;
}
@Override
public void start(int partition, IFrameWriter writer) throws Exception {
InputStream in = getInputStream(partition);
- parser = getTupleParser();
- parser.parse(in, writer);
+ tupleParser.parse(in, writer);
}
- @Override
- public abstract void initialize(IHyracksTaskContext ctx) throws Exception;
-
- @Override
- public abstract void configure(Map<String, Object> arguments) throws Exception;
-
- @Override
- public abstract AdapterType getAdapterType();
-
- @Override
- public abstract AlgebricksPartitionConstraint getPartitionConstraint() throws Exception;
-
- protected ITupleParser getTupleParser() throws Exception {
- return parserFactory.createTupleParser(ctx);
- }
-
- protected void configureFormat() throws Exception {
- String parserFactoryClassname = (String) configuration.get(KEY_PARSER_FACTORY);
- if (parserFactoryClassname == null) {
- String specifiedFormat = (String) configuration.get(KEY_FORMAT);
- if (specifiedFormat == null) {
- throw new IllegalArgumentException(" Unspecified data format");
- } else if (FORMAT_DELIMITED_TEXT.equalsIgnoreCase(specifiedFormat)) {
- parserFactory = getDelimitedDataTupleParserFactory((ARecordType) atype);
- } else if (FORMAT_ADM.equalsIgnoreCase((String)configuration.get(KEY_FORMAT))) {
- parserFactory = getADMDataTupleParserFactory((ARecordType) atype);
- } else {
- throw new IllegalArgumentException(" format " + configuration.get(KEY_FORMAT) + " not supported");
- }
- } else {
- parserFactory = (ITupleParserFactory) Class.forName(parserFactoryClassname).newInstance();
- }
-
- }
-
- protected ITupleParserFactory getDelimitedDataTupleParserFactory(ARecordType recordType) throws AsterixException {
- int n = recordType.getFieldTypes().length;
- IValueParserFactory[] fieldParserFactories = new IValueParserFactory[n];
- for (int i = 0; i < n; i++) {
- ATypeTag tag = recordType.getFieldTypes()[i].getTypeTag();
- IValueParserFactory vpf = typeToValueParserFactMap.get(tag);
- if (vpf == null) {
- throw new NotImplementedException("No value parser factory for delimited fields of type " + tag);
- }
- fieldParserFactories[i] = vpf;
- }
- String delimiterValue = (String) configuration.get(KEY_DELIMITER);
- if (delimiterValue != null && delimiterValue.length() > 1) {
- throw new AsterixException("improper delimiter");
- }
-
- Character delimiter = delimiterValue.charAt(0);
- return new NtDelimitedDataTupleParserFactory(recordType, fieldParserFactories, delimiter);
- }
-
- protected ITupleParserFactory getADMDataTupleParserFactory(ARecordType recordType) throws AsterixException {
- try {
- return new AdmSchemafullRecordParserFactory(recordType);
- } catch (Exception e) {
- throw new AsterixException(e);
- }
-
- }
-
- protected INodeResolver getNodeResolver() {
- if (nodeResolver == null) {
- nodeResolver = initNodeResolver();
- }
- return nodeResolver;
- }
-
- private static INodeResolver initNodeResolver() {
- INodeResolver nodeResolver = null;
- String configuredNodeResolverFactory = System.getProperty(NODE_RESOLVER_FACTORY_PROPERTY);
- if (configuredNodeResolverFactory != null) {
- try {
- nodeResolver = ((INodeResolverFactory) (Class.forName(configuredNodeResolverFactory).newInstance()))
- .createNodeResolver();
-
- } catch (Exception e) {
- if (LOGGER.isLoggable(Level.WARNING)) {
- LOGGER.log(Level.WARNING, "Unable to create node resolver from the configured classname "
- + configuredNodeResolverFactory + "\n" + e.getMessage());
- }
- nodeResolver = DEFAULT_NODE_RESOLVER;
- }
- } else {
- nodeResolver = DEFAULT_NODE_RESOLVER;
- }
- return nodeResolver;
- }
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
index f8b381b..11f44ea 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HDFSAdapter.java
@@ -16,7 +16,6 @@
import java.io.IOException;
import java.io.InputStream;
-import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Counters.Counter;
@@ -31,6 +30,7 @@
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
/**
* Provides functionality for fetching external data stored in an HDFS instance.
@@ -44,34 +44,16 @@
private transient boolean executed[];
private transient InputSplit[] inputSplits;
private transient JobConf conf;
- private transient AlgebricksPartitionConstraint clusterLocations;
-
private transient String nodeName;
public HDFSAdapter(IAType atype, String[] readSchedule, boolean[] executed, InputSplit[] inputSplits, JobConf conf,
- AlgebricksPartitionConstraint clusterLocations) {
- super(atype);
+ String nodeName, ITupleParserFactory parserFactory, IHyracksTaskContext ctx) {
+ super(parserFactory, atype, ctx);
this.readSchedule = readSchedule;
this.executed = executed;
this.inputSplits = inputSplits;
this.conf = conf;
- this.clusterLocations = clusterLocations;
- }
-
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- this.configuration = arguments;
- configureFormat();
- }
-
- public AdapterType getAdapterType() {
- return AdapterType.READ_WRITE;
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- this.ctx = ctx;
- this.nodeName = ctx.getJobletContext().getApplicationContext().getNodeId();
+ this.nodeName = nodeName;
}
private Reporter getReporter() {
@@ -227,9 +209,4 @@
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- return clusterLocations;
- }
-
}
\ No newline at end of file
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HiveAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HiveAdapter.java
index 39484a6..8e5adfc 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HiveAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/HiveAdapter.java
@@ -14,21 +14,22 @@
*/
package edu.uci.ics.asterix.external.dataset.adapter;
-import java.util.Map;
+import java.io.IOException;
+import java.io.InputStream;
-import edu.uci.ics.asterix.metadata.feeds.AbstractDatasourceAdapter;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
+
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
/**
* Provides the functionality of fetching data in form of ADM records from a Hive dataset.
*/
@SuppressWarnings("deprecation")
-public class HiveAdapter extends AbstractDatasourceAdapter {
+public class HiveAdapter extends FileSystemBasedAdapter {
private static final long serialVersionUID = 1L;
@@ -42,35 +43,16 @@
private HDFSAdapter hdfsAdapter;
public HiveAdapter(IAType atype, String[] readSchedule, boolean[] executed, InputSplit[] inputSplits, JobConf conf,
- AlgebricksPartitionConstraint clusterLocations) {
- this.hdfsAdapter = new HDFSAdapter(atype, readSchedule, executed, inputSplits, conf, clusterLocations);
- this.atype = atype;
+ AlgebricksPartitionConstraint clusterLocations, String nodeName, ITupleParserFactory parserFactory,
+ IHyracksTaskContext ctx) {
+ super(parserFactory, atype, ctx);
+ this.hdfsAdapter = new HDFSAdapter(atype, readSchedule, executed, inputSplits, conf, nodeName, parserFactory,
+ ctx);
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
-
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- this.configuration = arguments;
- this.hdfsAdapter.configure(arguments);
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- hdfsAdapter.initialize(ctx);
- }
-
- @Override
- public void start(int partition, IFrameWriter writer) throws Exception {
- hdfsAdapter.start(partition, writer);
- }
-
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- return hdfsAdapter.getPartitionConstraint();
+ public InputStream getInputStream(int partition) throws IOException {
+ return hdfsAdapter.getInputStream(partition);
}
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
index 182ccf7..f3075d0 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/NCFileSystemAdapter.java
@@ -19,15 +19,11 @@
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
-import java.util.Map;
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
/**
* Factory class for creating an instance of NCFileSystemAdapter. An
@@ -37,59 +33,13 @@
public class NCFileSystemAdapter extends FileSystemBasedAdapter {
private static final long serialVersionUID = 1L;
- private FileSplit[] fileSplits;
- public NCFileSystemAdapter(IAType atype) {
- super(atype);
- }
+ private final FileSplit[] fileSplits;
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- this.configuration = arguments;
- String[] splits = ((String) arguments.get(KEY_PATH)).split(",");
- configureFileSplits(splits);
- configureFormat();
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- this.ctx = ctx;
- }
-
- @Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
-
- private void configureFileSplits(String[] splits) throws AsterixException {
- if (fileSplits == null) {
- fileSplits = new FileSplit[splits.length];
- String nodeName;
- String nodeLocalPath;
- int count = 0;
- String trimmedValue;
- for (String splitPath : splits) {
- trimmedValue = splitPath.trim();
- if (!trimmedValue.contains("://")) {
- throw new AsterixException("Invalid path: " + splitPath
- + "\nUsage- path=\"Host://Absolute File Path\"");
- }
- nodeName = trimmedValue.split(":")[0];
- nodeLocalPath = trimmedValue.split("://")[1];
- FileSplit fileSplit = new FileSplit(nodeName, new FileReference(new File(nodeLocalPath)));
- fileSplits[count++] = fileSplit;
- }
- }
- }
-
- private void configurePartitionConstraint() throws AsterixException {
- String[] locs = new String[fileSplits.length];
- String location;
- for (int i = 0; i < fileSplits.length; i++) {
- location = getNodeResolver().resolveNode(fileSplits[i].getNodeName());
- locs[i] = location;
- }
- partitionConstraint = new AlgebricksAbsolutePartitionConstraint(locs);
+ public NCFileSystemAdapter(FileSplit[] fileSplits, ITupleParserFactory parserFactory, IAType atype,
+ IHyracksTaskContext ctx) {
+ super(parserFactory, atype, ctx);
+ this.fileSplits = fileSplits;
}
@Override
@@ -105,11 +55,4 @@
}
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- if (partitionConstraint == null) {
- configurePartitionConstraint();
- }
- return partitionConstraint;
- }
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedAdapter.java
index 0223700..534b495 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedAdapter.java
@@ -20,11 +20,11 @@
import edu.uci.ics.asterix.external.dataset.adapter.IPullBasedFeedClient.InflowState;
import edu.uci.ics.asterix.metadata.feeds.AbstractFeedDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.FeedPolicyEnforcer;
-import edu.uci.ics.asterix.metadata.feeds.IManagedFeedAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasourceAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
@@ -35,8 +35,8 @@
* Captures the common logic for obtaining bytes from an external source
* and packing them into frames as tuples.
*/
-public abstract class PullBasedAdapter extends AbstractFeedDatasourceAdapter implements ITypedDatasourceAdapter,
- IManagedFeedAdapter {
+public abstract class PullBasedAdapter extends AbstractFeedDatasourceAdapter implements IDatasourceAdapter,
+ IFeedAdapter {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = Logger.getLogger(PullBasedAdapter.class.getName());
@@ -50,9 +50,16 @@
protected boolean alterRequested = false;
private Map<String, Object> modifiedConfiguration = null;
private long tupleCount = 0;
+ private final IHyracksTaskContext ctx;
+ protected Map<String, Object> configuration;
public abstract IPullBasedFeedClient getFeedClient(int partition) throws Exception;
+ public PullBasedAdapter(Map<String, Object> configuration, IHyracksTaskContext ctx) {
+ this.ctx = ctx;
+ this.configuration = configuration;
+ }
+
public long getIngestedRecordsCount() {
return tupleCount;
}
@@ -122,11 +129,6 @@
}
}
- @Override
- public ARecordType getAdapterOutputType() {
- return adapterOutputType;
- }
-
/**
* Discontinue the ingestion of data and end the feed.
*
@@ -138,4 +140,8 @@
timer.cancel();
}
+ public Map<String, Object> getConfiguration() {
+ return configuration;
+ }
+
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedTwitterAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedTwitterAdapter.java
index 97b1d77..a9820b7 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedTwitterAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/PullBasedTwitterAdapter.java
@@ -16,19 +16,18 @@
import java.util.Map;
-import edu.uci.ics.asterix.metadata.feeds.IManagedFeedAdapter;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.BuiltinType;
import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* An adapter that provides the functionality of receiving tweets from the
* Twitter service in the form of ADM formatted records.
*/
-public class PullBasedTwitterAdapter extends PullBasedAdapter implements IManagedFeedAdapter {
+public class PullBasedTwitterAdapter extends PullBasedAdapter implements IFeedAdapter {
private static final long serialVersionUID = 1L;
@@ -36,7 +35,7 @@
public static final String INTERVAL = "interval";
private ARecordType recordType;
-
+ private final IHyracksTaskContext ctx;
private PullBasedTwitterFeedClient tweetClient;
@Override
@@ -44,27 +43,17 @@
return tweetClient;
}
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- configuration = arguments;
+ public PullBasedTwitterAdapter(Map<String, Object> configuration, IHyracksTaskContext ctx) throws AsterixException {
+ super(configuration, ctx);
String[] fieldNames = { "id", "username", "location", "text", "timestamp" };
IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
BuiltinType.ASTRING };
recordType = new ARecordType("FeedRecordType", fieldNames, fieldTypes, false);
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
this.ctx = ctx;
tweetClient = new PullBasedTwitterFeedClient(ctx, this);
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
-
- @Override
public void stop() {
tweetClient.stop();
}
@@ -86,12 +75,4 @@
return recordType;
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- if (partitionConstraint == null) {
- partitionConstraint = new AlgebricksCountPartitionConstraint(1);
- }
- return partitionConstraint;
- }
-
}
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/RSSFeedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/RSSFeedAdapter.java
index c3c5ae0..bd503f2 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/RSSFeedAdapter.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/RSSFeedAdapter.java
@@ -19,18 +19,17 @@
import java.util.List;
import java.util.Map;
-import edu.uci.ics.asterix.metadata.feeds.IManagedFeedAdapter;
+import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.BuiltinType;
import edu.uci.ics.asterix.om.types.IAType;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* RSSFeedAdapter provides the functionality of fetching an RSS based feed.
*/
-public class RSSFeedAdapter extends PullBasedAdapter implements IManagedFeedAdapter {
+public class RSSFeedAdapter extends PullBasedAdapter implements IFeedAdapter {
private static final long serialVersionUID = 1L;
@@ -43,13 +42,18 @@
private IPullBasedFeedClient rssFeedClient;
- public static final String KEY_RSS_URL = "url";
- public static final String KEY_INTERVAL = "interval";
-
public boolean isStopRequested() {
return isStopRequested;
}
+ public RSSFeedAdapter(Map<String, Object> configuration, IHyracksTaskContext ctx) throws AsterixException {
+ super(configuration, ctx);
+ recordType = new ARecordType("FeedRecordType", new String[] { "id", "title", "description", "link" },
+ new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING },
+ false);
+ id_prefix = ctx.getJobletContext().getApplicationContext().getNodeId();
+ }
+
public void setStopRequested(boolean isStopRequested) {
this.isStopRequested = isStopRequested;
}
@@ -66,25 +70,6 @@
isStopRequested = true;
}
- @Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
-
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- configuration = arguments;
- String rssURLProperty = (String) configuration.get(KEY_RSS_URL);
- if (rssURLProperty == null) {
- throw new IllegalArgumentException("no rss url provided");
- }
- initializeFeedURLs(rssURLProperty);
- configurePartitionConstraints();
- recordType = new ARecordType("FeedRecordType", new String[] { "id", "title", "description", "link" },
- new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING },
- false);
- }
-
private void initializeFeedURLs(String rssURLProperty) {
feedURLs.clear();
String[] feedURLProperty = rssURLProperty.split(",");
@@ -94,22 +79,12 @@
}
protected void reconfigure(Map<String, Object> arguments) {
- String rssURLProperty = (String) configuration.get(KEY_RSS_URL);
+ String rssURLProperty = (String) configuration.get("KEY_RSS_URL");
if (rssURLProperty != null) {
initializeFeedURLs(rssURLProperty);
}
}
- protected void configurePartitionConstraints() {
- partitionConstraint = new AlgebricksCountPartitionConstraint(feedURLs.size());
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- this.ctx = ctx;
- id_prefix = ctx.getJobletContext().getApplicationContext().getNodeId();
- }
-
public boolean isAlterRequested() {
return isAlterRequested;
}
@@ -126,17 +101,8 @@
return rssFeedClient;
}
- @Override
public ARecordType getAdapterOutputType() {
return recordType;
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- if (partitionConstraint == null) {
- configurePartitionConstraints();
- }
- return partitionConstraint;
- }
-
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
index f922169..f6106e6 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
@@ -162,18 +162,14 @@
public static final int FEED_DETAILS_ARECORD_DATASOURCE_ADAPTER_FIELD_INDEX = 5;
public static final int FEED_DETAILS_ARECORD_PROPERTIES_FIELD_INDEX = 6;
public static final int FEED_DETAILS_ARECORD_FUNCTION_FIELD_INDEX = 7;
- public static final int FEED_DETAILS_ARECORD_STATE_FIELD_INDEX = 8;
- public static final int FEED_DETAILS_ARECORD_INGEST_NODES_FIELD_INDEX = 9;
- public static final int FEED_DETAILS_ARECORD_COMPUTE_NODES_FIELD_INDEX = 10;
private static final ARecordType createFeedDetailsRecordType() throws AsterixException {
AOrderedListType orderedListType = new AOrderedListType(BuiltinType.ASTRING, null);
AOrderedListType orderedListOfPropertiesType = new AOrderedListType(DATASOURCE_ADAPTER_PROPERTIES_RECORDTYPE,
null);
- AUnorderedListType unorderedListType = new AUnorderedListType(BuiltinType.ASTRING, null);
String[] fieldNames = { "FileStructure", "PartitioningStrategy", "PartitioningKey", "PrimaryKey", "GroupName",
- "DatasourceAdapter", "Properties", "Function", "Status", "IngestNodes", "ComputeNodes" };
+ "DatasourceAdapter", "Properties", "Function" };
List<IAType> feedFunctionUnionList = new ArrayList<IAType>();
feedFunctionUnionList.add(BuiltinType.ANULL);
@@ -181,8 +177,7 @@
AUnionType feedFunctionUnion = new AUnionType(feedFunctionUnionList, null);
IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, orderedListType, orderedListType,
- BuiltinType.ASTRING, BuiltinType.ASTRING, orderedListOfPropertiesType, feedFunctionUnion,
- BuiltinType.ASTRING, unorderedListType, unorderedListType };
+ BuiltinType.ASTRING, BuiltinType.ASTRING, orderedListOfPropertiesType, feedFunctionUnion };
return new ARecordType(null, fieldNames, fieldTypes, true);
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/cluster/ClusterManager.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/cluster/ClusterManager.java
index dc072d4..b639559 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/cluster/ClusterManager.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/cluster/ClusterManager.java
@@ -46,32 +46,36 @@
private ClusterManager() {
Cluster asterixCluster = AsterixClusterProperties.INSTANCE.getCluster();
- String asterixDir = System.getProperty("user.dir") + File.separator + "asterix";
- String eventHome = asterixCluster.getWorkingDir().getDir();
- File configFile = new File(System.getProperty("user.dir") + File.separator + "configuration.xml");
- Configuration configuration = null;
+ String eventHome = asterixCluster == null ? null : asterixCluster.getWorkingDir().getDir();
- try {
- JAXBContext configCtx = JAXBContext.newInstance(Configuration.class);
- Unmarshaller unmarshaller = configCtx.createUnmarshaller();
- configuration = (Configuration) unmarshaller.unmarshal(configFile);
- AsterixEventService.initialize(configuration, asterixDir, eventHome);
- client = AsterixEventService.getAsterixEventServiceClient(AsterixClusterProperties.INSTANCE.getCluster());
+ if (asterixCluster != null) {
+ String asterixDir = System.getProperty("user.dir") + File.separator + "asterix";
+ File configFile = new File(System.getProperty("user.dir") + File.separator + "configuration.xml");
+ Configuration configuration = null;
- lookupService = ServiceProvider.INSTANCE.getLookupService();
- if (!lookupService.isRunning(configuration)) {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Lookup service not running. Starting lookup service ...");
+ try {
+ JAXBContext configCtx = JAXBContext.newInstance(Configuration.class);
+ Unmarshaller unmarshaller = configCtx.createUnmarshaller();
+ configuration = (Configuration) unmarshaller.unmarshal(configFile);
+ AsterixEventService.initialize(configuration, asterixDir, eventHome);
+ client = AsterixEventService.getAsterixEventServiceClient(AsterixClusterProperties.INSTANCE
+ .getCluster());
+
+ lookupService = ServiceProvider.INSTANCE.getLookupService();
+ if (!lookupService.isRunning(configuration)) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Lookup service not running. Starting lookup service ...");
+ }
+ lookupService.startService(configuration);
+ } else {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Lookup service running");
+ }
}
- lookupService.startService(configuration);
- } else {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Lookup service running");
- }
+
+ } catch (Exception e) {
+ throw new IllegalStateException("Unable to initialize cluster manager" + e);
}
-
- } catch (Exception e) {
- throw new IllegalStateException("Unable to initialize cluster manager" + e);
}
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
index b2ccbc0..8c03198 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
@@ -62,11 +62,9 @@
import edu.uci.ics.asterix.metadata.feeds.FeedIntakeOperatorDescriptor;
import edu.uci.ics.asterix.metadata.feeds.FeedMessageOperatorDescriptor;
import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory.SupportedOperation;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
import edu.uci.ics.asterix.metadata.feeds.IFeedMessage;
-import edu.uci.ics.asterix.metadata.feeds.IGenericDatasetAdapterFactory;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasetAdapterFactory;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasourceAdapter;
import edu.uci.ics.asterix.metadata.utils.DatasetUtils;
import edu.uci.ics.asterix.om.functions.AsterixBuiltinFunctions;
import edu.uci.ics.asterix.om.types.ARecordType;
@@ -119,6 +117,7 @@
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.dataset.ResultSetId;
+import edu.uci.ics.hyracks.api.exceptions.HyracksException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.data.std.accessors.PointableBinaryComparatorFactory;
@@ -168,8 +167,6 @@
private final AsterixStorageProperties storageProperties;
private static final Map<String, String> adapterFactoryMapping = initializeAdapterFactoryMapping();
- private static Scheduler hdfsScheduler;
- public static final String CLUSTER_LOCATIONS = "cluster-locations";
public static transient String SCHEDULER = "hdfs-scheduler";
public String getPropertyValue(String propertyName) {
@@ -192,16 +189,6 @@
this.defaultDataverse = defaultDataverse;
this.stores = AsterixAppContextInfo.getInstance().getMetadataProperties().getStores();
this.storageProperties = AsterixAppContextInfo.getInstance().getStorageProperties();
- ICCContext ccContext = AsterixAppContextInfo.getInstance().getCCApplicationContext().getCCContext();
- try {
- if (hdfsScheduler == null) {
- //set the singleton hdfs scheduler
- hdfsScheduler = new Scheduler(ccContext.getClusterControllerInfo().getClientNetAddress(), ccContext
- .getClusterControllerInfo().getClientNetPort());
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
}
public void setJobId(JobId jobId) {
@@ -355,8 +342,7 @@
throw new AlgebricksException("Can only scan datasets of records.");
}
- IGenericDatasetAdapterFactory adapterFactory;
- IDatasourceAdapter adapter;
+ IAdapterFactory adapterFactory;
String adapterName;
DatasourceAdapter adapterEntity;
String adapterFactoryClassname;
@@ -366,17 +352,18 @@
adapterName);
if (adapterEntity != null) {
adapterFactoryClassname = adapterEntity.getClassname();
- adapterFactory = (IGenericDatasetAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
+ adapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
} else {
adapterFactoryClassname = adapterFactoryMapping.get(adapterName);
if (adapterFactoryClassname == null) {
throw new AlgebricksException(" Unknown adapter :" + adapterName);
}
- adapterFactory = (IGenericDatasetAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
+ adapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
}
- adapter = ((IGenericDatasetAdapterFactory) adapterFactory).createAdapter(
- wrapProperties(datasetDetails.getProperties()), itemType);
+ Map<String, Object> configuration = this.wrapProperties(datasetDetails.getProperties());
+ configuration.put("source-datatype", itemType);
+ adapterFactory.configure(configuration);
} catch (AlgebricksException ae) {
throw ae;
} catch (Exception e) {
@@ -384,21 +371,20 @@
throw new AlgebricksException("Unable to create adapter " + e);
}
- if (!(adapter.getAdapterType().equals(IDatasourceAdapter.AdapterType.READ) || adapter.getAdapterType().equals(
- IDatasourceAdapter.AdapterType.READ_WRITE))) {
+ if (!(adapterFactory.getSupportedOperations().equals(SupportedOperation.READ) || adapterFactory
+ .getSupportedOperations().equals(SupportedOperation.READ_WRITE))) {
throw new AlgebricksException("external dataset adapter does not support read operation");
}
- ARecordType rt = (ARecordType) itemType;
ISerializerDeserializer payloadSerde = format.getSerdeProvider().getSerializerDeserializer(itemType);
RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { payloadSerde });
- ExternalDataScanOperatorDescriptor dataScanner = new ExternalDataScanOperatorDescriptor(jobSpec,
- wrapPropertiesEmpty(datasetDetails.getProperties()), rt, scannerDesc, adapterFactory);
+ ExternalDataScanOperatorDescriptor dataScanner = new ExternalDataScanOperatorDescriptor(jobSpec, scannerDesc,
+ adapterFactory);
AlgebricksPartitionConstraint constraint;
try {
- constraint = adapter.getPartitionConstraint();
+ constraint = adapterFactory.getPartitionConstraint();
} catch (Exception e) {
throw new AlgebricksException(e);
}
@@ -451,26 +437,28 @@
adapterFactoryClassname = adapterFactoryMapping.get(adapterName);
if (adapterFactoryClassname != null) {
} else {
- // adapterName has been provided as a fully qualified
- // classname
adapterFactoryClassname = adapterName;
}
adapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClassname).newInstance();
}
- if (adapterFactory instanceof ITypedDatasetAdapterFactory) {
- adapter = ((ITypedDatasetAdapterFactory) adapterFactory).createAdapter(wrapProperties(datasetDetails
- .getProperties()));
- adapterOutputType = ((ITypedDatasourceAdapter) adapter).getAdapterOutputType();
- } else if (adapterFactory instanceof IGenericDatasetAdapterFactory) {
- String outputTypeName = datasetDetails.getProperties().get(IGenericDatasetAdapterFactory.KEY_TYPE_NAME);
- adapterOutputType = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataset.getDataverseName(),
- outputTypeName).getDatatype();
- adapter = ((IGenericDatasetAdapterFactory) adapterFactory).createAdapter(
- wrapProperties(datasetDetails.getProperties()), adapterOutputType);
- } else {
- throw new IllegalStateException(" Unknown factory type for " + adapterFactoryClassname);
+ Map<String, Object> configuration = this.wrapProperties(datasetDetails.getProperties());
+
+ switch (adapterFactory.getAdapterType()) {
+ case TYPED:
+ adapterOutputType = null;
+ break;
+ case GENERIC:
+ String outputTypeName = datasetDetails.getProperties().get("output-type-name");
+ adapterOutputType = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataset.getDataverseName(),
+ outputTypeName).getDatatype();
+ break;
+ default:
+ throw new IllegalStateException(" Unknown factory type for " + adapterFactoryClassname);
}
+ configuration.put("source-datatype", adapterOutputType);
+ adapterFactory.configure(configuration);
+
} catch (AlgebricksException ae) {
throw ae;
} catch (Exception e) {
@@ -485,17 +473,13 @@
FeedPolicy feedPolicy = (FeedPolicy) ((AqlDataSource) dataSource).getProperties().get(
BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY);
- // public FeedIntakeOperatorDescriptor(JobSpecification spec, FeedId feedId, String adapter,
- // Map<String, Object> arguments, ARecordType atype, RecordDescriptor rDesc, Map<String, String> feedPolicy) {
-
FeedIntakeOperatorDescriptor feedIngestor = new FeedIntakeOperatorDescriptor(jobSpec, new FeedId(
- dataset.getDataverseName(), dataset.getDatasetName()), adapterFactoryClassname,
- this.wrapPropertiesEmpty(datasetDetails.getProperties()), (ARecordType) adapterOutputType, feedDesc,
- feedPolicy.getProperties());
+ dataset.getDataverseName(), dataset.getDatasetName()), adapterFactory, (ARecordType) adapterOutputType,
+ feedDesc, feedPolicy.getProperties());
AlgebricksPartitionConstraint constraint = null;
try {
- constraint = adapter.getPartitionConstraint();
+ constraint = adapterFactory.getPartitionConstraint();
} catch (Exception e) {
throw new AlgebricksException(e);
}
@@ -1537,8 +1521,8 @@
private Map<String, Object> wrapProperties(Map<String, String> properties) {
Map<String, Object> wrappedProperties = new HashMap<String, Object>();
wrappedProperties.putAll(properties);
- wrappedProperties.put(SCHEDULER, hdfsScheduler);
- wrappedProperties.put(CLUSTER_LOCATIONS, getClusterLocations());
+ //wrappedProperties.put(SCHEDULER, hdfsScheduler);
+ //wrappedProperties.put(CLUSTER_LOCATIONS, getClusterLocations());
return wrappedProperties;
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/FeedDatasetDetails.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/FeedDatasetDetails.java
index 81be3b5..e2268bf 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/FeedDatasetDetails.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/FeedDatasetDetails.java
@@ -23,7 +23,6 @@
import edu.uci.ics.asterix.builders.IARecordBuilder;
import edu.uci.ics.asterix.builders.OrderedListBuilder;
import edu.uci.ics.asterix.builders.RecordBuilder;
-import edu.uci.ics.asterix.builders.UnorderedListBuilder;
import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.common.functions.FunctionSignature;
@@ -32,7 +31,6 @@
import edu.uci.ics.asterix.om.base.AMutableString;
import edu.uci.ics.asterix.om.base.AString;
import edu.uci.ics.asterix.om.types.AOrderedListType;
-import edu.uci.ics.asterix.om.types.AUnorderedListType;
import edu.uci.ics.asterix.om.types.BuiltinType;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
@@ -49,9 +47,6 @@
private final String adapterFactory;
private final Map<String, String> properties;
private final FunctionSignature signature;
- private FeedState feedState;
- private List<String> ingestNodes;
- private List<String> computeNodes;
public enum FeedState {
// INACTIVE state signifies that the feed dataset is not
@@ -66,15 +61,11 @@
public FeedDatasetDetails(FileStructure fileStructure, PartitioningStrategy partitioningStrategy,
List<String> partitioningKey, List<String> primaryKey, String groupName, String adapterFactory,
- Map<String, String> properties, FunctionSignature signature, String feedState, List<String> ingestNodes,
- List<String> computeNodes) {
+ Map<String, String> properties, FunctionSignature signature) {
super(fileStructure, partitioningStrategy, partitioningKey, primaryKey, groupName);
this.properties = properties;
this.adapterFactory = adapterFactory;
this.signature = signature;
- this.feedState = feedState.equals(FeedState.ACTIVE.toString()) ? FeedState.ACTIVE : FeedState.INACTIVE;
- this.ingestNodes = ingestNodes;
- this.computeNodes = computeNodes;
}
@Override
@@ -162,44 +153,6 @@
feedRecordBuilder.addField(MetadataRecordTypes.FEED_DETAILS_ARECORD_FUNCTION_FIELD_INDEX, fieldValue);
}
- // write field 8
- fieldValue.reset();
- aString.setValue(getFeedState().toString());
- stringSerde.serialize(aString, fieldValue.getDataOutput());
- feedRecordBuilder.addField(MetadataRecordTypes.FEED_DETAILS_ARECORD_STATE_FIELD_INDEX, fieldValue);
-
- // write field 9
- UnorderedListBuilder unorderedlistBuilder = new UnorderedListBuilder();
- unorderedlistBuilder
- .reset((AUnorderedListType) MetadataRecordTypes.FEED_DETAILS_RECORDTYPE.getFieldTypes()[MetadataRecordTypes.FEED_DETAILS_ARECORD_INGEST_NODES_FIELD_INDEX]);
- itemValue = new ArrayBackedValueStorage();
- if (ingestNodes != null) {
- for (String node : ingestNodes) {
- itemValue.reset();
- aString.setValue(node);
- listBuilder.addItem(itemValue);
- }
- }
- fieldValue.reset();
- unorderedlistBuilder.write(fieldValue.getDataOutput(), true);
- feedRecordBuilder.addField(MetadataRecordTypes.FEED_DETAILS_ARECORD_INGEST_NODES_FIELD_INDEX, fieldValue);
-
- // write field 10
- unorderedlistBuilder = new UnorderedListBuilder();
- unorderedlistBuilder
- .reset((AUnorderedListType) MetadataRecordTypes.FEED_DETAILS_RECORDTYPE.getFieldTypes()[MetadataRecordTypes.FEED_DETAILS_ARECORD_COMPUTE_NODES_FIELD_INDEX]);
- itemValue = new ArrayBackedValueStorage();
- if (computeNodes != null) {
- for (String node : computeNodes) {
- itemValue.reset();
- aString.setValue(node);
- listBuilder.addItem(itemValue);
- }
- }
- fieldValue.reset();
- unorderedlistBuilder.write(fieldValue.getDataOutput(), true);
- feedRecordBuilder.addField(MetadataRecordTypes.FEED_DETAILS_ARECORD_COMPUTE_NODES_FIELD_INDEX, fieldValue);
-
try {
feedRecordBuilder.write(out, true);
} catch (IOException | AsterixException e) {
@@ -235,14 +188,6 @@
}
}
- public FeedState getFeedState() {
- return feedState;
- }
-
- public void setFeedState(FeedState feedState) {
- this.feedState = feedState;
- }
-
public String getAdapterFactory() {
return adapterFactory;
}
@@ -255,20 +200,4 @@
return signature;
}
- public List<String> getIngestNodes() {
- return ingestNodes;
- }
-
- public void setIngestNodes(List<String> ingestNodes) {
- this.ingestNodes = ingestNodes;
- }
-
- public List<String> getComputeNodes() {
- return computeNodes;
- }
-
- public void setComputeNodes(List<String> computeNodes) {
- this.computeNodes = computeNodes;
- }
-
}
\ No newline at end of file
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java
index 65b2d78..9ba5848 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java
@@ -166,26 +166,8 @@
Integer.parseInt(nameComponents[1]));
}
- String feedState = ((AString) datasetDetailsRecord
- .getValueByPos(MetadataRecordTypes.FEED_DETAILS_ARECORD_STATE_FIELD_INDEX)).getStringValue();
-
- cursor = ((AUnorderedList) datasetDetailsRecord
- .getValueByPos(MetadataRecordTypes.FEED_DETAILS_ARECORD_INGEST_NODES_FIELD_INDEX)).getCursor();
- List<String> ingestNodes = new ArrayList<String>();
- while (cursor.next()) {
- ingestNodes.add(((AString) cursor.get()).getStringValue());
- }
-
- cursor = ((AUnorderedList) datasetDetailsRecord
- .getValueByPos(MetadataRecordTypes.FEED_DETAILS_ARECORD_INGEST_NODES_FIELD_INDEX)).getCursor();
- List<String> computeNodes = new ArrayList<String>();
- while (cursor.next()) {
- computeNodes.add(((AString) cursor.get()).getStringValue());
- }
-
datasetDetails = new FeedDatasetDetails(fileStructure, partitioningStrategy, partitioningKey,
- partitioningKey, groupName, adapter, properties, signature, feedState, ingestNodes,
- computeNodes);
+ partitioningKey, groupName, adapter, properties, signature);
break;
}
case INTERNAL: {
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractDatasourceAdapter.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractDatasourceAdapter.java
index 7684969..9e8e5f7 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractDatasourceAdapter.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractDatasourceAdapter.java
@@ -17,17 +17,9 @@
import java.util.HashMap;
import java.util.Map;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.om.types.ATypeTag;
import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.DoubleParserFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.FloatParserFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.IntegerParserFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.LongParserFactory;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.UTF8StringParserFactory;
/**
* Represents the base class that is required to be extended by every
@@ -41,16 +33,6 @@
protected transient AlgebricksPartitionConstraint partitionConstraint;
protected IAType atype;
protected IHyracksTaskContext ctx;
- protected AdapterType adapterType;
-
- protected static final HashMap<ATypeTag, IValueParserFactory> typeToValueParserFactMap = new HashMap<ATypeTag, IValueParserFactory>();
- static {
- typeToValueParserFactMap.put(ATypeTag.INT32, IntegerParserFactory.INSTANCE);
- typeToValueParserFactMap.put(ATypeTag.FLOAT, FloatParserFactory.INSTANCE);
- typeToValueParserFactMap.put(ATypeTag.DOUBLE, DoubleParserFactory.INSTANCE);
- typeToValueParserFactMap.put(ATypeTag.INT64, LongParserFactory.INSTANCE);
- typeToValueParserFactMap.put(ATypeTag.STRING, UTF8StringParserFactory.INSTANCE);
- }
protected static final Map<String, Object> formatToParserFactoryMap = initializeFormatParserFactoryMap();
@@ -66,29 +48,4 @@
return map;
}
- /**
- * Get the partition constraint chosen by the adapter.
- * An adapter may have preferences as to where it needs to be instantiated and used.
- */
- public abstract AlgebricksPartitionConstraint getPartitionConstraint() throws Exception;
-
- /**
- * Get the configured value from the adapter configuration parameters, corresponding to the an attribute.
- *
- * @param attribute
- * The attribute whose value needs to be obtained.
- */
- public Object getAdapterProperty(String attribute) {
- return configuration.get(attribute);
- }
-
- /**
- * Get the adapter configuration parameters.
- *
- * @return A Map<String,String> instance representing the adapter configuration.
- */
- public Map<String, Object> getConfiguration() {
- return configuration;
- }
-
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractFeedDatasourceAdapter.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractFeedDatasourceAdapter.java
index 08d6972..96dc8dc 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractFeedDatasourceAdapter.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/AbstractFeedDatasourceAdapter.java
@@ -12,7 +12,7 @@
import edu.uci.ics.asterix.metadata.entities.FeedActivity.FeedActivityDetails;
import edu.uci.ics.asterix.metadata.entities.FeedActivity.FeedActivityType;
-public abstract class AbstractFeedDatasourceAdapter extends AbstractDatasourceAdapter {
+public abstract class AbstractFeedDatasourceAdapter implements IDatasourceAdapter {
private static final long serialVersionUID = 1L;
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/ExternalDataScanOperatorDescriptor.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/ExternalDataScanOperatorDescriptor.java
index 07e1594..c5937f6 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/ExternalDataScanOperatorDescriptor.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/ExternalDataScanOperatorDescriptor.java
@@ -14,9 +14,6 @@
*/
package edu.uci.ics.asterix.metadata.feeds;
-import java.util.Map;
-
-import edu.uci.ics.asterix.om.types.IAType;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
@@ -34,17 +31,13 @@
private static final long serialVersionUID = 1L;
- private final Map<String, Object> adapterConfiguration;
- private final IAType atype;
- private IGenericDatasetAdapterFactory datasourceAdapterFactory;
+ private IAdapterFactory adapterFactory;
- public ExternalDataScanOperatorDescriptor(JobSpecification spec, Map<String, Object> arguments, IAType atype,
- RecordDescriptor rDesc, IGenericDatasetAdapterFactory dataSourceAdapterFactory) {
+ public ExternalDataScanOperatorDescriptor(JobSpecification spec, RecordDescriptor rDesc,
+ IAdapterFactory dataSourceAdapterFactory) {
super(spec, 0, 1);
recordDescriptors[0] = rDesc;
- this.adapterConfiguration = arguments;
- this.atype = atype;
- this.datasourceAdapterFactory = dataSourceAdapterFactory;
+ this.adapterFactory = dataSourceAdapterFactory;
}
@Override
@@ -58,9 +51,7 @@
writer.open();
IDatasourceAdapter adapter = null;
try {
- adapter = ((IGenericDatasetAdapterFactory) datasourceAdapterFactory).createAdapter(
- adapterConfiguration, atype);
- adapter.initialize(ctx);
+ adapter = ((IAdapterFactory) adapterFactory).createAdapter(ctx);
adapter.start(partition, writer);
} catch (Exception e) {
throw new HyracksDataException("exception during reading from external data source", e);
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorDescriptor.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorDescriptor.java
index 96be46a..4270fe2 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorDescriptor.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorDescriptor.java
@@ -34,20 +34,16 @@
private static final long serialVersionUID = 1L;
- private final String adapterFactoryClassName;
- private final Map<String, Object> adapterConfiguration;
private final IAType atype;
private final FeedId feedId;
private final Map<String, String> feedPolicy;
+ private IAdapterFactory adapterFactory;
- private transient IAdapterFactory datasourceAdapterFactory;
-
- public FeedIntakeOperatorDescriptor(JobSpecification spec, FeedId feedId, String adapter,
- Map<String, Object> arguments, ARecordType atype, RecordDescriptor rDesc, Map<String, String> feedPolicy) {
+ public FeedIntakeOperatorDescriptor(JobSpecification spec, FeedId feedId, IAdapterFactory adapterFactory,
+ ARecordType atype, RecordDescriptor rDesc, Map<String, String> feedPolicy) {
super(spec, 1, 1);
recordDescriptors[0] = rDesc;
- this.adapterFactoryClassName = adapter;
- this.adapterConfiguration = arguments;
+ this.adapterFactory = adapterFactory;
this.atype = atype;
this.feedId = feedId;
this.feedPolicy = feedPolicy;
@@ -56,19 +52,9 @@
public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
IRecordDescriptorProvider recordDescProvider, final int partition, int nPartitions)
throws HyracksDataException {
- ITypedDatasourceAdapter adapter;
+ IDatasourceAdapter adapter;
try {
- datasourceAdapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClassName).newInstance();
- if (datasourceAdapterFactory instanceof IGenericDatasetAdapterFactory) {
- adapter = (ITypedDatasourceAdapter) ((IGenericDatasetAdapterFactory) datasourceAdapterFactory)
- .createAdapter(adapterConfiguration, atype);
- } else if (datasourceAdapterFactory instanceof ITypedDatasetAdapterFactory) {
- adapter = (ITypedDatasourceAdapter) ((ITypedDatasetAdapterFactory) datasourceAdapterFactory)
- .createAdapter(adapterConfiguration);
- } else {
- throw new IllegalStateException(" Unknown adapter factory type for " + adapterFactoryClassName);
- }
- adapter.initialize(ctx);
+ adapter = adapterFactory.createAdapter(ctx);
} catch (Exception e) {
throw new HyracksDataException("initialization of adapter failed", e);
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorNodePushable.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorNodePushable.java
index f6bde26..97153f0 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorNodePushable.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/FeedIntakeOperatorNodePushable.java
@@ -50,23 +50,25 @@
@Override
public void open() throws HyracksDataException {
- if (adapter instanceof IManagedFeedAdapter) {
- feedInboxMonitor = new FeedInboxMonitor((IManagedFeedAdapter) adapter, inbox, partition);
+ if (adapter instanceof IFeedAdapter) {
+ feedInboxMonitor = new FeedInboxMonitor((IFeedAdapter) adapter, inbox, partition);
AsterixThreadExecutor.INSTANCE.execute(feedInboxMonitor);
feedManager.registerFeedMsgQueue(feedId, inbox);
}
writer.open();
try {
- ((AbstractFeedDatasourceAdapter) adapter).setFeedPolicyEnforcer(policyEnforcer);
+ if (adapter instanceof AbstractFeedDatasourceAdapter) {
+ ((AbstractFeedDatasourceAdapter) adapter).setFeedPolicyEnforcer(policyEnforcer);
+ }
adapter.start(partition, writer);
} catch (Exception e) {
e.printStackTrace();
throw new HyracksDataException(e);
} finally {
writer.close();
- if (adapter instanceof IManagedFeedAdapter) {
+ if (adapter instanceof IFeedAdapter) {
try {
- ((IManagedFeedAdapter) adapter).stop();
+ ((IFeedAdapter) adapter).stop();
} catch (Exception e) {
e.printStackTrace();
}
@@ -98,9 +100,9 @@
class FeedInboxMonitor extends Thread {
private LinkedBlockingQueue<IFeedMessage> inbox;
- private final IManagedFeedAdapter adapter;
+ private final IFeedAdapter adapter;
- public FeedInboxMonitor(IManagedFeedAdapter adapter, LinkedBlockingQueue<IFeedMessage> inbox, int partition) {
+ public FeedInboxMonitor(IFeedAdapter adapter, LinkedBlockingQueue<IFeedMessage> inbox, int partition) {
this.inbox = inbox;
this.adapter = adapter;
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IAdapterFactory.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IAdapterFactory.java
index 85e09d4..cc8e165 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IAdapterFactory.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IAdapterFactory.java
@@ -15,6 +15,10 @@
package edu.uci.ics.asterix.metadata.feeds;
import java.io.Serializable;
+import java.util.Map;
+
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* Base interface for IGenericDatasetAdapterFactory and ITypedDatasetAdapterFactory.
@@ -24,9 +28,72 @@
public interface IAdapterFactory extends Serializable {
/**
+ * A 'GENERIC' adapter can be configured to return a given datatype.
+ * A 'TYPED' adapter returns records with a pre-defined datatype.
+ */
+ public enum AdapterType {
+ GENERIC,
+ TYPED
+ }
+
+ public enum SupportedOperation {
+ READ,
+ WRITE,
+ READ_WRITE
+ }
+
+ /**
+ * Returns the type of adapter indicating if the adapter can be used for
+ * reading from an external data source or writing to an external data
+ * source or can be used for both purposes.
+ *
+ * @see SupportedOperation
+ * @return
+ */
+ public SupportedOperation getSupportedOperations();
+
+ /**
* Returns the display name corresponding to the Adapter type that is created by the factory.
*
* @return the display name
*/
public String getName();
+
+ /**
+ * Returns the type of the adapter (GENERIC or TYPED)
+ *
+ * @return
+ */
+ public AdapterType getAdapterType();
+
+ /**
+ * @param configuration
+ */
+ public void configure(Map<String, Object> configuration) throws Exception;
+
+ /**
+ * Returns a list of partition constraints. A partition constraint can be a
+ * requirement to execute at a particular location or could be cardinality
+ * constraints indicating the number of instances that need to run in
+ * parallel. example, a IDatasourceAdapter implementation written for data
+ * residing on the local file system of a node cannot run on any other node
+ * and thus has a location partition constraint. The location partition
+ * constraint can be expressed as a node IP address or a node controller id.
+ * In the former case, the IP address is translated to a node controller id
+ * running on the node with the given IP address.
+ *
+ * @Caller The wrapper operator configures its partition constraints from
+ * the constraints obtained from the adapter factory.
+ */
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception;
+
+ /**
+ * Creates an instance of IDatasourceAdapter.
+ *
+ * @param HyracksTaskContext
+ * @return An instance of IDatasourceAdapter.
+ * @throws Exception
+ */
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception;
+
}
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IDatasourceAdapter.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IDatasourceAdapter.java
index a117f8a..d0f8cdd 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IDatasourceAdapter.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/feeds/IDatasourceAdapter.java
@@ -15,11 +15,8 @@
package edu.uci.ics.asterix.metadata.feeds;
import java.io.Serializable;
-import java.util.Map;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.comm.IFrameWriter;
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* A super interface implemented by a data source adapter. An adapter can be a
@@ -39,101 +36,10 @@
* external dataset determines the set of valid operations allowed
* on the dataset.
*/
- public enum AdapterType {
- READ,
- WRITE,
- READ_WRITE
- }
+
/**
- * Returns the type of adapter indicating if the adapter can be used for
- * reading from an external data source or writing to an external data
- * source or can be used for both purposes.
- *
- * @Caller: Compiler: The compiler uses this API to verify if an operation
- * is supported by the adapter. For example, an write query against
- * an external dataset will not compile successfully if the
- * external dataset was declared with a read_only adapter.
- * @see AdapterType
- * @return
- */
- public AdapterType getAdapterType();
-
- /**
- * Each adapter instance is configured with a set of parameters that are
- * key-value pairs. When creating an external or a feed dataset, an adapter
- * instance is used in conjunction with a set of configuration parameters
- * for the adapter instance. The configuration parameters are stored
- * internally with the adapter and can be retrieved using this API.
- *
- * @param propertyKey
- * @return String the value corresponding to the configuration parameter
- * represented by the key- attributeKey.
- */
- public Object getAdapterProperty(String propertyKey);
-
- /**
- * Configures the IDatasourceAdapter instance.
- *
- * @caller Scenario 1) Called during compilation of DDL statement that
- * creates a Feed dataset and associates the adapter with the
- * dataset. The (key,value) configuration parameters provided as
- * part of the DDL statement are collected by the compiler and
- * passed on to this method. The adapter may as part of
- * configuration connect with the external data source and determine
- * the IAType associated with data residing with the external
- * datasource.
- * Scenario 2) An adapter instance is created by an ASTERIX operator
- * that wraps around the adapter instance. The operator, as part of
- * its initialization invokes the configure method. The (key,value)
- * configuration parameters are passed on to the operator by the
- * compiler. Subsequent to the invocation, the wrapping operator
- * obtains the partition constraints (if any). In addition, in the
- * case of a read adapter, the wrapping operator obtains the output
- * ASTERIX type associated with the data that will be output from
- * the adapter.
- * @param arguments
- * A map with key-value pairs that contains the configuration
- * parameters for the adapter. The arguments are obtained from
- * the metadata. Recall that the DDL to create an external
- * dataset or a feed dataset requires using an adapter and
- * providing all arguments as a set of (key,value) pairs. These
- * arguments are put into the metadata.
- */
- public void configure(Map<String, Object> arguments) throws Exception;
-
- /**
- * Returns a list of partition constraints. A partition constraint can be a
- * requirement to execute at a particular location or could be cardinality
- * constraints indicating the number of instances that need to run in
- * parallel. example, a IDatasourceAdapter implementation written for data
- * residing on the local file system of a node cannot run on any other node
- * and thus has a location partition constraint. The location partition
- * constraint can be expressed as a node IP address or a node controller id.
- * In the former case, the IP address is translated to a node controller id
- * running on the node with the given IP address.
- *
- * @Caller The wrapper operator configures its partition constraints from
- * the constraints obtained from the adapter.
- */
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception;
-
- /**
- * Allows the adapter to establish connection with the external data source
- * expressing intent for data and providing any configuration parameters
- * required by the external data source for the transfer of data. This
- * method does not result in any data transfer, but is a prerequisite for
- * any subsequent data transfer to happen between the external data source
- * and the adapter.
- *
- * @caller This method is called by the wrapping ASTERIX operator that
- * @param ctx
- * @throws Exception
- */
- public void initialize(IHyracksTaskContext ctx) throws Exception;
-
- /**
- * Triggers the adapter to begin ingestion of data from the external source.
+ * Triggers the adapter to begin ingesting data from the external source.
*
* @param partition
* The adapter could be running with a degree of parallelism.
diff --git a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixClusterProperties.java b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixClusterProperties.java
index 6285385..29e3b3b 100644
--- a/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixClusterProperties.java
+++ b/asterix-om/src/main/java/edu/uci/ics/asterix/om/util/AsterixClusterProperties.java
@@ -44,16 +44,20 @@
public static final String CLUSTER_CONFIGURATION_FILE = "cluster.xml";
private final Cluster cluster;
-
+
private AsterixClusterProperties() {
InputStream is = this.getClass().getClassLoader().getResourceAsStream(CLUSTER_CONFIGURATION_FILE);
- try {
- JAXBContext ctx = JAXBContext.newInstance(Cluster.class);
- Unmarshaller unmarshaller = ctx.createUnmarshaller();
- cluster = (Cluster) unmarshaller.unmarshal(is);
-
- } catch (JAXBException e) {
- throw new IllegalStateException("Failed to read configuration file " + CLUSTER_CONFIGURATION_FILE);
+ if (is != null) {
+ try {
+ JAXBContext ctx = JAXBContext.newInstance(Cluster.class);
+ Unmarshaller unmarshaller = ctx.createUnmarshaller();
+ cluster = (Cluster) unmarshaller.unmarshal(is);
+
+ } catch (JAXBException e) {
+ throw new IllegalStateException("Failed to read configuration file " + CLUSTER_CONFIGURATION_FILE);
+ }
+ } else {
+ cluster = null;
}
}
diff --git a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapter.java b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapter.java
index 32b539c..81d19ae 100644
--- a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapter.java
+++ b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapter.java
@@ -18,24 +18,10 @@
import java.io.InputStream;
import java.util.Map;
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.external.dataset.adapter.FileSystemBasedAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IManagedFeedAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasourceAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapter;
import edu.uci.ics.asterix.om.types.ARecordType;
-import edu.uci.ics.asterix.om.types.ATypeTag;
-import edu.uci.ics.asterix.runtime.operators.file.ADMDataParser;
-import edu.uci.ics.asterix.runtime.operators.file.AbstractTupleParser;
-import edu.uci.ics.asterix.runtime.operators.file.DelimitedDataParser;
-import edu.uci.ics.asterix.runtime.operators.file.IDataParser;
-import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
-import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
-import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
-import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
-import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
-import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
/**
@@ -44,19 +30,16 @@
* the source file has been ingested.
*/
-public class RateControlledFileSystemBasedAdapter extends FileSystemBasedAdapter implements ITypedDatasourceAdapter,
- IManagedFeedAdapter {
+public class RateControlledFileSystemBasedAdapter extends FileSystemBasedAdapter implements IFeedAdapter {
private static final long serialVersionUID = 1L;
private FileSystemBasedAdapter coreAdapter;
- private String format;
public RateControlledFileSystemBasedAdapter(ARecordType atype, Map<String, Object> configuration,
- FileSystemBasedAdapter coreAdapter, String format) throws Exception {
- super(atype);
- this.configuration = configuration;
+ FileSystemBasedAdapter coreAdapter, String format, ITupleParserFactory parserFactory,
+ IHyracksTaskContext ctx) throws Exception {
+ super(parserFactory, atype, ctx);
this.coreAdapter = coreAdapter;
- this.format = format;
}
@Override
@@ -65,183 +48,14 @@
}
@Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- coreAdapter.initialize(ctx);
- this.ctx = ctx;
- }
-
- @Override
- public void configure(Map<String, Object> arguments) throws Exception {
- coreAdapter.configure(arguments);
- }
-
- @Override
- public AdapterType getAdapterType() {
- return coreAdapter.getAdapterType();
- }
-
- @Override
- protected ITupleParser getTupleParser() throws Exception {
- ITupleParser parser = null;
- if (format.equals(FORMAT_DELIMITED_TEXT)) {
- parser = getRateControlledDelimitedDataTupleParser((ARecordType) atype);
- } else if (format.equals(FORMAT_ADM)) {
- parser = getRateControlledADMDataTupleParser((ARecordType) atype);
- } else {
- throw new IllegalArgumentException(" format " + configuration.get(KEY_FORMAT) + " not supported");
- }
- return parser;
-
- }
-
- protected ITupleParser getRateControlledDelimitedDataTupleParser(ARecordType recordType) throws AsterixException {
- ITupleParser parser;
- int n = recordType.getFieldTypes().length;
- IValueParserFactory[] fieldParserFactories = new IValueParserFactory[n];
- for (int i = 0; i < n; i++) {
- ATypeTag tag = recordType.getFieldTypes()[i].getTypeTag();
- IValueParserFactory vpf = typeToValueParserFactMap.get(tag);
- if (vpf == null) {
- throw new NotImplementedException("No value parser factory for delimited fields of type " + tag);
- }
- fieldParserFactories[i] = vpf;
-
- }
- String delimiterValue = (String) configuration.get(KEY_DELIMITER);
- if (delimiterValue != null && delimiterValue.length() > 1) {
- throw new AsterixException("improper delimiter");
- }
-
- Character delimiter = delimiterValue.charAt(0);
- parser = new RateControlledTupleParserFactory(recordType, fieldParserFactories, delimiter, configuration)
- .createTupleParser(ctx);
- return parser;
- }
-
- protected ITupleParser getRateControlledADMDataTupleParser(ARecordType recordType) throws AsterixException {
- ITupleParser parser = null;
- try {
- parser = new RateControlledTupleParserFactory(recordType, configuration).createTupleParser(ctx);
- return parser;
- } catch (Exception e) {
- throw new AsterixException(e);
- }
-
- }
-
- @Override
- public ARecordType getAdapterOutputType() {
- return (ARecordType) atype;
- }
-
- @Override
public void alter(Map<String, Object> properties) {
- ((RateControlledTupleParser) parser).setInterTupleInterval(Long.parseLong((String) properties
+ ((RateControlledTupleParser) tupleParser).setInterTupleInterval(Long.parseLong((String) properties
.get(RateControlledTupleParser.INTER_TUPLE_INTERVAL)));
}
@Override
public void stop() {
- ((RateControlledTupleParser) parser).stop();
+ ((RateControlledTupleParser) tupleParser).stop();
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- return coreAdapter.getPartitionConstraint();
- }
-}
-
-class RateControlledTupleParserFactory implements ITupleParserFactory {
-
- private static final long serialVersionUID = 1L;
-
- private final ARecordType recordType;
- private final IDataParser dataParser;
- private final Map<String, Object> configuration;
-
- public RateControlledTupleParserFactory(ARecordType recordType, IValueParserFactory[] valueParserFactories,
- char fieldDelimiter, Map<String, Object> configuration) {
- this.recordType = recordType;
- dataParser = new DelimitedDataParser(recordType, valueParserFactories, fieldDelimiter);
- this.configuration = configuration;
- }
-
- public RateControlledTupleParserFactory(ARecordType recordType, Map<String, Object> configuration) {
- this.recordType = recordType;
- dataParser = new ADMDataParser();
- this.configuration = configuration;
- }
-
- @Override
- public ITupleParser createTupleParser(IHyracksTaskContext ctx) {
- return new RateControlledTupleParser(ctx, recordType, dataParser, configuration);
- }
-
-}
-
-class RateControlledTupleParser extends AbstractTupleParser {
-
- private final IDataParser dataParser;
- private long interTupleInterval;
- private boolean delayConfigured;
- private boolean continueIngestion = true;
-
- public static final String INTER_TUPLE_INTERVAL = "tuple-interval";
-
- public RateControlledTupleParser(IHyracksTaskContext ctx, ARecordType recType, IDataParser dataParser,
- Map<String, Object> configuration) {
- super(ctx, recType);
- this.dataParser = dataParser;
- String propValue = (String) configuration.get(INTER_TUPLE_INTERVAL);
- if (propValue != null) {
- interTupleInterval = Long.parseLong(propValue);
- } else {
- interTupleInterval = 0;
- }
- delayConfigured = interTupleInterval != 0;
- }
-
- public void setInterTupleInterval(long val) {
- this.interTupleInterval = val;
- this.delayConfigured = val > 0;
- }
-
- public void stop() {
- continueIngestion = false;
- }
-
- @Override
- public IDataParser getDataParser() {
- return dataParser;
- }
-
- @Override
- public void parse(InputStream in, IFrameWriter writer) throws HyracksDataException {
-
- appender.reset(frame, true);
- IDataParser parser = getDataParser();
- try {
- parser.initialize(in, recType, true);
- while (continueIngestion) {
- tb.reset();
- if (!parser.parse(tb.getDataOutput())) {
- break;
- }
- tb.addFieldEndOffset();
- if (delayConfigured) {
- Thread.sleep(interTupleInterval);
- }
- addTupleToFrame(writer);
- }
- if (appender.getTupleCount() > 0) {
- FrameUtils.flushFrame(frame, writer);
- }
- } catch (AsterixException ae) {
- throw new HyracksDataException(ae);
- } catch (IOException ioe) {
- throw new HyracksDataException(ioe);
- } catch (InterruptedException ie) {
- throw new HyracksDataException(ie);
- }
- }
}
diff --git a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapterFactory.java b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapterFactory.java
index ff3108f..16812d9 100644
--- a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapterFactory.java
+++ b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/RateControlledFileSystemBasedAdapterFactory.java
@@ -14,14 +14,32 @@
*/
package edu.uci.ics.asterix.tools.external.data;
+import java.io.IOException;
+import java.io.InputStream;
import java.util.Map;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.external.adapter.factory.FileSystemAdapterFactory;
+import edu.uci.ics.asterix.external.adapter.factory.HDFSAdapterFactory;
+import edu.uci.ics.asterix.external.adapter.factory.NCFileSystemAdapterFactory;
import edu.uci.ics.asterix.external.dataset.adapter.FileSystemBasedAdapter;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericDatasetAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
-import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.om.types.ATypeTag;
+import edu.uci.ics.asterix.runtime.operators.file.ADMDataParser;
+import edu.uci.ics.asterix.runtime.operators.file.AbstractTupleParser;
+import edu.uci.ics.asterix.runtime.operators.file.DelimitedDataParser;
+import edu.uci.ics.asterix.runtime.operators.file.IDataParser;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
+import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
+import edu.uci.ics.hyracks.dataflow.common.data.parsers.IValueParserFactory;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;
+import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
/**
* Factory class for creating @see{RateControllerFileSystemBasedAdapter} The
@@ -29,38 +47,24 @@
* on the local file system or on HDFS. The feed ends when the content of the
* source file has been ingested.
*/
-public class RateControlledFileSystemBasedAdapterFactory implements IGenericDatasetAdapterFactory {
+public class RateControlledFileSystemBasedAdapterFactory extends FileSystemAdapterFactory {
private static final long serialVersionUID = 1L;
-
+
public static final String KEY_FILE_SYSTEM = "fs";
public static final String LOCAL_FS = "localfs";
public static final String HDFS = "hdfs";
public static final String KEY_PATH = "path";
public static final String KEY_FORMAT = "format";
- private IGenericDatasetAdapterFactory adapterFactory;
+ private IAdapterFactory adapterFactory;
private String format;
- private boolean setup = false;
+ private Map<String, Object> configuration;
+ private ARecordType atype;
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration, IAType type) throws Exception {
- if (!setup) {
- checkRequiredArgs(configuration);
- String fileSystem = (String) configuration.get(KEY_FILE_SYSTEM);
- String adapterFactoryClass = null;
- if (fileSystem.equalsIgnoreCase(LOCAL_FS)) {
- adapterFactoryClass = "edu.uci.ics.asterix.external.adapter.factory.NCFileSystemAdapterFactory";
- } else if (fileSystem.equals(HDFS)) {
- adapterFactoryClass = "edu.uci.ics.asterix.external.adapter.factory.HDFSAdapterFactory";
- } else {
- throw new AsterixException("Unsupported file system type " + fileSystem);
- }
- format = (String) configuration.get(KEY_FORMAT);
- adapterFactory = (IGenericDatasetAdapterFactory) Class.forName(adapterFactoryClass).newInstance();
- setup = true;
- }
- return new RateControlledFileSystemBasedAdapter((ARecordType) type, configuration,
- (FileSystemBasedAdapter) adapterFactory.createAdapter(configuration, type), format);
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ FileSystemBasedAdapter coreAdapter = (FileSystemBasedAdapter) adapterFactory.createAdapter(ctx);
+ return new RateControlledFileSystemBasedAdapter(atype, configuration, coreAdapter, format, parserFactory, ctx);
}
@Override
@@ -72,7 +76,7 @@
if (configuration.get(KEY_FILE_SYSTEM) == null) {
throw new Exception("File system type not specified. (fs=?) File system could be 'localfs' or 'hdfs'");
}
- if (configuration.get(IGenericDatasetAdapterFactory.KEY_TYPE_NAME) == null) {
+ if (configuration.get(KEY_SOURCE_DATATYPE) == null) {
throw new Exception("Record type not specified (output-type-name=?)");
}
if (configuration.get(KEY_PATH) == null) {
@@ -83,4 +87,186 @@
}
}
+ @Override
+ public AdapterType getAdapterType() {
+ return AdapterType.GENERIC;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ checkRequiredArgs(configuration);
+ String fileSystem = (String) configuration.get(KEY_FILE_SYSTEM);
+ String adapterFactoryClass = null;
+ if (fileSystem.equalsIgnoreCase(LOCAL_FS)) {
+ adapterFactoryClass = NCFileSystemAdapterFactory.class.getName();
+ } else if (fileSystem.equals(HDFS)) {
+ adapterFactoryClass = HDFSAdapterFactory.class.getName();
+ } else {
+ throw new AsterixException("Unsupported file system type " + fileSystem);
+ }
+ format = (String) configuration.get(KEY_FORMAT);
+ adapterFactory = (IAdapterFactory) Class.forName(adapterFactoryClass).newInstance();
+ adapterFactory.configure(configuration);
+
+ atype = (ARecordType) configuration.get(KEY_SOURCE_DATATYPE);
+ configureFormat();
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return adapterFactory.getPartitionConstraint();
+ }
+
+ private void configureFormat() throws AsterixException {
+ switch (format) {
+ case FORMAT_ADM:
+ parserFactory = new RateControlledTupleParserFactory(atype, configuration);
+ break;
+
+ case FORMAT_DELIMITED_TEXT:
+ String delimiterValue = (String) configuration.get(KEY_DELIMITER);
+ if (delimiterValue != null && delimiterValue.length() > 1) {
+ throw new AsterixException("improper delimiter");
+ }
+ IValueParserFactory[] valueParserFactories = getValueParserFactories(atype);
+ parserFactory = new RateControlledTupleParserFactory(atype, valueParserFactories,
+ delimiterValue.charAt(0), configuration);
+ break;
+ }
+ }
+
+ protected IValueParserFactory[] getValueParserFactories(ARecordType recordType) throws AsterixException {
+ int n = recordType.getFieldTypes().length;
+ IValueParserFactory[] fieldParserFactories = new IValueParserFactory[n];
+ for (int i = 0; i < n; i++) {
+ ATypeTag tag = recordType.getFieldTypes()[i].getTypeTag();
+ IValueParserFactory vpf = typeToValueParserFactMap.get(tag);
+ if (vpf == null) {
+ throw new NotImplementedException("No value parser factory for delimited fields of type " + tag);
+ }
+ fieldParserFactories[i] = vpf;
+
+ }
+ return fieldParserFactories;
+ }
+
+}
+
+class RateControlledTupleParserFactory implements ITupleParserFactory {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ARecordType recordType;
+ private final Map<String, Object> configuration;
+ private IValueParserFactory[] valueParserFactories;
+ private char delimiter;
+ private final ParserType parserType;
+
+ public enum ParserType {
+ ADM,
+ DELIMITED_DATA
+ }
+
+ public RateControlledTupleParserFactory(ARecordType recordType, IValueParserFactory[] valueParserFactories,
+ char fieldDelimiter, Map<String, Object> configuration) {
+ this.recordType = recordType;
+ this.valueParserFactories = valueParserFactories;
+ this.delimiter = fieldDelimiter;
+ this.configuration = configuration;
+ this.parserType = ParserType.DELIMITED_DATA;
+ }
+
+ public RateControlledTupleParserFactory(ARecordType recordType, Map<String, Object> configuration) {
+ this.recordType = recordType;
+ this.configuration = configuration;
+ this.parserType = ParserType.ADM;
+ }
+
+ @Override
+ public ITupleParser createTupleParser(IHyracksTaskContext ctx) {
+ IDataParser dataParser = null;
+ switch (parserType) {
+ case ADM:
+ dataParser = new ADMDataParser();
+ break;
+ case DELIMITED_DATA:
+ dataParser = new DelimitedDataParser(recordType, valueParserFactories, delimiter);
+ break;
+ }
+ return new RateControlledTupleParser(ctx, recordType, dataParser, configuration);
+ }
+
+}
+
+class RateControlledTupleParser extends AbstractTupleParser {
+
+ private final IDataParser dataParser;
+ private long interTupleInterval;
+ private boolean delayConfigured;
+ private boolean continueIngestion = true;
+
+ public static final String INTER_TUPLE_INTERVAL = "tuple-interval";
+
+ public RateControlledTupleParser(IHyracksTaskContext ctx, ARecordType recType, IDataParser dataParser,
+ Map<String, Object> configuration) {
+ super(ctx, recType);
+ this.dataParser = dataParser;
+ String propValue = (String) configuration.get(INTER_TUPLE_INTERVAL);
+ if (propValue != null) {
+ interTupleInterval = Long.parseLong(propValue);
+ } else {
+ interTupleInterval = 0;
+ }
+ delayConfigured = interTupleInterval != 0;
+ }
+
+ public void setInterTupleInterval(long val) {
+ this.interTupleInterval = val;
+ this.delayConfigured = val > 0;
+ }
+
+ public void stop() {
+ continueIngestion = false;
+ }
+
+ @Override
+ public IDataParser getDataParser() {
+ return dataParser;
+ }
+
+ @Override
+ public void parse(InputStream in, IFrameWriter writer) throws HyracksDataException {
+
+ appender.reset(frame, true);
+ IDataParser parser = getDataParser();
+ try {
+ parser.initialize(in, recType, true);
+ while (continueIngestion) {
+ tb.reset();
+ if (!parser.parse(tb.getDataOutput())) {
+ break;
+ }
+ tb.addFieldEndOffset();
+ if (delayConfigured) {
+ Thread.sleep(interTupleInterval);
+ }
+ addTupleToFrame(writer);
+ }
+ if (appender.getTupleCount() > 0) {
+ FrameUtils.flushFrame(frame, writer);
+ }
+ } catch (AsterixException ae) {
+ throw new HyracksDataException(ae);
+ } catch (IOException ioe) {
+ throw new HyracksDataException(ioe);
+ } catch (InterruptedException ie) {
+ throw new HyracksDataException(ie);
+ }
+ }
}
\ No newline at end of file
diff --git a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapter.java b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapter.java
index 63ae8f2..fe4212d 100644
--- a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapter.java
+++ b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapter.java
@@ -38,7 +38,9 @@
private static final long serialVersionUID = 1L;
private Map<String, Object> configuration;
- public SyntheticTwitterFeedAdapter(Map<String, Object> configuration) throws AsterixException {
+ public SyntheticTwitterFeedAdapter(Map<String, Object> configuration, IHyracksTaskContext ctx)
+ throws AsterixException {
+ super(configuration, ctx);
this.configuration = configuration;
String[] userFieldNames = new String[] { "screen-name", "lang", "friends_count", "statuses_count", "name",
@@ -58,31 +60,11 @@
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.READ;
- }
-
- @Override
- public void configure(Map<String, Object> configuration) throws Exception {
- this.configuration = configuration;
-
- }
-
- @Override
- public void initialize(IHyracksTaskContext ctx) throws Exception {
- this.ctx = ctx;
- }
-
- @Override
public IPullBasedFeedClient getFeedClient(int partition) throws Exception {
return new SyntheticTwitterFeedClient(configuration, adapterOutputType, partition);
}
- @Override
- public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- return new AlgebricksCountPartitionConstraint(1);
- }
-
+
private static class SyntheticTwitterFeedClient extends PullBasedFeedClient implements IPullBasedFeedClient {
private static final Logger LOGGER = Logger.getLogger(SyntheticTwitterFeedClient.class.getName());
diff --git a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapterFactory.java b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapterFactory.java
index 7832f11..48b1d20 100644
--- a/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapterFactory.java
+++ b/asterix-tools/src/main/java/edu/uci/ics/asterix/tools/external/data/SyntheticTwitterFeedAdapterFactory.java
@@ -16,8 +16,11 @@
import java.util.Map;
+import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedDatasetAdapterFactory;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
+import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* Factory class for creating @see{RateControllerFileSystemBasedAdapter} The
@@ -25,21 +28,43 @@
* on the local file system or on HDFS. The feed ends when the content of the
* source file has been ingested.
*/
-public class SyntheticTwitterFeedAdapterFactory implements ITypedDatasetAdapterFactory {
+public class SyntheticTwitterFeedAdapterFactory implements IAdapterFactory {
/**
*
*/
private static final long serialVersionUID = 1L;
+ private Map<String, Object> configuration;
+
@Override
public String getName() {
return "synthetic_twitter_feed";
}
@Override
- public IDatasourceAdapter createAdapter(Map<String, Object> configuration) throws Exception {
- return new SyntheticTwitterFeedAdapter(configuration);
+ public AdapterType getAdapterType() {
+ return AdapterType.TYPED;
+ }
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public void configure(Map<String, Object> configuration) throws Exception {
+ this.configuration = configuration;
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return new AlgebricksCountPartitionConstraint(1);
+ }
+
+ @Override
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx) throws Exception {
+ return new SyntheticTwitterFeedAdapter(configuration, ctx);
}
}
\ No newline at end of file