Deploying an AI alerting system with Solr's Streaming Expressions

As of Solr 6.4, Streaming Expressions provide all the tools you need to deploy an Artificial Intelligence alerting system. Let's take a look at some of the tools involved and then walk through a simple AI alerting use case.

The Tools

1) A Text Classifier: Solr's Streaming Expressions allow you to train, store, and deploy a text classifier. With a text classifier you can train a model to determine the probability that a document belongs to a certain class. This provides the brains behind the alert.

2) A Messaging System: Solr's Streaming Expressions library provides the topic function, which allows clients to subscribe to a query. Once the subscription is established the topic provides one-time delivery of documents that match the topic query. This provides the nervous system of the alerting engine.

3) Actors: Solr's Streaming Expressions library provides the daemon function, which are processes that live inside Solr. Daemon's have the ability to subscribe to topics, apply a classifier, update collections and take other actions. This provides the muscle behind the alerts.

Use Case: A Threat Alerting Engine

Let's walk through the steps for building an engine that detects threats in social media posts and sends alerts.

Step 1: Train a model that classifies threats

To train the model we need to assemble a training set comprised of positive and negative examples of the class. In this case we need to gather a set of social media posts that are threats and another set of social media posts that are not threats.

Once we have our training set we can load the data into a Solr Cloud collection. Then we can use the features, train and update functions to extract the key features, train the model (based on the features) and store the model iterations in a Solr Cloud collection :

update(models,
             batchSize="50",
             train(trainingSet,
                      features(trainingSet,
                                     q="*:*",
                                     featureSet="threatFeatures",
                                     field="body",
                                     outcome="out_i",
                                     numTerms=250),
                      q="*:*",
                      name="threatModel",
                      field="body",
                      outcome="out_i",
                      maxIterations="100"))



Let's explore this expression in more detail.

The features function extracts the key features (terms) from the training data. The features function scores terms using a technique known as Information Gain to determine which features are the most important in distinguishing the positive and negative set. The features function also extracts the term statistics from the training set needed to build the model. The online documentation contains more detailed information for the features function.

The train function uses the terms and statistics emitted by the features function to train the model. The train function use a parallel iterative, batch gradient descent approach to train a logistic regression text classifier. The train function emits a tuple for each iteration of the model. Each model tuple includes the terms, weights, error rate and a confusion matrix that describes the classification errors for the iteration. The online documentation contains more detailed information for the train function.

The update function is used to store each iteration of the model in a Solr Cloud collection. The online documentation contains more detailed information for the update function. 

Step 2: Inspect and test the model

The next step is to validate that you have a good model. A quick inspection of the model iterations can be useful in understanding the model. Here are a few things to look for:

a) Look at the terms in the model. A quick overview of the terms can be useful in understanding what the key features are in the model.

b) Look at weights for the terms to get a feel for how the terms are weighted in the model.

c) Look at the confusion matrix for the last iteration of the model to get a feeling for error rates that occurred in the final training iteration.

d) Look at the iterations to see if the model converged, which means the error rates gradually reduced until they reach a point of stabilization.

After reviewing the model we can test the model on a test dataset. The test dataset should consist of positive and negative examples of the class and should not be the same as the training dataset.

The expression below runs the classifier on the test set:

classify(model(models,
                          id="threatModel",
                          cacheMillis=5000),
              search(testData,
                          q="*:*",
                          fl="id,  body",
                          sort="id desc"),
              field="body")

Let's breakdown the expression above.

The classify function calls the model function to retrieve a named model stored in a Solr Cloud collection. In the example above it retrieves the threatModel which we built earlier. The model function retrieves the highest iteration of the model found in the collection and caches it in memory for a specified period of time. For more detailed information on the classify, model and search functions you can review the documentation.

Once the classify function has the model, it reads tuples from an internal stream and scores the tuples by applying the model to a text field in the document. The classify function emits all the tuples from the underlying stream and adds a field called probability_d, which is a score between 0 and 1 that describes the probability that the tuple is in the class. The higher the score, the higher the probability. You can start with the score of .5 or greater as the threshold for positive or negative class assignment.

