SolrCloud

Note* optimistic locking currently not fully implemented yet

SolrCloud

SolrCloud is the name of a set of new distributed capabilities in Solr. Passing parameters to enable these capabilities will enable you to set up a highly available, fault tolerant cluster of Solr servers.

Getting Started

Check out and build Solr trunk: https://svn.apache.org/repos/asf/lucene/dev/trunk and build the example server with cd solr; ant example.

If you haven't yet, go through the simple Solr Tutorial to familiarize yourself with Solr.

Solr embeds and uses Zookeeper as a repository for cluster configuration and coordination - think of it as a distributed filesystem that contains information about all of the Solr servers.

Example A: Simple two shard cluster

This example simply creates a cluster consisting of two solr servers representing two different shards of a collection.

Since we'll need two solr servers for this example, simply make a copy of the example directory for the second server.

cp -r example example2

This command starts up a Solr server and bootstraps a new solr cluster.

cd example
java -Dbootstrap_confdir=./solr/conf -Dcollection.configName=myconf -DzkRun -DnumShards=2 -jar start.jar

• -DzkRun causes an embedded zookeeper server to be run as part of this Solr server.

• -Dbootstrap_confdir=./solr/conf Since we don't yet have a config in zookeeper, this parameter causes the local configuration directory ./solr/conf to be uploaded as the "myconf" config. The name "myconf" is taken from the "collection.configName" param below.

• -Dcollection.configName=myconf sets the config to use for the new collection. Omitting this param will cause the config name to default to "configuration1".

• -DnumShards=2 the number of logical partitions we plan on splitting the index into.

Browse to http://localhost:8983/solr/collection1/admin/zookeeper.jsp to see the state of the cluster (the zookeeper distributed filesystem).

NOTE: There is a new admin UI that is a work-in-progress that is also available: http://localhost:8983/solr/#/cloud (make sure to expand the rather lonely-looking root node in the initial display!). This is a work in progress, so please feel free to try it and provide feedback/enhancements. One known limitation is that each of the leaf nodes in the old UI links to the detail information, this is not yet implemented in the new UI (any help greatly appreciated!).

You can see from the zookeeper browser that the Solr configuration files were uploaded under "myconf", and that a new document collection called "collection1" was created. Under collection1 is a list of shards, the pieces that make up the complete collection.

Now we want to start up our second server - it will automatically be assigned to shard2 because we don't explicitly set the shard id.

Then start the second server, pointing it at the cluster:

cd example2
java -Djetty.port=7574 -DzkHost=localhost:9983 -jar start.jar

• -Djetty.port=7574 is just one way to tell the Jetty servlet container to use a different port.

• -DzkHost=localhost:9983 points to the Zookeeper ensemble containing the cluster state. In this example we're running a single Zookeeper server embedded in the first Solr server. By default, an embedded Zookeeper server runs at the Solr port plus 1000, so 9983.

If you refresh the zookeeper browser, you should now see both shard1 and shard2 in collection1. View here if you are just reading along.

Next, index some documents. If you want to whip up some Java you can use the CloudSolrServer solrj impl and simply init it with the address to ZooKeeper. Or simply randomly choose which instance to add documents too - they will be automatically forwarded to where they belong:

cd exampledocs
java -Durl=http://localhost:8983/solr/collection1/update -jar post.jar ipod_video.xml
java -Durl=http://localhost:8983/solr/collection1/update -jar post.jar monitor.xml
java -Durl=http://localhost:8983/solr/collection1/update -jar post.jar mem.xml

And now, a request to either server results in a distributed search that covers the entire collection:

http://localhost:8983/solr/collection1/select?q=*:*

If at any point you wish to start over fresh or experiment with different configurations, you can delete all of the cloud state contained within zookeeper by simply deleting the solr/zoo_data directory after shutting down the servers.

Example B: Simple two shard cluster with shard replicas

This example will simply build off of the previous example by creating another copy of shard1 and shard2. Extra shard copies can be used for high availability and fault tolerance, or simply for increasing the query capacity of the cluster.

First, run through the previous example so we already have two shards and some documents indexed into each. Then simply make a copy of those two servers:

cp -r example exampleB
cp -r example2 example2B

Then start the two new servers on different ports, each in its own window:

cd exampleB
java -Djetty.port=8900 -DzkHost=localhost:9983 -jar start.jar

cd example2B
java -Djetty.port=7500 -DzkHost=localhost:9983 -jar start.jar

Refresh the zookeeper browser page Solr Zookeeper Admin UI or new Solr Zookeeper Admin UI and verify that 4 solr nodes are up, and that each shard is present at 2 nodes.

Because we have been telling Solr that we want two logical shards, starting instances 3 and 4 are assigned to be replicas of instances one and two automatically.

