Indexing data in Solr from disparate sources using Camel

Apache Solr is "the popular, blazing fast open source enterprise search platform" built on top of Lucene.  In order to do a search (and find results) there is the initial requirement of data ingestion usually from disparate sources like content management systems, relational databases, legacy systems, you name it... Then there is also the challenge of keeping the index up to date by adding new data, updating existing records, removing obsolete data. The new sources of data could be the same as the initial ones, but could also be sources like twitter, AWS or rest endpoints.

Solr can understand different file formats and provides fair amount of options for data indexing:

1. Direct HTTP and remote streaming - allows you to interact with Solr over HTTP by posting a file for direct indexing or the path to the file for remote streaming.
2. DataImportHandler - is a module that enables both full and incremental delta imports from relational databases or file system.
3. SolrJ - a java client to access Solr using Apache Commons HTTP Client.

But in real life, indexing data from different sources with millions of documents, dozens of transformations, filtering, content enriching, replication, parallel processing  requires much more than that. One way to cope with such a challenge is by reinventing the wheel: write few custom applications, combine them with some scripts or run cronjobs. Another approach would be to use a tool that is flexible and designed to be configurable and plugable, that can help you to scale and distribute the load with ease. Such a tool is Apache Camel which has also a Solr connector now.

All started few months ago, during basecamp days at Sourcesense, where me and my colleague Alex were experimenting with different projects to implement a pipeline for indexing data into Solr. As expected we discovered Camel and after few days of pairing, we were ready with the initial version of the Solr component which got committed to Camel and extended further by Ben Oday. At the moment it is full featured Solr connector, that uses SolrJ behind the scene and lets you to: configure all parameters of SolrServer and StreamingUpdateSolrServer;  supports the operations: insert, add_bean, delete_by_id, delete_by_query, commit, rolback, optimize; index files, SolrInputDocument instances, beans with annotations or individual message headers.

Creating a Camel route to index all the data from a relational database table and local file system is simple:

public void configure() {
from("timer://clear?repeatCount=1")
        .to("direct:clearIndex");

from("file:src/data?noop=true")
        .to("direct:insert");

from("timer://database?repeatCount=1")
        .to("sql:select * from products?dataSourceRef=productDataSource")
        .split(body())
        .process(new SqlToSolrMapper())
        .to("direct:insert");

from("direct:insert")
        .setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_INSERT))
        .to(SOLR_URL)
        .setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_COMMIT))
        .to(SOLR_URL);

from("direct:clearIndex")
        .setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_DELETE_BY_QUERY))
        .setBody(constant("*:*"))
        .to(SOLR_URL)
        .setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_COMMIT))
        .to(SOLR_URL);
}

The above route will first clear the index by deleting all the documents followed by a commit. Then it will start polling files from src/data folder, read each file and send it to Solr endpoint. Assuming that the files are in a format Solr can understand, they will be indexed and committed. The third route will retrieve all the products from database (in memory), split them into individual records, map each record to Solr fields, and digest :)

Luckily, in 2012, the life of software developer is not that simple boring. Instead nowadays a more realistic indexing requirement would consist of something like this:

1. Get the backup files from amazon S3 and index. If a document is approved, commit it as soon as  possible, otherwise commit every 10 minutes.

How can Camel help you with this requirement? Camel supports most popular amazon APIs including S3. Using aws-s3 component, it is possible to read files from a S3 bucket, then apply a filter for approved documents, in order to send them into a separate route for instant commit.

<route>
  <from uri="aws-s3://MyBucket?delay=5000&maxMessagesPerPoll=5"/>
  <choice>
    <when>
      <xpath>/add/doc[@status='approved']</xpath>
      <to uri="direct:indexAndCommit"/>
    </when>
    <otherwise>
      <to uri="direct:index"/>
    </otherwise>
  </choice>
</route>
<route>
  <from uri="timer://commit?fixedRate=true&period=600s"/>
  <from uri="direct:commit"/>
</route>

2. Retrieve customer data from database every 5 seconds by reading10 records at a time. Also look for deltas. Enrich the address data with latitute/longitute by calling XXX external service to facilitate spatial search in Solr.

<route id="fromDB">
  <from uri="jpa://com.ofbizian.pipeline.Customer?consumer.namedQuery= newCustomers&amp;maximumResults=10&amp;delay=5000"/>
  <enrich uri="direct:coordinateEnricher" strategyRef="latLongAggregationStrategy"/>
  <to uri="direct:index"/>
</route>