You can then read the tuples to see how they were scored and determine how accurate the classifier is. If there are false positives or false negatives in the test results then inspect the records to see why they might have been misclassified. If there are missing features in the training set you may need to add more examples that include these missing features and rebuild the model.

You can also adjust the threshold to see if it creates a more accurate classifier. For example if changing the threshold from .5 to .6 provides fewer false positives without increasing false negatives then you can use that as the threshold for the classifier when deployed.

Step 3: Setup the Actor

Once you're satisfied with the model, its time to create an actor that can read new documents as they enter the index, classify them and index the threats in a separate collection.

To do this we'll setup a daemon to run the classify function. But instead of using a search function as the internal stream to classify, we'll use a topic.

Here is the basic syntax for this:

daemon(id="alertDaemon",
              update(threats,
                          batchSize="10",
                          having(classify(model(models,
                                                                id="threatModel",
                                                                cacheMillis="5000"),
                                                     topic(messages,
                                                               checkpoints,
                                                               id="messageTopic",
                                                               q="*:*",
                                                               fl="id, body"),
                                                     field="body"),
                                         gt(probability_d, 0.5)
                                       )
                           )
             )


Let's explore this expression from inside-out.

The inner topic expression is subscribing to the messages collection. The messages collection holds the social media messages we are alerting on. The topic will provide one time delivery of documents in this collection. Each time it is called it will return a new batch of documents.

The classify function wraps the topic and scores each tuple based on the model retrieved by the internal model function. The classify function emits the tuples from the topic and adds the class score to the outgoing tuples in the probability_d field.

The having expression wraps the classify function and only emits tuples with a probability_d greater than 0.5. The having function is setting the classification threshold for the alert.  The online documentation contains more detailed information for the having function.

The update function indexes the tuples emitted by the having function into a Solr Cloud collection called threats. The threats collection is where the messages classified as threats are stored.

The daemon function wraps the update function and calls it at intervals using an internal thread. Each time the daemon runs, a new batch of messages will be classified and the threats will be added to the threats collection. New records added to the messages collection will automatically be classified in the order they were added. The online documentation contains more detailed information for the daemon function.

Step 4: Adding actors to watch the threats collection

We now have a threats collection where the threats are being indexed. Any number of actors can subscribe to the threats collection using a topic function or TopicStream java class. These actors can be daemons running inside Solr or actors external to Solr.

In this scenario the threats collection is used as a message queue for other actors. These actors can then be programmed to take specific actions based on the threat.

Deploying Solr's New Parallel Executor

In an a recent blog we explored Solr's new parallel batch capabilities. This blog expands on parallel batch and introduces Solr's new parallel executor. Solr's parallel executor allows Streaming Expressions to be stored in a Solr Cloud collection where they can be streamed to worker nodes and executed in parallel.

The executor Function

The new executor function performs the compilation and execution of Streaming Expressions on worker nodes. The executor function has an internal thread pool that executes Streaming Expressions in parallel within a single worker node. The queue of streaming expressions can also be partitioned across a cluster of worker nodes providing a second level of parallelism.

Deploying a General Purpose Work Queue

The executor function can be used with the daemon and topic functions to deploy a general purpose work queue. An example of this expression construct is below:

daemon(id="daemon1",
               executor(threads=5,
                               topic(checkpointCollection,
                                         storedExpressions,
                                         id="topic1",
                                         initialCheckpoint=0,
                                         q="*:*",
                                         fl="id, expr_s")))

Let's break down the expression above starting with the topic function.

The topic function subscribes to a query and provides one-time delivery of documents that match the query. In the example, the topic function is subscribed to a collection of stored Streaming Expressions.

The executor function wraps the topic and for each tuple it compiles and runs the expression in the expr_s field. The executor has an internal thread pool and each expression is compiled and run in its own thread. The threads parameter controls the size of the thread pool.

The daemon function wraps the executor and calls it at intervals using an internal thread. This will cause the executor to iterate over the topic and execute all the Streaming Expressions in the work queue in batches.

The daemon function will continue to run at intervals when the queue is empty. As new tasks are indexed into the queue they will automatically be read by the topic and executed.

Prioritizing Tasks with the priority Function  