Now send a query to any of the servers to query the cluster:

http://localhost:7500/solr/collection1/select?q=*:*

Send this query multiple times and observe the logs from the solr servers. From your web browser, you may need to hold down CTRL while clicking on the browser refresh button to bypass the HTTP caching in your browser. You should be able to observe Solr load balancing the requests (done via LBHttpSolrServer ?) across shard replicas, using different servers to satisfy each request. There will be a log statement for the top-level request in the server the browser sends the request to, and then a log statement for each sub-request that are merged to produce the complete response.

To demonstrate fail over for high availability, go ahead and kill any one of the Solr servers (just press CTRL-C in the window running the server) and and send another query request to any of the remaining servers that are up.

SolrCloud uses leaders and an overseer as an implementation detail. This means that some shards/replicas will play special roles. You don't need to worry if the instance you kill is a leader or the cluster overseer - if you happen to kill one of these, automatic fail over will choose new leaders or a new overseer transparently to the user and they will seamlessly takeover their respective jobs. Any Solr instance can be promoted to one of these roles.

Example C: Two shard cluster with shard replicas and zookeeper ensemble

The problem with example B is that while there are enough Solr servers to survive any one of them crashing, there is only one zookeeper server that contains the state of the cluster. If that zookeeper server crashes, distributed queries will still work since the solr servers remember the state of the cluster last reported by zookeeper. The problem is that no new servers or clients will be able to discover the cluster state, and no changes to the cluster state will be possible.

Running multiple zookeeper servers in concert (a zookeeper ensemble) allows for high availability of the zookeeper service. Every zookeeper server needs to know about every other zookeeper server in the ensemble, and a majority of servers are needed to provide service. For example, a zookeeper ensemble of 3 servers allows any one to fail with the remaining 2 constituting a majority to continue providing service. 5 zookeeper servers are needed to allow for the failure of up to 2 servers at a time.

For production, it's recommended that you run an external zookeeper ensemble rather than having Solr run embedded zookeeper servers. You can read more about setting up a zookeeper ensemble here. For this example, we'll use the embedded servers for simplicity.

First, stop all 4 servers and then clean up the zookeeper data directories for a fresh start.

rm -r example*/solr/zoo_data

We will be running the servers again at ports 8983,7574,8900,7500. The default is to run an embedded zookeeper server at hostPort+1000, so if we run an embedded zookeeper on the first three servers, the ensemble address will be localhost:9983,localhost:8574,localhost:9900.

As a convenience, we'll have the first server upload the solr config to the cluster. You will notice it block until you have actually started the second server. This is due to zookeeper needing a quorum before it can operate.

cd example
java -Dbootstrap_confdir=./solr/conf -Dcollection.configName=myconf -DzkRun -DzkHost=localhost:9983,localhost:8574,localhost:9900 -DnumShards=2 -jar start.jar

cd example2
java -Djetty.port=7574 -DzkRun -DzkHost=localhost:9983,localhost:8574,localhost:9900 -jar start.jar

cd exampleB
java -Djetty.port=8900 -DzkRun -DzkHost=localhost:9983,localhost:8574,localhost:9900 -jar start.jar

cd example2B
java -Djetty.port=7500 -DzkHost=localhost:9983,localhost:8574,localhost:9900 -jar start.jar

Now since we are running three embedded zookeeper servers as an ensemble, everything can keep working even if a server is lost. To demonstrate this, kill the exampleB server by pressing CTRL+C in it's window and then browse to Solr Zookeeper Admin UI or new Solr Zookeeper Admin UI to verify that the zookeeper service still works.

ZooKeeper

Multiple Zookeeper servers running together for fault tolerance and high availability is called an ensemble. For production, it's recommended that you run an external zookeeper ensemble rather than having Solr run embedded servers. See the Apache ZooKeeper site for more information on downloading and running a zookeeper ensemble. More specifically, try Getting Started and ZooKeeper Admin. It's actually pretty simple to get going. You can stick to having Solr run ZooKeeper, but keep in mind that a ZooKeeper cluster is not easily changed dynamically. Until further support is added to ZooKeeper, changes are best done with rolling restarts. Handling this in a separate process from Solr will usually be preferable.

In terms of trying to make sure ZooKeeper is setup to be very fast, keep a few things in mind: Solr does not use ZooKeeper intensively - optimization's are likely not necessary. Also, while adding more ZooKeeper nodes will help some with read performance, it will slightly hurt write performance. Again, Solr does not really do much with ZooKeeper when your cluster is in a steady state.

