LinkedIn Search Talk - SDForum

The past Wednesday I had the pleasure of giving a technical talk at SDForum on LinkedIn Search.

This talk came about a month after the 100% rollout of LinkedIn Faceted People Search. See blog by Esteban Kozak.

In this talk, I talked about the LinkedIn search infrastructure that hosts various LinkedIn search properties, e.g. people search, news search, job search etc.

The main features we built are:

• realtime indexing/search
• streaming/live update
• faceted navigation
• section search
• distributed index partitioning

The slides provide a glance through how we built these technologies through the following open source projects we built/working on:

• Zoie - realtime indexing update system
• Bobo - faceted search engine based on Lucene.
• Sensei - distributed realtime faceted search system.

Some of the notable attendees are:

• Daniel Tunkelang - Former Chief-Scientist at Endeca - enterprise search solution with faceted search emphasis, and author of the blog, TheNoisyChannel.
• Stefan Groschupf - Creator of the popular distributed search open source project: Katta.

along with representations from companies such as Google, Apple and VMWare etc.

I am glad to have learned different uses for search technology and hope the technologies we have built to be helpful in different areas.

For learn about our team at LinkedIn and see other open source projects we are working on, visit http://sna-projects.com/.

Index Optimization for realtime search - Good idea?

Overview of Optimize:
There is a Lucene API: IndexWriter.optimize(), which combines all segments into 1 large segment and also expunges all deleted docids.

Searching over an optimized index is very fast because you neither pay penalty to evaluate deleted docs nor search time OR'ing over documents in different indexes. After some OS level warming, the 1 segment file is loaded into the IO cache to avoid IO costs. Hence the method name: optimize(). This is terrific for an offline indexing system, where a pristine optimized index is published for searching.

Segment merge:
Segment merge is essential incremental indexing. Lucene has a "hidden" extensible API: MergePolicy. (properly exposed in Lucene 2.9) By default, LogByteSizeMergePolicy is used. This policy will periodically choose to merge small segments into larger segments, and size of the segment is based on number of bytes of the segment file. Only during a merge, deleted docs are expunged.

Real-time indexing:
In a real-time indexing environment, indexing operations are being applied to the index constantly, and the index is fragmented quickly. A challenge here is how to maintain an optimal index for real-time indexing.

In our application, where there are many updates, e.g. old documents are deleted and then added with newer/fresher data. What happened was over time, the largest segment would contain more and more deleted docs, and they will never be expunged because the segment is never a candidate of a merge since deleted docs are merely marked, not removed from the segment, thus the segment size still remain to be large. In the worse scenario, the largest segment would contain only deleted docs.

We made an enhancement to LogMergePolicy to normalize on size taking into consideration number of deleted documents (and contributed back: LUCENE-1634)

This helped quite a bit. We still however, see the problem with the situation when segments are promoted so that they get the merge with the largest segment:

In a realtime scenario, when smaller segments are "escalated" to be merged with the larger segment, the search response time also escalates. This is because the merge itself gets more expensive as the sizes of the segments to be merged get larger. Furthermore, the newly merged segment needs to be loaded into IO cache, while that is happening, search time is impacted significantly.

To solve this problem, we have created a new MergePolicy implementation:

Idea:

Instead of defining an optimized index to be 1 large segment, we redefine it to be N segments of balanced size, where N is a configurable parameter. The idea is to spread the cost of a large segment merge into smaller merge costs.

Implementation:

At each point of the merge operation, the segments to merge is selected to main a balanced segment structure. The selection is modeled as a state and a merge is viewed as a transition between states, and each such transition is associated with a cost function of merge. We then applied the Viterbi algorithm to identify the optimal selection(s).

Performance numbers and details be found at this wiki.

Our MergePolicy implementation has also been contributed back to Lucene: LUCENE-1924

Conclusion:

In conclusion, I would like to emphasize how indexing can affect search performance especially in real-time search. There are often hidden problems as they are invisible to unit tests and simple performance tests. They can also be data dependent and show up after hours or even days of stressing the system. Thus, it is important to understand the details of the indexing to build a scalable and robust system.

Credit:

I'd like to credit this idea and implementation to my colleague Yasuhiro Matsuda.

Index Partitioning and Distributed Realtime Search

When the number of documents gets to be very large in one single index, it is usually a good idea to partition the index and distribute the search load to different processes or search nodes, each serving from a section of the corpus.


Some Background:

There are many different ways of partitioning the index depending on the application requirements, here are some examples.

• By terms - This is good because not every search node is handling every request. Only the search nodes containing terms from the query are called to serve the request. However, normally some terms are searched more than others, and on top of that, some terms contain more documents than others. Therefore it is usually very difficult to load balance this partitioning scheme.
• By documents - This is the most common partitioning method. Even though every search node is handling every search request, (assuming documents are uniformly distributed across the partitions,) it is very easy to balance and re-balance search traffic.
• By time - This is a good idea in corpus containing time sensitive data, e.g. job postings, news, blogs etc. In such cases, recent documents are by definition more relevant than older documents. Thus, we can add more replication on partitions with recent documents while slowly discard older partitions from the index.

Our Partitioning Scheme:


We, at LinkedIn, are in a real-time search situation while guaranteeing high availability, we want to have a partitioning scheme that would allows us to avoid re-indexing/re-partitioning as our corpus grows.

We have decided to choose partition by ranges of documents. We choose the maximum number of documents in a partition, say N. And partition along consecutive document IDs (UIDs, not Lucene docids), .e.g. partition K, would containing documents ranging from K*N to (K+1)*N-1, K = 0,1,2,...

As new documents are added to the index, we grow the partition by adding to the largest partition, and when the UID exceeds the maximum document ID on the largest partition, a new partition is created. This partitioning scheme allows us to grow to newer partitions without a "re-partition".


Cross domain "joins":

Another important benefit we enjoy with this partitioning scheme is that we can load Lucene docid to UID mapping (see my previous post) as well as the reverse: UID mapping to Lucene docid mapping, because we know the range UIDs fall in. We are thus very easily able to translate a filter set in the UID space to a Lucene Filter or a Lucene Query. This is important because it essentially allows us to do "joins" across different document spaces. The memory footprint for holding both lookup arrays are docCount*4 bytes each, nowadays, it's a piece of cake.


Realtime at the Node-Level:

To guarantee realtime on the search node level, each of are partitions are powered by Zoie. See my earlier post)