In the example above, the executor will run tasks in the order that they are emitted by the topic. Topic's emit tuples ordered by Solr's internal _version_ number. This behaves similar to a FIFO queue (but without strict FIFO enforcement). But the topic alone doesn't have any concept of task prioritization.

The priority function can be used to allow higher priority tasks to be scheduled ahead of lower priority tasks. The priority function wraps two topics. The first topic is the higher priority queue and the second topic is the lower priority queue. The priority function will only emit a lower priority task when there are no higher priority tasks in the queue.

daemon(id="daemon1",
               executor(threads=5,
                               priority(topic(checkpointCollection,
                                                      highPriorityTasks,
                                                      id="high",
                                                      initialCheckpoint=0,
                                                      q="*:*",
                                                     fl="id, expr_s"),
                                             topic(checkpointCollection,
                                                       lowPriorityTasks,
                                                       id="low",
                                                       initialCheckpoint=0,
                                                       q="*:*",
                                                      fl="id, expr_s")))

Deploying a Parallel Work Queue

The parallel function can be used to partition tasks across a worker collection. This provides parallel execution within a single worker and across a cluster of workers. The syntax for deploying a parallel work queue is below:

parallel(workerCollection,
              workers=6,
              sort="DaemonOp asc",
              daemon(id="daemon1",
                             executor(threads=5,
                                             topic(checkpointCollection,
                                                       storedExpressions,
                                                       id="topic1",
                                                       initialCheckpoint=0,
                                                       q="*:*",
                                                       fl="id, expr_s",
                                                       partitionKeys="id"))))

In the example above the parallel function sends daemons to 6 workers. Each worker executes a partition of the work queue.

Deploying Replicas to Increase Cluster Capacity

Expressions run by the executor search Solr Cloud collections to retrieve data. These searches will be spread across all of the replicas in the Solr Cloud collections. As the number of workers executing expressions increases, more replicas can be added to the Solr Cloud collections to increase the capacity of the entire system.

Solr 6.3: Finding the most relevant facets with the scoreNodes Streaming Expression

Starting with Solr 6.3 you can use the scoreNodes Streaming Expression to find the most relevant facets and significant relationships in a distributed graph. This blog describes how the scoreNodes function can be used with facets. A future blog will cover using scoreNodes with graph expressions.

Why Score Facets?

One typical use case for scoring facets would be for lightning fast recommendations based on market basket co-occurrence. We'll explore this scenario below:

First let's look at the syntax for scoring facets:

scoreNodes(facet(baskets,
                             q="products:A",
                             buckets="products",
                             bucketSorts="count(*) desc",
                             bucketSizeLimit=50,
                             count(*)))

Let's breakdown what the expression is doing.

The facet expression calls Solr's JSON facet API and emits tuples which contain the facet results. In this case it is searching the baskets collection. The query is looking for all records in the baskets collection that have product A in the products field.

The baskets collection contains a multi-valued field called products which contains all the products in the basket. For example

id                 products
basket1        [A, B, C]
basket2        [A, C, E]
basket3        [B, C, D]

The sample facet expression will return the following tuples:

products:  C
count(*):  2

products:  B
count(*):  1

products: E
count(*): 1

Product C is in two baskets with product A. Products B and E are both in one basket with product A.

So it would seem that the most relevant facet/product for product A would be product C.

But, there is something we don't know yet. How often product C occurs in all the baskets. If product C occurs in a large percentage of baskets, then it doesn't have any particular relevance to product A.

This is where the scoreNodes function does it's magic. The scoreNodes function scores the facets based on the raw facet counts and their frequency across the entire collection.

The scoring formula is similar to the tf*idf scoring algorithm used to score results from a full text search. In the full text context tf (term frequency) is the number of times the term appears in the document. idf (inverse document frequency) is computed based on the document frequency of the term, or how many documents the term appears in. The idf is used to provide a boost to terms that are more rare in the index.

scoreNodes uses the same  principal to score facets, but the facet count is used as the tf value in the formula. The idf is computed for each facet term based on global statistics across the entire collection. The effect of the scoreNodes algorithm is to provide a boost to facet terms that are rarer in the collection.

The scoreNodes functions adds a field to each facet tuple called nodeScore, which is the relevance score for the facet. You can use the top expression to find the most relevant facet:

top(n=2, sort="nodeScore desc",
      scoreNodes(facet(baskets,
                                   q="products:A",
                                   buckets="products",
                                   bucketSorts="count(*) desc",
                                   bucketSizeLimit=50,
                                   count(*)))

The expression above emits the two highest scoring facets based on the nodeScore.

Solr 6.3: Batch jobs, Parallel ETL and Streaming Text Transformation

Solr 6.3 is on it's way and along with it comes a new execution paradigm for Solr: Parallel batch. Solr's new batch capabilities open up a new world of use cases for Solr. This blog will cover the basics of how the parallel batch framework operates and describe how it can be used to perform parallel ETL and parallel text transformations.

Not built on MapReduce

Solr's Streaming Expressions have had MapReduce capabilities for quite a while now. But Solr's MapReduce implementation is designed to support interactive queries over large data sets and to power the Parallel SQL interface when run in MapReduce mode.

And notably Solr's MapReduce implementation does not support streaming of text fields, which makes it unsuitable for performing ETL and text transformations in Solr.

Parallel batch is built on message queues

Solr also has massaging capabilities that allow an entire SolrCloud collection to be treated like a message queue, similar in nature to Apache Kafka. While Apache Kafka is more scalable, Solr's messaging queue is more flexible in that it allows you to subscribe to a query.

The Streaming Expression that allows you to subscribe to a query is the topic expression. Here is the basic syntax:

topic(checkpointCollection,
          dataCollection,
          q="*:*",
          fl="from, to, body",
          id="myTopic",
          rows="300",
          initialCheckpoint="0")

Here is an explanation of the parameters:

1. checkpointCollection: The topic function tracks the checkpoints for a specific topic id. It stores the checkpoints in this collection.
2. dataCollection: This is the collection that the topic results are pulled from.
3. q: The query to use to pull records for this topic.
4. id: The unique id of the topic. A different set of checkpoints will be maintained for each unique topic id.
5. rows: The number of rows to fetch from each shard, each time the topic function is called.
6. initialCheckpoint: Where in the queue to start fetching results from. Setting to 0 will cause the topic to match all the records that match the topic query in the collection. Not setting the initialCheckpoint will cause the topic to begin fetching records that have been added after the topic has been initiated. 

When the topic function is sent to the /stream handler it will retrieve a batch of rows from the topic and update the checkpoints for the topic. The next time it's called it will retrieve the next batch of rows.

The topic function has no restriction on the data that can be retrieved. So it can return any stored fields including stored text fields.

Iterating the topic with a daemon

In order to process all the records in a topic, we will need to call the topic repeatedly until it stops returning results. The daemon function can do this for us.

The daemon function wraps another function and calls it at intervals using an internal thread. When a daemon is passed to the /stream handler, the /stream handler recognizes it and keeps it in memory so that it can run until it's completed it's job.

Here is the basic syntax:

daemon(id="myDaemon",
               terminate="true",
               topic(checkpointCollection,
                        dataCollection,
                        q="*:*",
                        fl="from, to, body",
                        id="myTopic",
                        rows="300",
                        initialCheckpoint="0"))

Notice the terminate parameter, which is new in Solr 6.3. This tells the daemon to terminate after the topic has no more records to process.

Sending Tuples to another SolrCloud collection

As the daemon iterates a topic it can send the results to another SolrCloud collection using the update function.

Here is the basic syntax:

daemon(id="myDaemon",
               terminate="true",
               update(destinationCollection,
                           batchSize=300,
                           topic(checkpointCollection,
                                     dataCollection,
                                     q="*:*",
                                     fl="from, to, body",
                                     id="myTopic",
                                     rows="300",
                                     initialCheckpoint="0")))

The example above is sending all the Tuples that are emitted by the topic to another SolrCloud collection.

Performing transformations on the Tuples

The data in the Tuples emitted by the topic can be adjusted/transformed before they are sent to the destinationCollection. Here is a very simple example:

daemon(id="myDaemon",
               terminate="true",
               update(destinationCollection,
                           batchSize=300,
                           select(from as from_s,
                                      to as to_s,
                                      body as body_t,
                                      topic(checkpointCollection,
                                                dataCollection,
                                                q="*:*",
                                                fl="from, to, body",
                                                id="myTopic",
                                                rows="300",
                                                initialCheckpoint="0"))))

In the example above the select function is changing the field names in the Tuples before the tuples are processed by the update function.

Text analysis and transformation

Starting with Solr 6.3 Streaming Expressions have access to Lucene/Solr's text analyzers. This allows functions to be written that process text using the full power of these analyzers. The first example of this is the classify funtion in Solr 6.3, which analyzes text in a text field and extracts the features needed to perform classification using a linear classifier. This deserves an entire blog in it's own right so it's only mentioned here as an example of how Streaming Expressions can use analyzers.

Also in Solr 6.3 you can add your own Stream Expressions and register them in the solrconfig.xml. So you can write and plugin your own text analysis functions.

Fetching data from other collections with the fetch function

In Solr 6.3 the fetch function has been added that allows fields from other collections to be fetched and added to documents. Here is the sample syntax:

daemon(id="myDaemon",
               terminate="true",
               update(destinationCollection,
                           batchSize=300,
                           fetch(userAddresses,
                                     on="from=userID",
                                     fl="address",
                                     topic(checkpointCollection,
                                                dataCollection,
                                                q="*:*",
                                                fl="from, to, body",
                                                id="myTopic",
                                                rows="300",
                                                initialCheckpoint="0"))))

The example above fetches the address from the usersAddresses collection, by mapping the from field in the Tuples to the userID in the userAddresses collection.

Parallel batch processing

The parallel function can be used to partition a batch job across a cluster of worker nodes. Here is the basic syntax:

parallel(workerCollection,
               workers=10,
               sort="DaemonOp desc",
               daemon(id="myDaemon",
                              terminate="true",
                              update(destinationCollection,
                                           batchSize=300,
                                           select(from as from_s,
                                                      to as to_s,
                                                      body as body_t,
                                                      topic(checkpointCollection,
                                                                dataCollection,
                                                                q="*:*",
                                                                fl="id, from, to, body",
                                                                id="myTopic",
                                                                rows="300",
                                                                initialCheckpoint="0",
                                                                partitionKeys="id")))))

In this example the parallel function sends the daemon to 10 worker nodes. Each worker will process a partition of the topic. Notice that the partitionKeys field has been added to the topic. This tells the topic to hash partition on the id field in the dataCollection.

Quick note about the DaemonOp sort parameter. The parallel function sends the daemon to 10 worker nodes. The worker nodes return a Tuple that confirms that a daemon operation was started. The DaemonOp is simply the confirmation Tuple. The parallel function never sees the tuples generated by the topic function as they are sent to another SolrCloud collection.

Parallel batch throughput

When performing parallel batch operations, each worker will be iterating over the topic in parallel. The topic function randomly selects a replica from each shard to query for it's data. So when performing parallel batch operations all of the replicas in the cluster will be streaming content at once.    

Training and Storing Machine Learning Models with Solr 6.2

Solr's Streaming Expressions can already perform Parallel SQL, Graph Traversal, Messaging, MapReduce and Stream Processing. In Solr 6.2 Streaming Expressions adds machine learning capabilities.

Starting with Solr 6.2 you'll be able to train a Logistic Regression Text Classifier on training sets stored in Solr Cloud. This means Solr can now build machine learning models natively.

Now Solr can learn.

What is a Logistic Regression Text Classifier?

Logistic regression is a supervised machine learning classification algorithm. Logistic regression is used to train an AI model that classifies data. Logistic Regression is a binary classifier which means that it's used to classify if something belongs to a class or not.

Logistic Regression is a supervised algorithm, which means it needs a training set that provides positive and negative examples of the class to build the model.

Logistic Regression models are trained using specific features that describe the data. With text classification the features of the data are the terms in the documents.

Once the model is trained it can be used to classify other documents based on their features (terms).

The model returns a score predicting the probability that a document is in the class or not.

Solr's Implementation Through Streaming Expressions

Solr 6.2 has two new Streaming Expression functions: features and train.

Features: Extracts the key terms from a training set stored in a Solr Cloud collection. The features function uses an algorithm called Information Gain to determine which terms are most important for the specific training set.

Train: Uses the extracted features to train the Logitistic Regression Model. The train function uses a parallel iterative, batch Gradient Descent approach to optimize the model. The algorithm is embedded into Solr's core so only the model is streamed across the network with each iteration.

Storing Features and Models

The output from both the features and train function can be redirected to another SolrCloud collection using the update function.

This approach allows Solr to store millions of models that can be easily retrieved and deployed.

Learning What Users Like

One of the interesting use cases for this feature is to build models for every user which encapsulates what users like to read.

In order to build a model for each user, we need to pull together positive and negative data sets for each user. The positive set includes documents that the user has an interest in and the negative set contains documents the user has not shown a particular interest in.

Using Graph Expressions To Build The Positive Set

Application usage logs can be queried to find what the user likes to read (positive set). Indexing the usage logs in Solr Cloud allows us to run graph queries that identify the positive set.

Graph queries can be run that return a set of documents that the user has read. Graph queries can also be used to expand the training set. For example the training set can be expanded to include the documents that are most frequently viewed in the same session with documents that the user has viewed.

Graph queries can also be used to expand the positive training set through collaborative filtering. Using this approach documents read by users that have similar reading habits to the user can be pulled into the positive set.

Collecting the Negative Training Set

Logistic Regression requires a negative training set which can be gathered by down sampling the main corpus. This means taking a random sample from the main corpus that is similar in size to the positive training set.

The random Streaming Expression can be used for down sampling. The random Streaming Expression returns a random set of documents that match a query.

Low frequency terms from the positive training set can be extracted and used to negate documents from the random negative training set. This would ensure that the negative training set doesn't include key terms from the positive training set.

Using Easy Ensemble to Provide Larger Samples of Negative Data

If the negative training set is too small to represent the main corpus an approach known as Easy Ensemble can be used to expand the sample size of the negative set.

With Easy Ensemble multiple down sampled negative sets are collected and the positive training set is trained against each negative set. This will train an ensemble of models that can be used to classify documents.

With the ensemble approach the output from all the models are averaged to arrive at an ensemble classification.

Automating the Building of Training Sets With Daemons

Solr 6 introduced daemons. Daemons are Streaming Expressions that live inside Solr and run in the background at intervals.

Daemons can now be used to setup processes inside Solr that monitor the logs and build models for users as new data enters the logs.

Once the daemon processes are in place Solr can learn on it's own.

Future blogs will discuss how to setup daemons to monitor the logs and take action.

Re-Ranking and Recommending

Once the models are stored in a Solr Cloud collection they can be retrieved and used to score documents. The score reflects the probability that the user will like the document.

The scores can be applied in Solr's re-ranker to re-order the top N search results. This provides custom rankings for each user.

The scores can also be used to score recommendations created from graph expressions or More Like This queries. This provides personalized recommendations.

Understanding Solr's AnalyticsQuery

Back in Solr 4.9 an AnalyticsQuery API was added that allows developers to plug custom analytic logic into Solr. The AnalyticsQuery class provides a clean and simple API that gives developers access to all the rich functionality in Lucene and is strategically placed within Solr’s distributed search architecture. Using this API you can harness all the power of Lucene and Solr to write custom analytic logic.

Let’s dive in!

The AnalyticsQuery API is a natural extension of the PostFilter API. So let’s start with PostFilters and then we’ll move quickly to the AnalyticsQuery API.

PostFilters

Lucene has the concept of a Collector. A Collector collects the results that match a search.

The PostFilter API wraps a DelegatingCollector around a ranking collector to filter results. The DelegatingCollector is passed each search result, applies filtering logic to the result and then “delegates” the result to the ranking Collector, if it chooses.

DelegatingCollectors can wrap other DelegatingCollectors to form a pipeline. Each Delegating collector looks at the search result and delegates to the next collector. The order that DelegatingCollectors will be executed is controlled by a “cost” parameter that is built into Solr’s local parameter query syntax.

This was put in so that costly PostFilters could be run after less costly PostFilters, thus reducing the number of documents seen by the more costly PostFilter.

DelegatingCollectors also have a finish() method that notifies the collector that all documents have been collected.

The AnalyticsQuery API

The PostFilter API is very strategically positioned because it sees every search result. The AnalyticsQuery API capitilizes on that strategic positioning by adding a few key features.

• It provides easy access to the ResponseBuilder object, so analytics data can be easily output and piped between AnalyticsQueries. 

• It provides a way to insert a custom MergeStrategy, so that analytic output from the shards can be merged. 

Extending the AnalyticsQuery class

To hook into the AnalyticsQuery API you need to extend the AnalyticsQuery class. This class has two constructors:

public AnalyticsQuery();

Use the above constructor if you’re doing analytics on a single Solr server.

public AnalyticsQuery(MergeStrategy mergeStrategy);

Use the above constructor if you’re doing distributed analytics. If you use this constructor you’ll need to also write a MergeStrategy implementation. In my next blog I’ll cover the MergeStrategy API, but for now it’s enough to know that it exists, and what it’s used for.

Then you have to implement the hook method:

public abstract DelegatingCollector getAnalyticsCollector(ResponseBuilder rb, IndexSearcher searcher); 

This is where you return your DelegatingCollector implementation that will collect the analytics. Notice that you are passed the current IndexSearcher and ResponseBuilder. You can access both the request context and the Solr response objects through the response builder.

Through the request context you can pipeline the result of one AnalyticsQuery to another AnalyticsQuery, controlling the order of execution using the “cost” parameter. You can place your analytic output directly into the Solr response through the response object in the ResponseBuilder.

Below is a very simple class that extends AnalyticsQuery

 public class MyAnalyticsQuery extends AnalyticsQuery {

    public MyAnalyticsQuery(MergeStrategy mergeStrategy)
    {
      super(mergeStrategy);
    }

    public DelegatingCollector getAnalyticsCollector(ResponseBuilder rb, IndexSearcher searcher)
    {
      return new MyAnalyticsCollector(rb);
    }
  }

And here is a very simple implementation of a DelegatingCollector (MyAnalyticsCollector):

 class MyAnalyticsCollector extends DelegatingCollector {
    ResponseBuilder rb;
    int count;

    public MyAnalyticsCollector(ResponseBuilder rb) {
      this.rb = rb;
    }

    public void collect(int doc) throws IOException {
      ++count;
      leafDelegate.collect(doc);
    }

    public void finish() throws IOException {
      NamedList analytics = new NamedList();
      rb.rsp.add("analytics", analytics);
      analytics.add("mycount", count);
      if(this.delegate instanceof DelegatingCollector) {
        ((DelegatingCollector)this.delegate).finish();
      }
    }
  }

Notice that in the collect(int doc) method, it is incrementing a counter and then calling the collect method on it’s delegate. The collect method is called for each document in the result set. In the finish() method the count is placed into the output response. See the full API for the DelegatingCollector class to understand how it works.

Plugging In Your Code

Just like PostFilters, an AnalyticsQuery is passed in through a custom QParserPlugin as a filter query. Here is an example AnalyticsQuery being passed in through the “fq” parameter:

q=*:*&fq={!myanalytic param1=a param2=b cost=101}&fq={!myanalytic2 param1=a param2=b cost=102}

In the local parameter syntax above there are two filter queries pointing to custom QParserPlugins. The local parameter syntax “!myanalytic” tells Solr to load the “myanalytic” QParserPlugin, which will return a Query that extends AnalyticsQuery.

Notice the “cost” parameters which will control the order of execution. In this case the “myanalytic” AnalyticsQuery will execute ahead of the “myanalytic2″ AnalyticsQuery.

Interaction with the CollapsingQParserPlugin

The CollapsingQParserPlugin is a PostFilter that performs field collapsing. Because it’s a PostFilter it can be layered in among AnalyticsQueries using the “cost” parameter to control order of execution. So, you can conveniently collect analytics before and after field collapsing.

Parallel Streaming Transformations

Prior blogs have covered the basics of solrj.io. In this blog we begin to learn how to harness the parallel computing capabilities of solrj.io.

This blog introduces the ParallelStream which sends a TupleStream to worker nodes where it can be processed in parallel. Both streaming transformations and streaming aggregations can be parallelized using the ParallelStream. This blog focuses on an example of a parallel streaming transformation. Future blogs will provide examples of parallel streaming aggregation.

Worker Collections

Parallel streaming operations are handled by Solr nodes within a worker collection. Worker collections are SolrCloud collections that have been configured to handle streaming operations. The only requirement for worker collections is that the Solr nodes in the collection must be configured with a StreamHandler to handle the streaming requests.

Worker collections can hold an index with data or they can be empty collections that only exist to handle streaming operations.

Stream Partitioning

Each worker node is shuffled 1/Nth of the search results. So if there are 7 worker nodes specified each node will be shuffled 1/7th of the search results. The partitioning is setup under the covers by the StreamHandler on the worker nodes.

Search results are hash partitioned by keys. This ensures that all results with the same key(s) are sent to the same worker node. When coupled with the ability to sort by keys, hash partitioning provides shuffling functionality that allows many streaming operations to be parallelized.

ParallelStream

A TupleStream decorator called the ParallelStream wraps a TupleStream and sends it to N worker nodes to be operated on in parallel.

The ParallelStream serializes the byte code of the underlying TupleStream and sends it to the workers before the TupleStream is opened. The workers deserialize the TupleStream, open the stream, read the Tuples and stream them back to the ParallelStream.

The TupleStream sent to the workers can be programmed to stream only aggregated results or the top N results back to the ParallelStream.

UniqueStream

The example below also introduces a TupleStream decorator called the UniqueStream. The UniqueStream emits a unique stream of Tuples. Tuples are de-duplicated based on a Comparator.

The example will show how the UniqueStream can be used with the ParallelStream to perform the unique operation in parallel.

Simple Parallel Streaming Transformation Example

org.apache.solr.client.solrj.io.*;
import java.util.*;

public class StreamingClient {

   public static void main(String args[]) throws IOException {
      String zkHost = args[0];
      String collection = args[1];
      String workerCollection = args[2];

      Map props = new HashMap();
      props.put("q", "*:*");
      props.put("qt", "/export");
      props.put("sort", "fieldA asc");
      props.put("fl", "fieldA");

      //Set the partition keys
      props.put("partitionKeys", "fieldA");
      
      CloudSolrStream streamC = new CloudSolrStream(zkHost, 
                                                    collection, 
                                                    props);
      Comparator comp = new AscFieldComp("fieldA");
      UniqueStream streamU = new UniqueStream(streamC, comp);
       
      ParallelStream streamP = new ParallelStream(zkHost,
                                                  workerCollection,
                                                  streamU,
                                                  7,
                                                  comp);
                             
      try {
       
        streamP.open();
        while(true) {
          
          Tuple tuple = streamP.read();
          if(tuple.EOF) {
             break;
          }

          String uniqueValue = tuple.getString("fieldA");
          //Print all the unique values
          System.out.println(uniqueValue);

      } finally {
        streamP.close();
      }
   }
}

The example above does the following things:

1. Creates a CloudSolrStream that connects to a SolrCloud collection.
2. In the query parameters the sort is set to fieldA asc. This will return the Tuples in ascending order based on fieldA.
3. The query parameters also set the partitionKeys to fieldA. This will hash partition the search results on fieldA.
4. The CloudSolrStream is then decorated with a UniqueStream. Note the Comparator which is used by the UniqueStream. In this case the AscFieldComp will cause the UniqueStream to emit unique Tuples based on the value in fieldA. The sorting and partitioning of search results on fieldA allows the UniqueStreams operation to be run in parallel on the worker nodes.
5. The UniqueStream is decorated with a ParallelStream. Note that the ParallelStream is constructed with the following parameters:
   • zkHost
   • The name of the workerCollection
   • The TupleStream that is being sent to the workers
   • The number of workers
   • A Comparator to order the results coming back from the workers
6. Then the ParalleStream is opened, iterated and closed.
7. Under the covers the ParallelStream serializes the UniqueStream and sends it to 7 worker nodes. The workers open the UniqueStream, read the unique Tuples and stream them back to the ParallelStream.