<route>
  <from uri="direct:coordinateEnricher"/>
  <setHeader headerName="CamelHttpQuery">
    <simple>address='${body.address}'&amp;sensor=false</simple>
  </setHeader>
  <to uri="http://maps.google.com/maps/api/geocode/xml"/>
  <setHeader headerName="lat">
    <xpath resultType="java.lang.Double">//result[1]/geometry/location/lat/text()</xpath>
  </setHeader>
  <setHeader headerName="lng">
    <xpath resultType="java.lang.Double">//result[1]/geometry/location/lng/text()</xpath>
  </setHeader>
</route>

The above route reads from Customer table 10 records at a time, and for each one will call google's maps API to get latitude and longitude using the customer address field. The coordinates are extracted from response using XPath and merged back into Customer object. Simple, isn't it.

3. Index the content under this/that/path in our content management system and also monitor for updates.

<route>
  <from uri="jcr://user:pass@repository/import/inbox/signal?eventTypes=3&deep=true&synchronous=false"/>
  <to uri="direct:index"/>
</route>

Camel has a jcr connector, which allows you to create content in any java content repository. There is also an improvement submitted in CAMEL-5155 which will allow reading content from JCR v.2 supporting repositories soon.
If you are lucky and your CMS supports CMIS you can use my camel-cmis connector from github for the same purpose.

4. Listen for tweets about our product/company, do sentiment analysis, and index only positive tweets.

<route id="fromTwitter">
  <from uri="twitter://streaming/filter?type=event&keywords=productName&consumerKey={{consumer.key}}&consumerSecret={{consumer.secret}}"/>
  <setHeader headerName="CamelHttpQuery">
    <language language="beanshell">
      "q=" + java.net.URLEncoder.encode(request.getBody().getText(), "UTF-8")
    </language>
  </setHeader>
  <throttle timePeriodMillis="1500">
    <constant>1</constant>
    <to uri="http://data.tweetsentiments.com:8080/api/analyze.xml"/>
    <setHeader headerName="sentiment">
      <xpath resultType="java.lang.Double">/sentiment/value/text()</xpath>
    </setHeader>
    <filter>
      <simple>${in.header.sentiment} > 0</simple>
      <to uri="direct:index"/>
    </filter>
  </throttle>
</route>

This route is going to listen for tweets using twitter's real time api, url encode the tweet and call tweetsentiments api for sentiment analysis. In addition it will apply throttling, so only one request at most is made every 1500 milliseconds, because there is restriction on the number of calls per second. Then the route is applying a filter to ignore all the negative tweets, before indexing.

As you can see Camel can interact with many disparate systems (including Solr) easily, and even if you have a very custom application, writing a connector for it would not be difficult. But this is only one side of the story. At the other side, there is a full list of Enterprise Integration Patterns implemented by Camel which are needed for any serious data ingestion pipeline: Routers, Translator, Filter, Splitter, Aggregator, Content Enricher, Load Balancer... Last but not least: Exception Handling, Logging, Monitoring, DSLs... In two words: Camel Rocks!

PS: The full source code of the examples can be found on my github account.

My first commit to Camel - FOP component

After being invited to join the Apache Camel team, it is time for my first commit. It is actually for one of my very first contributions to Camel that was waiting in the jira for long time - a camel-fop component.
This component uses Apache FOP project to render messages into variety of output formats such as PDF, PostScript, RTF, PNG etc.

Camel-fop has only a producer, which expects messages containing XML data in XSL-FO format in the body. Assuming that your data and representation are separate, xsl-fo can be generated from xml data and xslt template, or instead using freemarker or velocity templates and passing the data with message headers.
Once the content is ready in the message body, camel-fop will transform it to the output format specified in the endpoint URL or message header.

It is also possible to set some metadata for each document (like producer, author, creation date, resolution...) or encrypt the document with password.

Additionally it allows you to override the default fop configuration file by specifying a new userConfig location in the endpoint URL. With a custom configuration you can control fop's rendering behavior and help it find its resources.

Here is an example route that creates PDFs from xml data and xslt template:

      .from("file:source/data/xml")

          .to("xslt:xslt/template.xsl")

          .to("fop:application/pdf")

          .to("file:target/data");

To see the example files in actions, check the test folder in camel-fop component.
Let me know if you find this component any useful.

REST with Apache Camel

There are many ways to expose an HTTP endpoint in Camel: jetty, tomcat, servlet, cxfrs and restlet. Two of these components - cxfrs and restlet also support REST semantics just with few lines of code. This simple example demonstrates how to do CRUD operations with camel-restlet and camel-jdbc.

