You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by sj...@apache.org on 2015/06/29 21:45:08 UTC
[11/24] incubator-asterixdb git commit: Introduces Feeds 2.0
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-doc/src/site/markdown/feeds/tutorial.md
----------------------------------------------------------------------
diff --git a/asterix-doc/src/site/markdown/feeds/tutorial.md b/asterix-doc/src/site/markdown/feeds/tutorial.md
new file mode 100644
index 0000000..5884449
--- /dev/null
+++ b/asterix-doc/src/site/markdown/feeds/tutorial.md
@@ -0,0 +1,287 @@
+# Support for Data Ingestion in AsterixDB #
+
+## <a id="toc">Table of Contents</a> ##
+
+* [Introduction](#Introduction)
+ * [Data Feed Basics](#DataFeedBasics)
+ * [Collecting Data: Feed Adaptors](#FeedAdaptors)
+ * [Preprocessing Collected Data](#PreprocessingCollectedData)
+ * [Creating an External Dataset](#IntroductionCreatingAnExternalDataset)
+
+## <a id="Introduction">Introduction</a> <font size="4"><a href="#toc">[Back to TOC]</a></font> ##
+In this document, we describe the support for data ingestion in AsterixDB, an open-source Big Data Management System (BDMS) that provides a platform for storage and analysis of large volumes of semi-structured data. Data feeds are a new mechanism for having
+continuous data arrive into a BDMS from external sources and incrementally populate a persisted dataset and associated indexes. We add a new BDMS architectural component, called a data feed, that makes a Big Data system the caretaker for functionality that
+used to live outside, and we show how it improves users’ lives and system performance.
+
+### <a id="DataFeedBasics">Data Feed Basics</a> <font size="4"><a href="#toc">[Back to TOC]</a></font> ###
+ ####Collecting Data: Feed Adaptors####
+The functionality of establishing a connection with a data source
+and receiving, parsing and translating its data into ADM records
+(for storage inside AsterixDB) is contained in a feed adaptor. A
+feed adaptor is an implementation of an interface and its details are
+specific to a given data source. An adaptor may optionally be given
+parameters to configure its runtime behavior. Depending upon the
+data transfer protocol/APIs offered by the data source, a feed adaptor
+may operate in a push or a pull mode. Push mode involves just
+one initial request by the adaptor to the data source for setting up
+the connection. Once a connection is authorized, the data source
+“pushes” data to the adaptor without any subsequent requests by
+the adaptor. In contrast, when operating in a pull mode, the adaptor
+makes a separate request each time to receive data.
+AsterixDB currently provides built-in adaptors for several popular
+data sources—Twitter, CNN, and RSS feeds. AsterixDB additionally
+provides a generic socket-based adaptor that can be used
+to ingest data that is directed at a prescribed socket.
+
+Next, we consider creating an example feed that consists of tweets obtained from
+the Twitter service. To do so, we use the built-in push-based Twitter adaptor.
+To being with, we must define a Tweet using the AsterixDB Data Model (ADM) and the AsterixDB Query Language (AQL). Given below are the type definitions in AQL that create a Tweet datatype which is representative of a real tweet as obtained from Twitter.
+
+ create dataverse feeds;
+ use dataverse feeds;
+
+ create type TwitterUser if not exists as open{
+ screen_name: string,
+ language: string,
+ friends_count: int32,
+ status_count: int32,
+ name: string,
+ followers_count: string
+ };
+ create type Tweet if not exists as open{
+ id: string,
+ user: TwitterUser,
+ latitude:double,
+ longitude:double,
+ created_at:string,
+ message_text:string
+ };
+
+
+Next we make use of the create feed AQL statement to define our example data feed.
+
+ create feed TwitterFeed if not exists using "push_twitter"
+ (("type-name"="Tweet"),("location"="US"));
+
+Note that the create feed statement does not initiate the flow of data from Twitter into our AsterixDB instance. Instead, the create feed statement only results in registering the feed with AsterixDB. The flow of data along a feed is initiated when it is connected
+to a target dataset using the connect feed statement (which we shall revisit later).
+
+ ####Preprocessing Collected Data####
+A feed definition may optionally include the specification of a
+user-defined function that is to be applied to each feed record prior
+to persistence. Examples of pre-processing might include adding
+attributes, filtering out records, sampling, sentiment analysis, feature
+extraction, etc. The pre-processing is expressed as a userdefined
+function (UDF) that can be defined in AQL or in a programming
+language like Java. An AQL UDF is a good fit when
+pre-processing a record requires the result of a query (join or aggregate)
+over data contained in AsterixDB datasets. More sophisticated
+processing such as sentiment analysis of text is better handled
+by providing a Java UDF. A Java UDF has an initialization phase
+that allows the UDF to access any resources it may need to initialize
+itself prior to being used in a data flow. It is assumed by the
+AsterixDB compiler to be stateless and thus usable as an embarassingly
+parallel black box. In constrast, the AsterixDB compiler can
+reason about an AQL UDF and involve the use of indexes during
+its invocation.
+
+The tweets collected by the Twitter Adaptor (push_twiiter) (Figure 4) conform
+to the Tweet datatype (defined earlier). We consider an example transformation of a raw tweet into its lightweight version - ProcessedTweet - which is defined next.
+
+ create type ProcessedTweet if not exists as open {
+ id: string,
+ user_name:string,
+ location:point,
+ created_at:string,
+ message_text:string,
+ country: string,
+ topics: [string]
+ };
+
+
+The processing required in transforming a collected tweet to its lighter version (of type ProcessedTweet) involves extracting the topicsor hash-tags (if any) in a tweet
+and collecting them in the referred-topics attribute for the tweet.
+Additionally, the latitude and longitude values (doubles) are combined into the spatial point type. Note that spatial data types are considered as first class citizens that come with the support for creating indexes. Next we show a revised version of our example TwitterFeed that involves the use of a UDF. We assume that the UDF that contains the transformation logic into a ProcessedTweet is avaialable as a Java UDF inside an AsterixDB library named 'testlib'. We defer the writing of a Java UDF and its installation as part of an AsterixDB library to a later section of this document.
+
+ create feed ProcessedTwitterFeed if not exists
+ using "push_twitter"
+ (("type-name"="Tweet"),("location"="US"));
+ apply function testlib#processRawTweet;
+
+Note that a feed adaptor and a UDF act as pluggable components. These
+contribute towards providing a generic ‘plug-and-play‘ model where
+custom implementations can be provided to cater to specific requirements.
+
+####Building a Cascade Network of Feeds####
+Multiple high-level applications may wish to consume the data
+ingested from a data feed. Each such application might perceive the
+feed in a different way and require the arriving data to be processed
+and/or persisted differently. Building a separate flow of data from
+the external source for each application is wasteful of resources as
+the pre-processing or transformations required by each application
+might overlap and could be done together in an incremental fashion
+to avoid redundancy. A single flow of data from the external source
+could provide data for multiple applications. To achieve this, we
+introduce the notion of primary and secondary feeds in AsterixDB.
+
+A feed in AsterixDB is considered to be a primary feed if it gets
+its data from an external data source. The records contained in a
+feed (subsequent to any pre-processing) are directed to a designated
+AsterixDB dataset. Alternatively or additionally, these records can
+be used to derive other feeds known as secondary feeds. A secondary
+feed is similar to its parent feed in every other aspect; it can
+have an associated UDF to allow for any subsequent processing,
+can be persisted into a dataset, and/or can be made to derive other
+secondary feeds to form a cascade network. A primary feed and a
+dependent secondary feed form a hierarchy. As an example, we next show an
+example AQL statement that redefines the previous feed—
+ProcessedTwitterFeed in terms of their
+respective parent feed (TwitterFeed).
+
+ create secondary feed ProcessedTwitterFeed from feed TwitterFeed
+ apply function testlib#addFeatures;
+
+
+####Lifecycle of a Feed####
+A feed is a logical artifact that is brought to life (i.e. its data flow
+is initiated) only when it is connected to a dataset using the connect
+feed AQL statement (Figure 7). Subsequent to a connect feed
+statement, the feed is said to be in the connected state. Multiple
+feeds can simultaneously be connected to a dataset such that the
+contents of the dataset represent the union of the connected feeds.
+In a supported but unlikely scenario, one feed may also be simultaneously
+connected to different target datasets. Note that connecting
+a secondary feed does not require the parent feed (or any ancestor
+feed) to be in the connected state; the order in which feeds are connected
+to their respective datasets is not important. Furthermore,
+additional (secondary) feeds can be added to an existing hierarchy
+and connected to a dataset at any time without impeding/interrupting
+the flow of data along a connected ancestor feed.
+
+ connect feed ProcessedTwitterFeed to
+ dataset ProcessedTweets ;
+
+ disconnect feed ProcessedTwitterFeed from
+ dataset ProcessedTweets ;
+
+The connect feed statement above directs AsterixDB to persist
+the ProcessedTwitterFeed feed in the ProcessedTweets dataset.
+If it is required (by the high-level application) to also retain the raw
+tweets obtained from Twitter, the end user may additionally choose
+to connect TwitterFeed to a (different) dataset. Having a set of primary
+and secondary feeds offers the flexibility to do so. Let us
+assume that the application needs to persist TwitterFeed and that,
+to do so, the end user makes another use of the connect feed statement.
+A logical view of the continuous flow of data established by
+connecting the feeds to their respective target datasets is shown in
+Figure 8.
+
+The flow of data from a feed into a dataset can be terminated
+explicitly by use of the disconnect feed statement.
+Disconnecting a feed from a particular dataset does not interrupt
+the flow of data from the feed to any other dataset(s), nor does it
+impact other connected feeds in the lineage.
+
+####Policies for Feed Ingestion####
+Multiple feeds may be concurrently operational on an AsterixDB
+cluster, each competing for resources (CPU cycles, network bandwidth,
+disk IO) to maintain pace with their respective data sources.
+A data management system must be able to manage a set of concurrent
+feeds and make dynamic decisions related to the allocation of
+resources, resolving resource bottlenecks and the handling of failures.
+Each feed has its own set of constraints, influenced largely
+by the nature of its data source and the application(s) that intend
+to consume and process the ingested data. Consider an application
+that intends to discover the trending topics on Twitter by analyzing
+the ProcessedTwitterFeed feed. Losing a few tweets may be
+acceptable. In contrast, when ingesting from a data source that
+provides a click-stream of ad clicks, losing data would translate to
+a loss of revenue for an application that tracks revenue by charging
+advertisers per click.
+
+AsterixDB allows a data feed to have an associated ingestion
+policy that is expressed as a collection of parameters and associated
+values. An ingestion policy dictates the runtime behavior of
+the feed in response to resource bottlenecks and failures. AsterixDB provides
+a list of policy parameters (Table 1) that help customize the
+system’s runtime behavior when handling excess records. AsterixDB
+provides a set of built-in policies, each constructed by setting
+appropriate value(s) for the policy parameter(s) from the table below.
+
+Policy Parameter | Description | Default Value
+------------------|------------|---------------|
+excess.records.spill | Set to true if records that cannot be processed by an operator for lack of resources (referred to as excess records hereafter) should be persisted to the local disk for deferred processing. | false |
+excess.records.discard | Set to true if excess records should be discarded. | false |
+excess.records.throttle | Set to true if rate of arrival of records is required to be reduced in an adaptive manner to prevent having any excess records. | false |
+excess.records.elastic | Set to true if the system should attempt to resolve resource bottlenecks by re-structuring and/or rescheduling the feed ingestion pipeline. | false |
+recover.soft.failure | Set to true if the feed must attempt to survive any runtime exception. A false value permits an early termination of a feed in such an event. | true |
+recover.hard.failure | Set to true if the feed must attempt to survive a hardware failures (loss of AsterixDB node(s)). A false value permits the early termination of a feed in the event of a hardware failure | |
+
+Note that the end user may choose to form a custom policy. E.g.
+it is possible in AsterixDB to create a custom policy that spills excess
+records to disk and subsequently resorts to throttling if the
+spillage crosses a configured threshold. In all cases, the desired
+ingestion policy is specified as part of the connect feed statement
+(Figure 9) or else the ‘Basic’ policy will be chosen as the default.
+It is worth noting that a feed can be connected to a dataset at any
+time, which is independent from other related feeds in the hierarchy.
+
+
+ connect feed TwitterFeed to dataset Tweets
+ using policy Basic ;
+
+
+####Writing an External UDF####
+A Java UDF in AsterixDB is required to implement an prescribe interface. We shall next write a basic UDF that extracts the hashtags contained in the tweet's text and appends each into an unordered list. The list is added as an additional attribute to the tweet to form the augment version - ProcessedTweet.
+
+ package edu.uci.ics.asterix.external.library;
+
+ import edu.uci.ics.asterix.external.library.java.JObjects.JRecord;
+ import edu.uci.ics.asterix.external.library.java.JObjects.JString;
+ import edu.uci.ics.asterix.external.library.java.JObjects.JUnorderedList;
+ import edu.uci.ics.asterix.external.library.java.JTypeTag;
+
+ public class HashTagsFunction implements IExternalScalarFunction {
+
+ private JUnorderedList list = null;
+
+ @Override
+ public void initialize(IFunctionHelper functionHelper) {
+ list = new JUnorderedList(functionHelper.getObject(JTypeTag.STRING));
+ }
+
+ @Override
+ public void deinitialize() {
+ }
+
+ @Override
+ public void evaluate(IFunctionHelper functionHelper) throws Exception {
+ list.clear();
+ JRecord inputRecord = (JRecord) functionHelper.getArgument(0);
+ JString text = (JString) inputRecord.getValueByName("message_text");
+
+ // extraction of hashtags
+ String[] tokens = text.getValue().split(" ");
+ for (String tk : tokens) {
+ if (tk.startsWith("#")) {
+ JString newField = (JString) functionHelper.getObject(JTypeTag.STRING);
+ newField.setValue(tk);
+ list.add(newField);
+ }
+ }
+
+ // forming the return value - an augmented tweet with an additional attribute - topics
+ JRecord result = (JRecord) functionHelper.getResultObject();
+ result.setField("tweetid", inputRecord.getFields()[0]);
+ result.setField("user", inputRecord.getFields()[1]);
+ result.setField("location_lat", inputRecord.getFields()[2]);
+ result.setField("location_long", inputRecord.getFields()[3]);
+ result.setField("send_time", inputRecord.getFields()[4]);
+ result.setField("message_text", inputRecord.getFields()[5]);
+ result.setField("topics", list);
+
+ functionHelper.setResult(result);
+ }
+
+
+
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/pom.xml
----------------------------------------------------------------------
diff --git a/asterix-external-data/pom.xml b/asterix-external-data/pom.xml
index 0b045f1..7d64b61 100644
--- a/asterix-external-data/pom.xml
+++ b/asterix-external-data/pom.xml
@@ -136,9 +136,14 @@
<dependency>
<groupId>org.twitter4j</groupId>
<artifactId>twitter4j-core</artifactId>
- <version>2.2.3</version>
+ <version>[3.0,)</version>
</dependency>
<dependency>
+ <groupId>org.twitter4j</groupId>
+ <artifactId>twitter4j-stream</artifactId>
+ <version>4.0.2</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<type>jar</type>
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/CNNFeedAdapterFactory.java
----------------------------------------------------------------------
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 fa66715..e4f118f 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
@@ -19,12 +19,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import edu.uci.ics.asterix.common.feeds.FeedPolicyAccessor;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
+import edu.uci.ics.asterix.common.feeds.api.IIntakeProgressTracker;
import edu.uci.ics.asterix.external.dataset.adapter.RSSFeedAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedAdapterFactory;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapterFactory;
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;
@@ -32,7 +32,7 @@ import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* A factory class for creating the @see {CNNFeedAdapter}.
*/
-public class CNNFeedAdapterFactory implements ITypedAdapterFactory {
+public class CNNFeedAdapterFactory implements IFeedAdapterFactory {
private static final long serialVersionUID = 1L;
private Map<String, String> configuration;
@@ -40,6 +40,7 @@ public class CNNFeedAdapterFactory implements ITypedAdapterFactory {
private List<String> feedURLs = new ArrayList<String>();
private static Map<String, String> topicFeeds = new HashMap<String, String>();
private ARecordType recordType;
+ private FeedPolicyAccessor policyAccessor;
public static final String KEY_RSS_URL = "topic";
public static final String KEY_INTERVAL = "interval";
@@ -91,22 +92,14 @@ public class CNNFeedAdapterFactory implements ITypedAdapterFactory {
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.TYPED;
- }
-
- @Override
- public void configure(Map<String, String> configuration) throws Exception {
+ public void configure(Map<String, String> configuration, ARecordType outputType) 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);
- recordType = new ARecordType("FeedRecordType", new String[] { "id", "title", "description", "link" },
- new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING },
- false);
-
+ this.recordType = outputType;
}
private void initializeFeedURLs(String rssURLProperty) {
@@ -147,4 +140,14 @@ public class CNNFeedAdapterFactory implements ITypedAdapterFactory {
return recordType;
}
+ @Override
+ public boolean isRecordTrackingEnabled() {
+ return false;
+ }
+
+ @Override
+ public IIntakeProgressTracker createIntakeProgressTracker() {
+ return null;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSAdapterFactory.java
----------------------------------------------------------------------
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 dbf3fd7..ee481de 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
@@ -31,15 +31,17 @@ import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import edu.uci.ics.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.HDFSAdapter;
import edu.uci.ics.asterix.external.indexing.dataflow.HDFSObjectTupleParserFactory;
import edu.uci.ics.asterix.metadata.entities.ExternalFile;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericAdapterFactory;
+import edu.uci.ics.asterix.metadata.external.IAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
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.asterix.runtime.operators.file.AsterixTupleParserFactory;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory.InputDataFormat;
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;
@@ -53,7 +55,7 @@ import edu.uci.ics.hyracks.hdfs.scheduler.Scheduler;
/**
* A factory class for creating an instance of HDFSAdapter
*/
-public class HDFSAdapterFactory extends StreamBasedAdapterFactory implements IGenericAdapterFactory {
+public class HDFSAdapterFactory extends StreamBasedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String HDFS_ADAPTER_NAME = "hdfs";
@@ -80,6 +82,7 @@ public class HDFSAdapterFactory extends StreamBasedAdapterFactory implements IGe
private boolean configured = false;
public static Scheduler hdfsScheduler;
private static boolean initialized = false;
+ protected List<ExternalFile> files;
private static Scheduler initializeHDFSScheduler() {
ICCContext ccContext = AsterixAppContextInfo.getInstance().getCCApplicationContext().getCCContext();
@@ -137,11 +140,6 @@ public class HDFSAdapterFactory extends StreamBasedAdapterFactory implements IGe
}
@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");
@@ -203,25 +201,41 @@ public class HDFSAdapterFactory extends StreamBasedAdapterFactory implements IGe
return new AlgebricksAbsolutePartitionConstraint(cluster);
}
+ @Override
+ public ARecordType getAdapterOutputType() {
+ return (ARecordType) atype;
+ }
+
+ @Override
+ public InputDataFormat getInputDataFormat() {
+ return InputDataFormat.UNKNOWN;
+ }
+
/*
* This method is overridden to do the following:
* if data is text data (adm or delimited text), it will use a text tuple parser,
* otherwise it will use hdfs record object parser
*/
protected void configureFormat(IAType sourceDatatype) throws Exception {
- 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) sourceDatatype, false);
- } else if (FORMAT_ADM.equalsIgnoreCase((String) configuration.get(KEY_FORMAT))) {
- parserFactory = getADMDataTupleParserFactory((ARecordType) sourceDatatype, false);
- } else if (FORMAT_BINARY.equalsIgnoreCase((String) configuration.get(KEY_FORMAT))) {
- parserFactory = new HDFSObjectTupleParserFactory((ARecordType) atype, this, configuration);
- } else {
- throw new IllegalArgumentException(" format " + configuration.get(KEY_FORMAT) + " not supported");
- }
- }
+ String specifiedFormat = (String) configuration.get(AsterixTupleParserFactory.KEY_FORMAT);
+ if (specifiedFormat == null) {
+ throw new IllegalArgumentException(" Unspecified data format");
+ }
+
+ if(AsterixTupleParserFactory.FORMAT_BINARY.equalsIgnoreCase(specifiedFormat)){
+ parserFactory = new HDFSObjectTupleParserFactory((ARecordType) atype, this, configuration);
+ } else {
+ InputDataFormat inputFormat = InputDataFormat.UNKNOWN;
+ if (AsterixTupleParserFactory.FORMAT_DELIMITED_TEXT.equalsIgnoreCase(specifiedFormat)) {
+ inputFormat = InputDataFormat.DELIMITED;
+ } else if (AsterixTupleParserFactory.FORMAT_ADM.equalsIgnoreCase(specifiedFormat)) {
+ inputFormat = InputDataFormat.ADM;
+ }
+ parserFactory = new AsterixTupleParserFactory(configuration, (ARecordType) sourceDatatype
+ , inputFormat);
+ }
+
+ }
/**
* Instead of creating the split using the input format, we do it manually
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSIndexingAdapterFactory.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSIndexingAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSIndexingAdapterFactory.java
index 2a4836c..37b8050 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSIndexingAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HDFSIndexingAdapterFactory.java
@@ -16,22 +16,24 @@ package edu.uci.ics.asterix.external.adapter.factory;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.HDFSIndexingAdapter;
import edu.uci.ics.asterix.external.indexing.dataflow.HDFSIndexingParserFactory;
import edu.uci.ics.asterix.external.indexing.dataflow.IndexingScheduler;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.ATypeTag;
import edu.uci.ics.asterix.om.types.AUnionType;
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.asterix.runtime.operators.file.AsterixTupleParserFactory;
import edu.uci.ics.asterix.runtime.operators.file.DelimitedDataParser;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
@@ -39,7 +41,12 @@ import edu.uci.ics.hyracks.algebricks.common.exceptions.NotImplementedException;
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.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;
import edu.uci.ics.hyracks.hdfs.dataflow.ConfFactory;
import edu.uci.ics.hyracks.hdfs.dataflow.InputSplitsFactory;
@@ -83,11 +90,6 @@ public class HDFSIndexingAdapterFactory extends HDFSAdapterFactory {
}
@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");
@@ -104,7 +106,8 @@ public class HDFSIndexingAdapterFactory extends HDFSAdapterFactory {
((HDFSIndexingParserFactory) parserFactory).setArguments(configuration);
HDFSIndexingAdapter hdfsIndexingAdapter = new HDFSIndexingAdapter(atype, readSchedule, executed, inputSplits,
conf, clusterLocations, files, parserFactory, ctx, nodeName,
- (String) configuration.get(HDFSAdapterFactory.KEY_INPUT_FORMAT), (String) configuration.get(KEY_FORMAT));
+ (String) configuration.get(HDFSAdapterFactory.KEY_INPUT_FORMAT),
+ (String) configuration.get(AsterixTupleParserFactory.KEY_FORMAT));
return hdfsIndexingAdapter;
}
@@ -131,12 +134,13 @@ public class HDFSIndexingAdapterFactory extends HDFSAdapterFactory {
protected void configureFormat(IAType sourceDatatype) throws Exception {
- char delimiter = StreamBasedAdapterFactory.getDelimiter(configuration);
- char quote = StreamBasedAdapterFactory.getQuote(configuration, delimiter);
+ char delimiter = AsterixTupleParserFactory.getDelimiter(configuration);
+ char quote = AsterixTupleParserFactory.getQuote(configuration, delimiter);
parserFactory = new HDFSIndexingParserFactory((ARecordType) atype,
- configuration.get(HDFSAdapterFactory.KEY_INPUT_FORMAT), configuration.get(KEY_FORMAT), delimiter,
- quote, configuration.get(HDFSAdapterFactory.KEY_PARSER));
+ configuration.get(HDFSAdapterFactory.KEY_INPUT_FORMAT),
+ configuration.get(AsterixTupleParserFactory.KEY_FORMAT), delimiter, quote,
+ configuration.get(HDFSAdapterFactory.KEY_PARSER));
}
/**
@@ -165,7 +169,7 @@ public class HDFSIndexingAdapterFactory extends HDFSAdapterFactory {
if (tag == null) {
throw new NotImplementedException("Failed to get the type information for field " + i + ".");
}
- IValueParserFactory vpf = typeToValueParserFactMap.get(tag);
+ IValueParserFactory vpf = valueParserFactoryMap.get(tag);
if (vpf == null) {
throw new NotImplementedException("No value parser factory for delimited fields of type " + tag);
}
@@ -190,4 +194,16 @@ public class HDFSIndexingAdapterFactory extends HDFSAdapterFactory {
cluster = locs.toArray(cluster);
return new AlgebricksAbsolutePartitionConstraint(cluster);
}
+
+ private static Map<ATypeTag, IValueParserFactory> valueParserFactoryMap = initializeValueParserFactoryMap();
+
+ private static Map<ATypeTag, IValueParserFactory> initializeValueParserFactoryMap() {
+ Map<ATypeTag, IValueParserFactory> m = new HashMap<ATypeTag, IValueParserFactory>();
+ m.put(ATypeTag.INT32, IntegerParserFactory.INSTANCE);
+ m.put(ATypeTag.FLOAT, FloatParserFactory.INSTANCE);
+ m.put(ATypeTag.DOUBLE, DoubleParserFactory.INSTANCE);
+ m.put(ATypeTag.INT64, LongParserFactory.INSTANCE);
+ m.put(ATypeTag.STRING, UTF8StringParserFactory.INSTANCE);
+ return m;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/HiveAdapterFactory.java
----------------------------------------------------------------------
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 ab59241..c5ebe67 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
@@ -17,20 +17,22 @@ package edu.uci.ics.asterix.external.adapter.factory;
import java.util.List;
import java.util.Map;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.HDFSAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.HiveAdapter;
import edu.uci.ics.asterix.metadata.entities.ExternalFile;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericAdapterFactory;
+import edu.uci.ics.asterix.metadata.external.IAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory.InputDataFormat;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
/**
* A factory class for creating an instance of HiveAdapter
*/
-public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IGenericAdapterFactory {
+public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String HIVE_DATABASE = "database";
@@ -61,11 +63,7 @@ public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IGe
return "hive";
}
- @Override
- public AdapterType getAdapterType() {
- return AdapterType.GENERIC;
- }
-
+
@Override
public SupportedOperation getSupportedOperations() {
return SupportedOperation.READ;
@@ -76,6 +74,7 @@ public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IGe
if (!configured) {
populateConfiguration(configuration);
hdfsAdapterFactory.configure(configuration, outputType);
+ this.atype = (IAType) outputType;
}
}
@@ -90,8 +89,8 @@ public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IGe
+ configuration.get(HIVE_TABLE);
}
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");
+ if (!configuration.get(AsterixTupleParserFactory.KEY_FORMAT).equals(AsterixTupleParserFactory.FORMAT_DELIMITED_TEXT)) {
+ throw new IllegalArgumentException("format" + configuration.get(AsterixTupleParserFactory.KEY_FORMAT) + " is not supported");
}
if (!(configuration.get(HDFSAdapterFactory.KEY_INPUT_FORMAT).equals(HDFSAdapterFactory.INPUT_FORMAT_TEXT) || configuration
@@ -106,9 +105,19 @@ public class HiveAdapterFactory extends StreamBasedAdapterFactory implements IGe
return hdfsAdapterFactory.getPartitionConstraint();
}
+
+ @Override
+ public ARecordType getAdapterOutputType() {
+ return (ARecordType) atype;
+ }
+
@Override
+ public InputDataFormat getInputDataFormat() {
+ return InputDataFormat.UNKNOWN;
+ }
+
public void setFiles(List<ExternalFile> files) {
- this.files = files;
+ hdfsAdapterFactory.setFiles(files);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/NCFileSystemAdapterFactory.java
----------------------------------------------------------------------
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 110a965..446199e 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
@@ -20,15 +20,17 @@ import java.util.Map;
import java.util.logging.Level;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
import edu.uci.ics.asterix.external.dataset.adapter.NCFileSystemAdapter;
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.entities.ExternalFile;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IGenericAdapterFactory;
+import edu.uci.ics.asterix.metadata.external.IAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.IAType;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory.InputDataFormat;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -41,7 +43,7 @@ import edu.uci.ics.hyracks.dataflow.std.file.FileSplit;
* NCFileSystemAdapter reads external data residing on the local file system of
* an NC.
*/
-public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implements IGenericAdapterFactory {
+public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String NC_FILE_SYSTEM_ADAPTER_NAME = "localfs";
@@ -50,6 +52,8 @@ public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implem
private IAType sourceDatatype;
private FileSplit[] fileSplits;
+ private ARecordType outputType;
+
@Override
public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
@@ -62,10 +66,6 @@ public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implem
return NC_FILE_SYSTEM_ADAPTER_NAME;
}
- @Override
- public AdapterType getAdapterType() {
- return AdapterType.GENERIC;
- }
@Override
public SupportedOperation getSupportedOperations() {
@@ -75,7 +75,8 @@ public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implem
@Override
public void configure(Map<String, String> configuration, ARecordType outputType) throws Exception {
this.configuration = configuration;
- String[] splits = ((String) configuration.get(KEY_PATH)).split(",");
+ this.outputType = outputType;
+ String[] splits = ((String) configuration.get(AsterixTupleParserFactory.KEY_PATH)).split(",");
IAType sourceDatatype = (IAType) outputType;
configureFileSplits(splits);
configureFormat(sourceDatatype);
@@ -127,7 +128,7 @@ public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implem
private static INodeResolver initializeNodeResolver() {
INodeResolver nodeResolver = null;
- String configuredNodeResolverFactory = System.getProperty(NODE_RESOLVER_FACTORY_PROPERTY);
+ String configuredNodeResolverFactory = System.getProperty(AsterixTupleParserFactory.NODE_RESOLVER_FACTORY_PROPERTY);
if (configuredNodeResolverFactory != null) {
try {
nodeResolver = ((INodeResolverFactory) (Class.forName(configuredNodeResolverFactory).newInstance()))
@@ -145,8 +146,17 @@ public class NCFileSystemAdapterFactory extends StreamBasedAdapterFactory implem
}
return nodeResolver;
}
-
+
@Override
+ public ARecordType getAdapterOutputType() {
+ return outputType;
+ }
+
+ @Override
+ public InputDataFormat getInputDataFormat() {
+ return InputDataFormat.UNKNOWN;
+ }
+
public void setFiles(List<ExternalFile> files) throws AlgebricksException {
throw new AlgebricksException("can't set files for this Adapter");
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedAzureTwitterAdapterFactory.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedAzureTwitterAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedAzureTwitterAdapterFactory.java
index b4dbe13..298a443 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedAzureTwitterAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedAzureTwitterAdapterFactory.java
@@ -3,12 +3,14 @@ package edu.uci.ics.asterix.external.adapter.factory;
import java.util.Map;
import edu.uci.ics.asterix.common.exceptions.AsterixException;
+import edu.uci.ics.asterix.common.feeds.FeedPolicyAccessor;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
+import edu.uci.ics.asterix.common.feeds.api.IIntakeProgressTracker;
import edu.uci.ics.asterix.external.dataset.adapter.PullBasedAzureTwitterAdapter;
import edu.uci.ics.asterix.metadata.MetadataManager;
import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
import edu.uci.ics.asterix.metadata.entities.Datatype;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedAdapterFactory;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
import edu.uci.ics.asterix.om.types.ATypeTag;
import edu.uci.ics.asterix.om.types.IAType;
@@ -16,7 +18,7 @@ import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksAbsoluteParti
import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory {
+public class PullBasedAzureTwitterAdapterFactory implements IFeedAdapterFactory {
private static final long serialVersionUID = 1L;
@@ -27,13 +29,14 @@ public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory
private static final String ACCOUNT_NAME_KEY = "account-name";
private static final String ACCOUNT_KEY_KEY = "account-key";
- private ARecordType recordType;
+ private ARecordType outputType;
private Map<String, String> configuration;
private String tableName;
private String azureAccountName;
private String azureAccountKey;
private String[] locations;
private String[] partitions;
+ private FeedPolicyAccessor ingestionPolicy;
@Override
public SupportedOperation getSupportedOperations() {
@@ -46,11 +49,6 @@ public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.TYPED;
- }
-
- @Override
public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
String locationsStr = configuration.get(INGESTOR_LOCATIONS_KEY);
if (locationsStr == null) {
@@ -63,17 +61,18 @@ public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory
@Override
public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
return new PullBasedAzureTwitterAdapter(azureAccountName, azureAccountKey, tableName, partitions,
- configuration, ctx, recordType);
+ configuration, ctx, outputType);
}
@Override
public ARecordType getAdapterOutputType() {
- return recordType;
+ return outputType;
}
@Override
- public void configure(Map<String, String> configuration) throws Exception {
+ public void configure(Map<String, String> configuration, ARecordType outputType) throws Exception {
this.configuration = configuration;
+ this.outputType = outputType;
tableName = configuration.get(TABLE_NAME_KEY);
if (tableName == null) {
@@ -125,7 +124,7 @@ public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory
if (type.getTypeTag() != ATypeTag.RECORD) {
throw new IllegalStateException();
}
- recordType = (ARecordType) t.getDatatype();
+ outputType = (ARecordType) t.getDatatype();
MetadataManager.INSTANCE.commitTransaction(ctx);
} catch (Exception e) {
if (ctx != null) {
@@ -136,4 +135,21 @@ public class PullBasedAzureTwitterAdapterFactory implements ITypedAdapterFactory
MetadataManager.INSTANCE.releaseReadLatch();
}
}
+
+ @Override
+ public boolean isRecordTrackingEnabled() {
+ return false;
+ }
+
+ @Override
+ public IIntakeProgressTracker createIntakeProgressTracker() {
+ return null;
+ }
+
+ public FeedPolicyAccessor getIngestionPolicy() {
+ return ingestionPolicy;
+ }
+
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PullBasedTwitterAdapterFactory.java
----------------------------------------------------------------------
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 6058bd2..c43ca05 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
@@ -15,13 +15,18 @@
package edu.uci.ics.asterix.external.adapter.factory;
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.common.feeds.FeedPolicyAccessor;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
+import edu.uci.ics.asterix.common.feeds.api.IIntakeProgressTracker;
import edu.uci.ics.asterix.external.dataset.adapter.PullBasedTwitterAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedAdapterFactory;
+import edu.uci.ics.asterix.external.util.TwitterUtil;
+import edu.uci.ics.asterix.external.util.TwitterUtil.SearchAPIConstants;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapterFactory;
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;
@@ -31,29 +36,22 @@ import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
* This adapter provides the functionality of fetching tweets from Twitter service
* via pull-based Twitter API.
*/
-public class PullBasedTwitterAdapterFactory implements ITypedAdapterFactory {
+public class PullBasedTwitterAdapterFactory implements IFeedAdapterFactory {
private static final long serialVersionUID = 1L;
+ private static final Logger LOGGER = Logger.getLogger(PullBasedTwitterAdapterFactory.class.getName());
+
public static final String PULL_BASED_TWITTER_ADAPTER_NAME = "pull_twitter";
+ private static final String DEFAULT_INTERVAL = "10"; // 10 seconds
+ private static final int INTAKE_CARDINALITY = 1; // degree of parallelism at intake stage
+
+ private ARecordType outputType;
+
private Map<String, String> configuration;
- private static ARecordType recordType = initOutputType();
-
- private static ARecordType initOutputType() {
- ARecordType recordType = null;
- String[] fieldNames = { "id", "username", "location", "text", "timestamp" };
- IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
- BuiltinType.ASTRING };
- try {
- recordType = new ARecordType("TweetType", fieldNames, fieldTypes, false);
- } catch (Exception e) {
- throw new IllegalStateException("Unable to create adapter output type");
- }
- return recordType;
- }
@Override
public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
- return new PullBasedTwitterAdapter(configuration, recordType, ctx);
+ return new PullBasedTwitterAdapter(configuration, outputType, ctx);
}
@Override
@@ -62,28 +60,57 @@ public class PullBasedTwitterAdapterFactory implements ITypedAdapterFactory {
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.TYPED;
- }
-
- @Override
public SupportedOperation getSupportedOperations() {
return SupportedOperation.READ;
}
@Override
- public void configure(Map<String, String> configuration) throws Exception {
+ public void configure(Map<String, String> configuration, ARecordType outputType) throws Exception {
+ this.outputType = outputType;
this.configuration = configuration;
+ TwitterUtil.initializeConfigurationWithAuthInfo(configuration);
+
+ if (configuration.get(SearchAPIConstants.QUERY) == null) {
+ throw new AsterixException("parameter " + SearchAPIConstants.QUERY
+ + " not specified as part of adaptor configuration");
+ }
+
+ String interval = configuration.get(SearchAPIConstants.INTERVAL);
+ if (interval != null) {
+ try {
+ Integer.parseInt(interval);
+ } catch (NumberFormatException nfe) {
+ throw new IllegalArgumentException("parameter " + SearchAPIConstants.INTERVAL
+ + " is defined incorrectly, expecting a number");
+ }
+ } else {
+ configuration.put(SearchAPIConstants.INTERVAL, DEFAULT_INTERVAL);
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning(" Parameter " + SearchAPIConstants.INTERVAL + " not defined, using default ("
+ + DEFAULT_INTERVAL + ")");
+ }
+ }
+
}
@Override
public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
- return new AlgebricksCountPartitionConstraint(1);
+ return new AlgebricksCountPartitionConstraint(INTAKE_CARDINALITY);
+ }
+
+ @Override
+ public boolean isRecordTrackingEnabled() {
+ return false;
+ }
+
+ @Override
+ public IIntakeProgressTracker createIntakeProgressTracker() {
+ return null;
}
@Override
public ARecordType getAdapterOutputType() {
- return recordType;
+ return outputType;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PushBasedTwitterAdapterFactory.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PushBasedTwitterAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PushBasedTwitterAdapterFactory.java
new file mode 100644
index 0000000..c947a6a
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/PushBasedTwitterAdapterFactory.java
@@ -0,0 +1,69 @@
+package edu.uci.ics.asterix.external.adapter.factory;
+
+import java.util.Map;
+
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
+import edu.uci.ics.asterix.common.feeds.api.IIntakeProgressTracker;
+import edu.uci.ics.asterix.external.dataset.adapter.PushBasedTwitterAdapter;
+import edu.uci.ics.asterix.external.util.TwitterUtil;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapterFactory;
+import edu.uci.ics.asterix.om.types.ARecordType;
+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;
+
+public class PushBasedTwitterAdapterFactory implements IFeedAdapterFactory {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String NAME = "push_twitter";
+
+ private ARecordType outputType;
+
+ private Map<String, String> configuration;
+
+ @Override
+ public SupportedOperation getSupportedOperations() {
+ return SupportedOperation.READ;
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public AlgebricksPartitionConstraint getPartitionConstraint() throws Exception {
+ return new AlgebricksCountPartitionConstraint(1);
+ }
+
+ @Override
+ public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
+ PushBasedTwitterAdapter twitterAdapter = new PushBasedTwitterAdapter(configuration, outputType, ctx);
+ return twitterAdapter;
+ }
+
+ @Override
+ public void configure(Map<String, String> configuration, ARecordType outputType) throws Exception {
+ this.outputType = outputType;
+ this.configuration = configuration;
+ TwitterUtil.initializeConfigurationWithAuthInfo(configuration);
+ }
+
+ @Override
+ public ARecordType getAdapterOutputType() {
+ return outputType;
+ }
+
+
+ @Override
+ public boolean isRecordTrackingEnabled() {
+ return false;
+ }
+
+ @Override
+ public IIntakeProgressTracker createIntakeProgressTracker() {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/RSSFeedAdapterFactory.java
----------------------------------------------------------------------
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 41f1d56..01ae2f2 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
@@ -18,12 +18,12 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import edu.uci.ics.asterix.common.feeds.FeedPolicyAccessor;
+import edu.uci.ics.asterix.common.feeds.api.IDatasourceAdapter;
+import edu.uci.ics.asterix.common.feeds.api.IIntakeProgressTracker;
import edu.uci.ics.asterix.external.dataset.adapter.RSSFeedAdapter;
-import edu.uci.ics.asterix.metadata.feeds.IDatasourceAdapter;
-import edu.uci.ics.asterix.metadata.feeds.ITypedAdapterFactory;
+import edu.uci.ics.asterix.metadata.feeds.IFeedAdapterFactory;
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;
@@ -32,7 +32,7 @@ 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 ITypedAdapterFactory {
+public class RSSFeedAdapterFactory implements IFeedAdapterFactory {
private static final long serialVersionUID = 1L;
public static final String RSS_FEED_ADAPTER_NAME = "rss_feed";
@@ -40,12 +40,13 @@ public class RSSFeedAdapterFactory implements ITypedAdapterFactory {
public static final String KEY_INTERVAL = "interval";
private Map<String, String> configuration;
- private ARecordType recordType;
+ private ARecordType outputType;
private List<String> feedURLs = new ArrayList<String>();
+ private FeedPolicyAccessor ingestionPolicy;
@Override
public IDatasourceAdapter createAdapter(IHyracksTaskContext ctx, int partition) throws Exception {
- RSSFeedAdapter rssFeedAdapter = new RSSFeedAdapter(configuration, recordType, ctx);
+ RSSFeedAdapter rssFeedAdapter = new RSSFeedAdapter(configuration, outputType, ctx);
return rssFeedAdapter;
}
@@ -55,28 +56,20 @@ public class RSSFeedAdapterFactory implements ITypedAdapterFactory {
}
@Override
- public AdapterType getAdapterType() {
- return AdapterType.TYPED;
- }
-
- @Override
public SupportedOperation getSupportedOperations() {
return SupportedOperation.READ;
}
@Override
- public void configure(Map<String, String> configuration) throws Exception {
+ public void configure(Map<String, String> configuration, ARecordType outputType) throws Exception {
this.configuration = configuration;
+ this.outputType = outputType;
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
@@ -98,7 +91,21 @@ public class RSSFeedAdapterFactory implements ITypedAdapterFactory {
@Override
public ARecordType getAdapterOutputType() {
- return recordType;
+ return outputType;
+ }
+
+ @Override
+ public boolean isRecordTrackingEnabled() {
+ return false;
+ }
+
+ @Override
+ public IIntakeProgressTracker createIntakeProgressTracker() {
+ return null;
+ }
+
+ public FeedPolicyAccessor getIngestionPolicy() {
+ return ingestionPolicy;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/StreamBasedAdapterFactory.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/StreamBasedAdapterFactory.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/StreamBasedAdapterFactory.java
index 0f340bc..7d45448 100644
--- a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/StreamBasedAdapterFactory.java
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/adapter/factory/StreamBasedAdapterFactory.java
@@ -14,30 +14,15 @@
*/
package edu.uci.ics.asterix.external.adapter.factory;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.logging.Logger;
-import edu.uci.ics.asterix.common.exceptions.AsterixException;
import edu.uci.ics.asterix.external.util.INodeResolver;
-import edu.uci.ics.asterix.metadata.entities.ExternalFile;
-import edu.uci.ics.asterix.metadata.feeds.ConditionalPushTupleParserFactory;
-import edu.uci.ics.asterix.metadata.feeds.IAdapterFactory;
+import edu.uci.ics.asterix.metadata.external.IAdapterFactory;
import edu.uci.ics.asterix.om.types.ARecordType;
-import edu.uci.ics.asterix.om.types.ATypeTag;
-import edu.uci.ics.asterix.om.types.AUnionType;
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.exceptions.NotImplementedException;
-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;
-import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory;
+import edu.uci.ics.asterix.runtime.operators.file.AsterixTupleParserFactory.InputDataFormat;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParserFactory;
public abstract class StreamBasedAdapterFactory implements IAdapterFactory {
@@ -45,146 +30,18 @@ public abstract class StreamBasedAdapterFactory implements IAdapterFactory {
private static final long serialVersionUID = 1L;
protected static final Logger LOGGER = Logger.getLogger(StreamBasedAdapterFactory.class.getName());
- protected Map<String, String> configuration;
protected static INodeResolver nodeResolver;
- public static final String KEY_FORMAT = "format";
- public static final String KEY_PARSER_FACTORY = "parser";
- public static final String KEY_DELIMITER = "delimiter";
- public static final String KEY_QUOTE = "quote";
- public static final String KEY_HEADER = "header";
- public static final String KEY_PATH = "path";
- public static final String KEY_SOURCE_DATATYPE = "output-type-name";
- // The length of a delimiter should be 1.
- public static final String DEFAULT_DELIMITER = ",";
- // A quote is used to enclose a string if it includes delimiter(s) in it.
- // The length of a quote should be 1.
- public static final String DEFAULT_QUOTE = "\"";
- public static final String FORMAT_DELIMITED_TEXT = "delimited-text";
- public static final String FORMAT_ADM = "adm";
- public static final String NODE_RESOLVER_FACTORY_PROPERTY = "node.Resolver";
- public static final String BATCH_SIZE = "batch-size";
- public static final String BATCH_INTERVAL = "batch-interval";
-
+ protected Map<String, String> configuration;
protected ITupleParserFactory parserFactory;
- protected ITupleParser parser;
-
- protected List<ExternalFile> files;
-
- 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 ITupleParserFactory getDelimitedDataTupleParserFactory(ARecordType recordType, boolean conditionalPush)
- throws AsterixException {
- int n = recordType.getFieldTypes().length;
- IValueParserFactory[] fieldParserFactories = new IValueParserFactory[n];
- for (int i = 0; i < n; i++) {
- ATypeTag tag = null;
- if (recordType.getFieldTypes()[i].getTypeTag() == ATypeTag.UNION) {
- List<IAType> unionTypes = ((AUnionType) recordType.getFieldTypes()[i]).getUnionList();
- if (unionTypes.size() != 2 && unionTypes.get(0).getTypeTag() != ATypeTag.NULL) {
- throw new NotImplementedException("Non-optional UNION type is not supported.");
- }
- tag = unionTypes.get(1).getTypeTag();
- } else {
- tag = recordType.getFieldTypes()[i].getTypeTag();
- }
- if (tag == null) {
- throw new NotImplementedException("Failed to get the type information for field " + i + ".");
- }
- IValueParserFactory vpf = typeToValueParserFactMap.get(tag);
- if (vpf == null) {
- throw new NotImplementedException("No value parser factory for delimited fields of type " + tag);
- }
- fieldParserFactories[i] = vpf;
- }
-
- char delimiter = getDelimiter(configuration);
- char quote = getQuote(configuration, delimiter);
- boolean hasHeader = getHasHeader(configuration);
-
- return conditionalPush ? new ConditionalPushTupleParserFactory(recordType, fieldParserFactories, delimiter,
- quote, hasHeader, configuration) : new NtDelimitedDataTupleParserFactory(recordType,
- fieldParserFactories, delimiter, quote, hasHeader);
- }
+
- protected ITupleParserFactory getADMDataTupleParserFactory(ARecordType recordType, boolean conditionalPush)
- throws AsterixException {
- try {
- return conditionalPush ? new ConditionalPushTupleParserFactory(recordType, configuration)
- : new AdmSchemafullRecordParserFactory(recordType);
- } catch (Exception e) {
- throw new AsterixException(e);
- }
-
- }
+ public abstract InputDataFormat getInputDataFormat();
protected void configureFormat(IAType sourceDatatype) throws Exception {
- String propValue = (String) configuration.get(BATCH_SIZE);
- int batchSize = propValue != null ? Integer.parseInt(propValue) : -1;
- propValue = configuration.get(BATCH_INTERVAL);
- long batchInterval = propValue != null ? Long.parseLong(propValue) : -1;
- boolean conditionalPush = batchSize > 0 || batchInterval > 0;
-
- String parserFactoryClassname = configuration.get(KEY_PARSER_FACTORY);
- if (parserFactoryClassname == null) {
- String specifiedFormat = configuration.get(KEY_FORMAT);
- if (specifiedFormat == null) {
- throw new IllegalArgumentException(" Unspecified data format");
- } else if (FORMAT_DELIMITED_TEXT.equalsIgnoreCase(specifiedFormat)) {
- parserFactory = getDelimitedDataTupleParserFactory((ARecordType) sourceDatatype, conditionalPush);
- } else if (FORMAT_ADM.equalsIgnoreCase((String) configuration.get(KEY_FORMAT))) {
- parserFactory = getADMDataTupleParserFactory((ARecordType) sourceDatatype, conditionalPush);
- } else {
- throw new IllegalArgumentException(" format " + configuration.get(KEY_FORMAT) + " not supported");
- }
- } else {
- parserFactory = (ITupleParserFactory) Class.forName(parserFactoryClassname).newInstance();
- }
-
- }
+ parserFactory = new AsterixTupleParserFactory(configuration, (ARecordType) sourceDatatype, getInputDataFormat());
- // Get a delimiter from the given configuration
- public static char getDelimiter(Map<String, String> configuration) throws AsterixException {
- String delimiterValue = configuration.get(KEY_DELIMITER);
- if (delimiterValue == null) {
- delimiterValue = DEFAULT_DELIMITER;
- } else if (delimiterValue.length() != 1) {
- throw new AsterixException("'" + delimiterValue
- + "' is not a valid delimiter. The length of a delimiter should be 1.");
- }
- return delimiterValue.charAt(0);
}
- // Get a quote from the given configuration when the delimiter is given
- // Need to pass delimiter to check whether they share the same character
- public static char getQuote(Map<String, String> configuration, char delimiter) throws AsterixException {
- String quoteValue = configuration.get(KEY_QUOTE);
- if (quoteValue == null) {
- quoteValue = DEFAULT_QUOTE;
- } else if (quoteValue.length() != 1) {
- throw new AsterixException("'" + quoteValue + "' is not a valid quote. The length of a quote should be 1.");
- }
-
- // Since delimiter (char type value) can't be null,
- // we only check whether delimiter and quote use the same character
- if (quoteValue.charAt(0) == delimiter) {
- throw new AsterixException("Quote '" + quoteValue + "' cannot be used with the delimiter '" + delimiter
- + "'. ");
- }
-
- return quoteValue.charAt(0);
- }
-
- // Get the header flag
- public static boolean getHasHeader(Map<String, String> configuration) {
- return Boolean.parseBoolean(configuration.get(KEY_HEADER));
- }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/ClientBasedFeedAdapter.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/ClientBasedFeedAdapter.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/ClientBasedFeedAdapter.java
new file mode 100644
index 0000000..4712e7c
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/ClientBasedFeedAdapter.java
@@ -0,0 +1,150 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.asterix.external.dataset.adapter;
+
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import edu.uci.ics.asterix.common.feeds.api.IFeedAdapter;
+import edu.uci.ics.asterix.common.parse.ITupleForwardPolicy;
+import edu.uci.ics.asterix.external.dataset.adapter.IFeedClient.InflowState;
+import edu.uci.ics.asterix.metadata.feeds.FeedPolicyEnforcer;
+import edu.uci.ics.asterix.om.types.ARecordType;
+import edu.uci.ics.hyracks.api.comm.IFrame;
+import edu.uci.ics.hyracks.api.comm.IFrameWriter;
+import edu.uci.ics.hyracks.api.comm.VSizeFrame;
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+
+/**
+ * Acts as an abstract class for all pull-based external data adapters. Captures
+ * the common logic for obtaining bytes from an external source and packing them
+ * into frames as tuples.
+ */
+public abstract class ClientBasedFeedAdapter implements IFeedAdapter {
+
+ private static final long serialVersionUID = 1L;
+ private static final Logger LOGGER = Logger.getLogger(ClientBasedFeedAdapter.class.getName());
+ private static final int timeout = 5; // seconds
+
+ protected ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(1);
+ protected IFeedClient pullBasedFeedClient;
+ protected ARecordType adapterOutputType;
+ protected boolean continueIngestion = true;
+ protected Map<String, String> configuration;
+
+ private FrameTupleAppender appender;
+ private IFrame frame;
+ private long tupleCount = 0;
+ private final IHyracksTaskContext ctx;
+ private int frameTupleCount = 0;
+
+ protected FeedPolicyEnforcer policyEnforcer;
+
+ public FeedPolicyEnforcer getPolicyEnforcer() {
+ return policyEnforcer;
+ }
+
+ public void setFeedPolicyEnforcer(FeedPolicyEnforcer policyEnforcer) {
+ this.policyEnforcer = policyEnforcer;
+ }
+
+ public abstract IFeedClient getFeedClient(int partition) throws Exception;
+
+ public abstract ITupleForwardPolicy getTupleParserPolicy();
+
+ public ClientBasedFeedAdapter(Map<String, String> configuration, IHyracksTaskContext ctx) {
+ this.ctx = ctx;
+ this.configuration = configuration;
+ }
+
+ public long getIngestedRecordsCount() {
+ return tupleCount;
+ }
+
+ @Override
+ public void start(int partition, IFrameWriter writer) throws Exception {
+ appender = new FrameTupleAppender();
+ frame = new VSizeFrame(ctx);
+ appender.reset(frame, true);
+ ITupleForwardPolicy policy = getTupleParserPolicy();
+ policy.configure(configuration);
+ pullBasedFeedClient = getFeedClient(partition);
+ InflowState inflowState = null;
+ policy.initialize(ctx, writer);
+ while (continueIngestion) {
+ tupleBuilder.reset();
+ try {
+ inflowState = pullBasedFeedClient.nextTuple(tupleBuilder.getDataOutput(), timeout);
+ switch (inflowState) {
+ case DATA_AVAILABLE:
+ tupleBuilder.addFieldEndOffset();
+ policy.addTuple(tupleBuilder);
+ frameTupleCount++;
+ break;
+ case NO_MORE_DATA:
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Reached end of feed");
+ }
+ policy.close();
+ tupleCount += frameTupleCount;
+ frameTupleCount = 0;
+ continueIngestion = false;
+ break;
+ case DATA_NOT_AVAILABLE:
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Timed out on obtaining data from pull based adapter. Trying again!");
+ }
+ break;
+ }
+
+ } catch (Exception failureException) {
+ try {
+ failureException.printStackTrace();
+ boolean continueIngestion = policyEnforcer.continueIngestionPostSoftwareFailure(failureException);
+ if (continueIngestion) {
+ tupleBuilder.reset();
+ continue;
+ } else {
+ throw failureException;
+ }
+ } catch (Exception recoveryException) {
+ throw new Exception(recoveryException);
+ }
+ }
+ }
+ }
+
+ /**
+ * Discontinue the ingestion of data and end the feed.
+ *
+ * @throws Exception
+ */
+ public void stop() throws Exception {
+ continueIngestion = false;
+ }
+
+ public Map<String, String> getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public boolean handleException(Exception e) {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/ae85a1dc/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FeedClient.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FeedClient.java b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FeedClient.java
new file mode 100644
index 0000000..da2cde0
--- /dev/null
+++ b/asterix-external-data/src/main/java/edu/uci/ics/asterix/external/dataset/adapter/FeedClient.java
@@ -0,0 +1,170 @@
+/*
+ * Copyright 2009-2013 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.asterix.external.dataset.adapter;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+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.exceptions.AsterixException;
+import edu.uci.ics.asterix.dataflow.data.nontagged.serde.ARecordSerializerDeserializer;
+import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import edu.uci.ics.asterix.om.base.ABoolean;
+import edu.uci.ics.asterix.om.base.AInt32;
+import edu.uci.ics.asterix.om.base.AMutableDateTime;
+import edu.uci.ics.asterix.om.base.AMutableInt32;
+import edu.uci.ics.asterix.om.base.AMutableOrderedList;
+import edu.uci.ics.asterix.om.base.AMutablePoint;
+import edu.uci.ics.asterix.om.base.AMutableRecord;
+import edu.uci.ics.asterix.om.base.AMutableString;
+import edu.uci.ics.asterix.om.base.AMutableUnorderedList;
+import edu.uci.ics.asterix.om.base.AString;
+import edu.uci.ics.asterix.om.base.IACursor;
+import edu.uci.ics.asterix.om.base.IAObject;
+import edu.uci.ics.asterix.om.types.AOrderedListType;
+import edu.uci.ics.asterix.om.types.ARecordType;
+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.data.std.util.ArrayBackedValueStorage;
+
+public abstract class FeedClient implements IFeedClient {
+
+ protected static final Logger LOGGER = Logger.getLogger(FeedClient.class.getName());
+
+ protected ARecordSerializerDeserializer recordSerDe;
+ protected AMutableRecord mutableRecord;
+ protected boolean messageReceived;
+ protected boolean continueIngestion = true;
+ protected IARecordBuilder recordBuilder = new RecordBuilder();
+
+ protected AMutableString aString = new AMutableString("");
+ protected AMutableInt32 aInt32 = new AMutableInt32(0);
+ protected AMutablePoint aPoint = new AMutablePoint(0, 0);
+ protected AMutableDateTime aDateTime = new AMutableDateTime(0);
+
+ @SuppressWarnings("unchecked")
+ protected ISerializerDeserializer<AString> stringSerde = AqlSerializerDeserializerProvider.INSTANCE
+ .getSerializerDeserializer(BuiltinType.ASTRING);
+ @SuppressWarnings("unchecked")
+ protected ISerializerDeserializer<ABoolean> booleanSerde = AqlSerializerDeserializerProvider.INSTANCE
+ .getSerializerDeserializer(BuiltinType.ABOOLEAN);
+ @SuppressWarnings("unchecked")
+ protected ISerializerDeserializer<AInt32> int32Serde = AqlSerializerDeserializerProvider.INSTANCE
+ .getSerializerDeserializer(BuiltinType.AINT32);
+
+ public abstract InflowState retrieveNextRecord() throws Exception;
+
+ @Override
+ public InflowState nextTuple(DataOutput dataOutput, int timeout) throws AsterixException {
+ try {
+ InflowState state = null;
+ int waitCount = 0;
+ boolean continueWait = true;
+ while ((state == null || state.equals(InflowState.DATA_NOT_AVAILABLE)) && continueWait) {
+ state = retrieveNextRecord();
+ switch (state) {
+ case DATA_AVAILABLE:
+ recordBuilder.reset(mutableRecord.getType());
+ recordBuilder.init();
+ writeRecord(mutableRecord, dataOutput, recordBuilder);
+ break;
+ case DATA_NOT_AVAILABLE:
+ if (waitCount > timeout) {
+ continueWait = false;
+ } else {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Waiting to obtain data from pull based adaptor");
+ }
+ Thread.sleep(1000);
+ waitCount++;
+ }
+ break;
+ case NO_MORE_DATA:
+ break;
+ }
+ }
+ return state;
+ } catch (Exception e) {
+ throw new AsterixException(e);
+ }
+
+ }
+
+ private void writeRecord(AMutableRecord record, DataOutput dataOutput, IARecordBuilder recordBuilder)
+ throws IOException, AsterixException {
+ ArrayBackedValueStorage fieldValue = new ArrayBackedValueStorage();
+ int numFields = record.getType().getFieldNames().length;
+ for (int pos = 0; pos < numFields; pos++) {
+ fieldValue.reset();
+ IAObject obj = record.getValueByPos(pos);
+ writeObject(obj, fieldValue.getDataOutput());
+ recordBuilder.addField(pos, fieldValue);
+ }
+ recordBuilder.write(dataOutput, true);
+ }
+
+ private void writeObject(IAObject obj, DataOutput dataOutput) throws IOException, AsterixException {
+ switch (obj.getType().getTypeTag()) {
+ case RECORD: {
+ IARecordBuilder recordBuilder = new RecordBuilder();
+ recordBuilder.reset((ARecordType) obj.getType());
+ recordBuilder.init();
+ writeRecord((AMutableRecord) obj, dataOutput, recordBuilder);
+ break;
+ }
+
+ case ORDEREDLIST: {
+ OrderedListBuilder listBuilder = new OrderedListBuilder();
+ listBuilder.reset((AOrderedListType) ((AMutableOrderedList) obj).getType());
+ IACursor cursor = ((AMutableOrderedList) obj).getCursor();
+ ArrayBackedValueStorage listItemValue = new ArrayBackedValueStorage();
+ while (cursor.next()) {
+ listItemValue.reset();
+ IAObject item = cursor.get();
+ writeObject(item, listItemValue.getDataOutput());
+ listBuilder.addItem(listItemValue);
+ }
+ listBuilder.write(dataOutput, true);
+ break;
+ }
+
+ case UNORDEREDLIST: {
+ UnorderedListBuilder listBuilder = new UnorderedListBuilder();
+ listBuilder.reset((AUnorderedListType) ((AMutableUnorderedList) obj).getType());
+ IACursor cursor = ((AMutableUnorderedList) obj).getCursor();
+ ArrayBackedValueStorage listItemValue = new ArrayBackedValueStorage();
+ while (cursor.next()) {
+ listItemValue.reset();
+ IAObject item = cursor.get();
+ writeObject(item, listItemValue.getDataOutput());
+ listBuilder.addItem(listItemValue);
+ }
+ listBuilder.write(dataOutput, true);
+ break;
+ }
+
+ default:
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(obj.getType()).serialize(obj,
+ dataOutput);
+ break;
+ }
+ }
+}