Support for Data Ingestion in AsterixDB

Ingesting Twitter Stream

We shall use the built-in push-based Twitter adapter. As a pre-requisite, 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 as closed {
        screen_name: string,
        lang: string,
        friends_count: int32,
        statuses_count: int32
    };

    create type Tweet as open {
        id: int64,
        user: TwitterUser
    }

    create dataset Tweets (Tweet)
    primary key id;

We also create a dataset that we shall use to persist the tweets in AsterixDB. Next we make use of the create feed AQL statement to define our example data feed.

    use dataverse feeds;

    create feed TwitterFeed if not exists using "push_twitter"
    (("type-name"="Tweet"),
     ("format"="twitter-status"),
     ("consumer.key"="************"),
     ("consumer.secret"="**************"),
     ("access.token"="**********"),
     ("access.token.secret"="*************"));

    use dataverse feeds;

    create feed TwitterFeed if not exists using "push_twitter"
    (("type-name"="Tweet"),
     ("format"="twitter-status"),
     ("consumer.key"="************"),
     ("consumer.secret"="**************"),
     ("access.token"="**********"),
     ("access.token.secret"="*************"),
     ("keywords"="news"));

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 activated using the start feed statement. Before we active a feed, we need to designate the dataset where the data to be persisted using connect feed statement. Subsequent to a connect feed statement, the feed is said to be in the connected state. After that, start feed statement will activate the feed, and start the dataflow from feed to its connected dataset. Multiple feeds can simultaneously be connected to a dataset such that the contents of the dataset represent the union of the connected feeds. Also one feed can be simultaneously connected to multiple target datasets.

    use dataverse feeds;

    connect feed TwitterFeed to dataset Tweets;

    start feed TwitterFeed;

The connect feed statement above directs AsterixDB to persist the data from TwitterFeed feed into the Tweets dataset. The start feed statement will activate the feed and start the dataflow. 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.

Let the feed run for a minute, then run the following query to see the latest tweets that are stored into the data set.

    use dataverse feeds;

    for $i in dataset Tweets limit 10 return $i;

The dataflow of data from a feed can be terminated explicitly by stop feed statement.

    use dataverse feeds;

    stop feed TwitterFeed;

The disconnnect statement can be used to disconnect the feed from certain dataset.

    use dataverse feeds;

    disconnect feed TwitterFeed from dataset Tweets;

Using the “localfs” feed adapter

localfs adapter enables data ingestion from local file system. It allows user to feed data records on local disk into a dataset. A DDL example for creating a localfs feed is given as follow:

    use dataverse feeds;

    create type TweetType as closed {
      id: string,
      username : string,
      location : string,
      text : string,
      timestamp : string
    }

    create dataset Tweets(TweetType)
    primary key id;

    create feed TweetFeed
    using localfs
    (("type-name"="TweetType"),("path"="HOSTNAME://LOCAL_FILE_PATH"),("format"="adm"))

Similar to previous examples, we need to define the datatype and dataset this feed uses. The “path” parameter refers to the local datafile that we want to ingest data from. HOSTNAME can either be the IP address or node name of the machine which holds the file. LOCAL_FILE_PATH indicates the absolute path to the file on that machine. Similarly to socket_adapter, this feed takes adm formatted data records.

Ingestion with autogenerated key

AsterixDB supports using autogenerated uuid as the primary key for dataset. When we use this feature, we will need to define a datatype with the primary key field, and specify that field to be autogenerated when creating the dataset. Use that same datatype in feed definition will cause a type discrepancy since there is no such field in the datasource. Thus, we will need to define two separate datatypes for feed and dataset:

    use dataverse feeds;

    create type DBLPFeedType as closed {
      dblpid: string,
      title: string,
      authors: string,
      misc: string
    }

    create type DBLPDataSetType as open {
      id: uuid,
      dblpid: string,
      title: string,
      authors: string,
      misc: string
    }
    create dataset DBLPDataset(DBLPDataSetType) primary key id autogenerated;

    create feed DBLPFeed using socket_adapter
    (
        ("sockets"="127.0.0.1:10001"),
        ("address-type"="IP"),
        ("type-name"="DBLPFeedType"),
        ("format"="adm")
    );

    connect feed DBLPFeed to dataset DBLPDataset;

    start feed DBLPFeed;

Policies

• Spill: Objects that cannot be processed by an operator for lack of resources (referred to as excess objects hereafter) should be persisted to the local disk for deferred processing.

• Discard: Excess objects should be discarded.

Note that the end user may choose to form a custom policy. For example, it is possible in AsterixDB to create a custom policy that spills excess objects 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 or else the “Basic” policy will be chosen as the default.

    use dataverse feeds;

    connect feed TwitterFeed to dataset Tweets
    using policy Basic;