The four HTTP verbs execute different operations and map to the following single URI template:

•  POST - create a new user: /user
•  GET - request the current state of the user specified by the URI: /user/{userId}
•  PUT - update an user at the given URI with new information: /user/{userId}
•  DELETE - remove the user identified by the given URI: /user/{userId}

• There is also a /users URI which returns all the users regardless of the HTTP method used.

Creating such an application with Camel is straightforward. After adding all the necessary dependencies (restlet, spring, jdbc...) configure web.xml to load Camel context:

<context-param>
  <param-name>contextConfigLocation</param-name>
  <param-value>classpath:camel-config.xml</param-value>
</context-param>
<listener>
  <listener-class>org.springframework.web.context.ContextLoaderListener   </listener-class>
</listener>

  and map the Restlet servlet

<servlet>
  <servlet-name>RestletServlet</servlet-name>
  <servlet-class>org.restlet.ext.spring.SpringServerServlet</servlet-class>
  <init-param>
    <param-name>org.restlet.component</param-name>
    <param-value>RestletComponent</param-value>
  </init-param>
</servlet>
<servlet-mapping>
  <servlet-name>RestletServlet</servlet-name>
  <url-pattern>/rs/*</url-pattern>
</servlet-mapping>

In the Spring context, there is a little more Restlet and an in-memory datasource setup code:

<bean id="RestletComponent" class="org.restlet.Component"/>
<bean id="RestletComponentService" class="org.apache.camel.component.restlet.RestletComponent">
   <constructor-arg index="0">
     <ref bean="RestletComponent"/>
   </constructor-arg>
</bean>
<jdbc:embedded-database id="dataSource" type="HSQL">
   <jdbc:script location="classpath:sql/init.sql"/>
</jdbc:embedded-database>

After all the setup is done, the next step is to create Camel routes that will process the HTTP requests and execute appropriate CRUD operations. The first one is createUser route that executes SQL insert command with the parameters from POST requests only and return the newly created user in the response body:

<route id="createUser">
   <from uri="restlet:/user?restletMethod=POST"/>
   <setBody>
     <simple>insert into user(firstName, lastName) values('${header.firstName}','${header.lastName}');  </simple>
   </setBody>
   <to uri="jdbc:dataSource"/>
   <setBody>
     <simple>select * from user ORDER BY id desc LIMIT 1</simple>
   </setBody>
   <to uri="jdbc:dataSource"/>
</route>

The "manipulateUser" route handles GET, PUT and DELETE HTTP methods, but depending on the method used, it executes different SQL commands:

<route id="manipulateUser">

  <from uri="restlet:/user/{userId}?restletMethods=GET,PUT,DELETE"/>

  <choice>

    <when>

    <simple>${header.CamelHttpMethod} == 'GET'</simple>

    <setBody>

      <simple>select * from user where id = ${header.userId}</simple>

    </setBody>

   </when>

   <when>

     <simple>${header.CamelHttpMethod} == 'PUT'</simple>

       <setBody>

       <simple>update user set firstName='${header.firstName}', lastName='${header.lastName}' where id = ${header.userId}</simple>

       </setBody>

   </when>

   <when>

     <simple>${header.CamelHttpMethod} == 'DELETE'</simple>

     <setBody>

       <simple>delete from user where id = ${header.userId}</simple>

     </setBody>

   </when>

   <otherwise>

     <stop/>

   </otherwise>

  </choice>

  <to uri="jdbc:dataSource"/>

</route>

And the last route for listing all the users is self explanatory:

<route id="listUsers">
  <from uri="restlet:/users"/>
  <setBody>
    <constant>select * from user</constant>
  </setBody>
  <to uri="jdbc:dataSource"/>
</route>

If you want to see the application in action, grab the source code from github and run it with the embedded maven-jetty plugin by typing: mvn jetty:run .You can even try some quick queries if you have curl installed:

To create an user, make a http POST request with firstName and lastName parameters
curl -d "firstName=test&lastName=user" http://localhost:8080/rs/user/

To update an existing user, make a http PUT request with firstName and lastName parameters
curl -X PUT -d "firstName=updated&lastName=user" http://localhost:8080/rs/user/2

To retrieve an existing user, make a http GET request with the userId as part of the url
curl -X GET  http://localhost:8080/rs/user/2

To delete an existing user, make a http DELETE request with the userId as part of the url
curl -X DELETE  http://localhost:8080/rs/user/2

To retrieve all the existing users, make a http GET request to users url
curl -X GET  http://localhost:8080/rs/users

First blog post

First a failing post :)