When Solr runs an embedded zookeeper server, it defaults to using the solr port plus 1000 for the zookeeper client port. In addition, it defaults to adding one to the client port for the zookeeper server port, and two for the zookeeper leader election port. So in the first example with Solr running at 8983, the embedded zookeeper server used port 9983 for the client port and 9984,9985 for the server ports.

Creating cores via CoreAdmin

New Solr cores may also be created and associated with a collection via CoreAdmin.

Additional cloud related parameters for the CREATE action:

• collection - the name of the collection this core belongs to. Default is the name of the core.

• shard - the shard id this core represents (Optional - normally you want to be auto assigned a shard id)

• collection.<param>=<value> - causes a property of <param>=<value> to be set if a new collection is being created.

  • Use collection.configName=<configname> to point to the config for a new collection.

Example:

curl 'http://localhost:8983/solr/admin/cores?action=CREATE&name=mycore&collection=collection1&shard=shard2'

Distributed Requests

Explicitly specify the addresses of shards you want to query:

shards=localhost:8983/solr,localhost:7574/solr

Explicitly specify the addresses of shards you want to query, giving alternatives (delimited by |) used for load balancing and fail-over:

shards=localhost:8983/solr|localhost:8900/solr,localhost:7574/solr|localhost:7500/solr

Query all shards of a collection (the collection is implicit in the URL):

http://localhost:8983/solr/collection1/select?

Query specific shard ids. In this example, the user has partitioned the index by date, creating a new shard every month.

http://localhost:8983/solr/collection1/select?shards=shard_200812,shard_200912,shard_201001

Query all shards of a compatible collection, explicitly specified:

http://localhost:8983/solr/collection1/select?collection=collection1_recent

Query all shards of multiple compatible collections, explicitly specified:

http://localhost:8983/solr/collection1/select?collection=collection1_NY,collection1_NJ,collection1_CT

Required Config

All of the required config is already setup in the example configs shipped with Solr.

schema.xml

You must have a _version_ field defined:

<field name="_version_" type="long" indexed="true" stored="true"/>

solrconfig.xml

You must have an UpdateLog defined; the sample puts this in the DirectUpdateHandler2 updateHandler element.

    <!-- Enables a transaction log, currently used for real-time get.
         "dir" - the target directory for transaction logs, defaults to the
         solr data directory.  --> 
    <updateLog>
      <str name="dir">${solr.data.dir:}</str>
    </updateLog>

You must have a replication handler called /replication defined:

    <requestHandler name="/replication" class="solr.ReplicationHandler" startup="lazy" /> 

You must have a realtime get handler called /get defined:

    <requestHandler name="/get" class="solr.RealTimeGetHandler">
      <lst name="defaults">
        <str name="omitHeader">true</str>
     </lst>
    </requestHandler>

You must have the admin handlers defined:

    <requestHandler name="/admin/" class="solr.admin.AdminHandlers" />

And you must leave the admin path in solr.xml as the default:

    <cores adminPath="/admin/cores"

The DistributedUpdateProcessor is part of the default update chain, but if you define a custom update chain you would like to use with SolrCloud, you must put the DistributedUpdateProcessor in that chain:

   <updateRequestProcessorChain name="sample">
     <processor class="solr.LogUpdateProcessorFactory" />
     <processor class="solr.DistributedUpdateProcessorFactory"/>
     <processor class="my.package.UpdateFactory"/>
     <processor class="solr.RunUpdateProcessorFactory" />
   </updateRequestProcessorChain>

Re-sizing a Cluster

You can control cluster size by passing the numShards when you start up the first SolrCore in a collection. This parameter is used to auto assign which shard each instance should be part of. Any SolrCores that you start after starting numShards instances are evenly added to each shard as replicas (as long as they all belong to the same collection).

To add more SolrCores to your collection, simply keep starting new SolrCores up. You can do this at any time and the new SolrCore will sync up its data with the current replicas in the shard before becoming active.

You can also avoid numShards and manually assign a SolrCore a shard id if you choose.

You can not yet change the number of shards of the index after initially setting up the cluster - eg you want to turn a shard1,shard2 cluster into a shard1,shard2,shard3 cluster. This is a feature that is on the road-map however.

If you want to start your cluster on fewer machines and then expand over time beyond just adding replicas, you can choose to start by hosting multiple shards per machine (using multiple SolrCores) and then later migrate shards onto new machines by starting up a new replica for a given shard and eventually removing the shard from the original machine.

Near Realtime Search

If you want to use the Near Realtime search support, you will probably want to enable auto soft commits in your solrconfig.xml file before putting it into zookeeper. Otherwise you can send explicit soft commits to the cluster as you desire. See NearRealtimeSearch

Parameter Reference

Cluster Params

Node Params

ZooKeeper Params

Config Bootstrap Params

Other