Using Hadoop to Create SOLR Indexes

One of the most challenging projects I faced at work recently was to create a Apache SOLR index consisting of approx 15 million records. This index had been created once in the history of the company using a MySQL database and SOLR’s Data Import Handler (DIH). It had not been attempted since then because the original indexing process was time consuming (12-14 hours), required human supervision, and on failure had to be restarted from the very beginning.

For small data sets (say, less than 100,000 records) SOLR’s DIH and MySQL works fine. However, for these larger sets it’s just too much of a drain on resources.  Some members of our team and the architecture team had had success working with large data sets by leveraging the Apache Hadoop project. One of the most attractive aspect of Hadoop is that the processing is distributed which should reduce the total time to index. Also Hadoop has a robust fail-over system which would remove the need for human supervision. We architected a data pipeline by which data would be processed by modules. When one module completed its task it would alert the system and the next module would begin work on the the output of the previous module. The SOLR indexing is one module.

Many of the modules, including the SOLR indexing module, use the Hadoop Map/Reduce programming model to process the data over a distributed system. Having never used SOLR in this type of environment I had questions:

• Can SOLR access the Hadoop file system?
• Can SOLR indexing be applied in a distributed fashion?
• Can we control where SOLR would write its indexes?

Can SOLR access the Hadoop file system?

It turns out that in its current state SOLR must read/write to a local file system.  There may be a way around this but we were fighting against time to prove that the pipeline is a viable (and cost effective) system. As such, there would have to be a mix of local file system and HDFS.

Can SOLR indexing be applied in a distributed fashion?

The mapping can but not the reducing (at least not in this first version).  As the mapper is fed data those data are used to generate an index. The indexes generated by the mappers are a subset of the overall data set. During the cleanup process all of the smaller indexes generated by the mapper are copied to the HDFS.

Since SOLR has to read/write to a local file system we could not distribute the reducer part of the process. The number of reducers was limited to 1 (using the setNumReduceTasks method in org.apache.hadoop.mapreduce.Job).  The reducer simply copied all the smaller indexes from the HDFS to a node’s local file system. On cleanup the smaller indexes were merged into one large index that would be the final product (using org.apache.lucene.index.IndexWriter).

Challenges

The first time we ran the indexer job we found that Hadoop would not allow the reducer to complete. Repeatedly Hadoop would start a task and kill it after exactly 10 minutes.  If the Hadoop sees a task being idle for a certain amount of time it assumes something is wrong, kills the task, and starts another. The code did send Hadoop progress updates (using Context.progress()). However, the merge process performed by the IndexWriter takes longer than the configured time out (the default value for mapred.task.timeout is 600000 milliseconds). There is no way to put the merge process on hold to give Hadoop a status update. Once IndexWriter has been told to merge it goes off and returns only when it is done. The short term solution to this was to increase the time-out from the default 600000 milliseconds to 2 hours.

Another issue had to do with being in a distributed environment. Since there could potentially be more than one mapper running on a given node we had to make sure that the directory to which the mapper writes its index has a unique name otherwise index files generated by one mapper could get overwritten by another. I decided to use the id of the map task as the unique name of the directory (the value of mapred.task.id obtained from the Configuration) but how to tell SOLR to use this value for the index directory name?

Using the debugger I traced the process of how SOLR writes its index files. I found that SOLR uses the updateHandler class as defined by the solrconfig.xml to instantiate a SolrIndexWriter class which writes the indexes. The second argument to SolrIndexWriter is the path to the directory in which the index will be written.  The update handler, by default org.apache.solr.update.DirectUpdateHandler2, defines a createMainIndexWriter(final String name, final boolean removeAllExisting) method in which the SolrIndexWriter is instantiated. By extending DirectUpdateHandler2 and overriding the createMainIndexWriter method I am able to control where the indexes are written.

The trickier part is how to get the map task id value down into the createMainIndexWriter method. The only object I found that flows from the higher level where the value of the map task id is available down the the update handler is the SolrInputDocument. This means that:

1. A field in the core’s schema.xml has to be defined: <field name="map_task_id"
type="text" indexed="false" stored="false" required="false" multiValued="false" />
2. When the SOLR document is generated in the mapper the value of the map task id must be added to the document
3. After the update handler has fetched the value of the map task id field from the document
   the field can be removed from the document (unless you want that value to appear in the final index).

While in the update handler where do you get the SolrInputDocument? Good question. There is another method in org.apache.solr.update.DirectUpdateHandler2 that needs to be overridden: public int addDoc(final AddUpdateCommand cmd) The AddUpdateCommand object contains the SolrInputDocument which contains the value of the map_task_id field. In the overridden addDoc method assign the AddUpdateCommand received to a member variable (making sure to call super(cmd) in case it is used elsewhere).

 
  /**
   * The cmd contains the SolrInputDocument which will/should
   * have the map task id as one of its keys. It is this value
   * that is used to name the index directory.
   *
   * @param cmd
   *          the command object that contains the SolrInputDocument
   * @return the parent class' code does not document the return value
   * @throws IOException
   */
  @Override
  public int addDoc(final AddUpdateCommand cmd) throws IOException {
    this.cmd = cmd;
    return super.addDoc(cmd);
  }

Finally in the createMainIndexWriter method get the SolrInputDocument from cmd.getSolrInputDocument() and use the value of the map_task_id field to create the directory name.

  /**
   * This method generates the new writer that will actually write the data to
   * the directory.
   *
   * @param name
   *          the name of the writer; in the parent it is the name of the class
   * @param removeAllExisting
   *          true means to create/overwrite existing; false means append
   * @return the writer
   * @throws IOException
   */
  @Override
  protected SolrIndexWriter createMainIndexWriter(final String name, final boolean removeAllExisting)
      throws IOException {
    final String indexDir = getNewIndexDir();
    final DirectoryFactory directoryFactory = getDirectoryFactory();
    final SolrIndexConfig mainIndexConfig = getMainIndexConfig();
    final IndexDeletionPolicyWrapper deletionPolicy = getDeletionPolicy();

    return new SolrIndexWriter("HadoopUpdateHandler", indexDir,
        directoryFactory, removeAllExisting, schema, mainIndexConfig,
        deletionPolicy);
  }

  /**
   * Generate the path to the index directory
   *
   * @return
   */
  private String getNewIndexDir() {
    String indexDirName = getDataDir() + getMapTaskIdDirectoryName();

    final File file = new File(indexDirName);
    if(!file.exists()) {
      indexDirName = file.getAbsolutePath();
      if(!createEmptyIndexDirectory(indexDirName)) {
        indexDirName = null;
      }
    }
    return indexDirName;
  }

  /**
   * From the cmd passed into addDoc fetch the SolrInputDocument. That
   * document should have the map task id. Grab that value then remove it from
   * the document (no need to index it).
   *
   * @return the name of the index directory ("index" is default value if the
   *         map task id value cannot be obtained from the solr input document.
   */
  private String getMapTaskIdDirectoryName() {
    String directory = "index";
    if(cmd != null) {
      final SolrInputDocument doc = cmd.getSolrInputDocument();
      if((doc != null) && doc.containsKey(SolrConstants.FIELD_MAP_TASK_ID)
          && (doc.get(SolrConstants.FIELD_MAP_TASK_ID) != null)) {
        directory = doc.get(SolrConstants.FIELD_MAP_TASK_ID).getFirstValue().toString();
        doc.removeField(SolrConstants.FIELD_MAP_TASK_ID);
      }
    }
    return directory;
  }