Apache Solr

Apache Solr

Solr (pronounced “solar”) is an open source enterprise search platform, written in Java, from the Apache Lucene project. Its major features include full-text search, hit highlighting, faceted search, real-time indexing, dynamic clustering, database integration, NoSQL features and rich document (e.g., Word, PDF) handling. Providing distributed search and index replication, Solr is designed for scalability and fault tolerance. Solr is widely used for enterprise search and analytics use cases and has an active development community and regular releases.

Wikipedia

Alpakka Solr provides Akka Stream sources and sinks for Apache Solr.

For more information about Solr please visit the Solr documentation.

Project Info: Alpakka Solr
Artifact
com.lightbend.akka
akka-stream-alpakka-solr
2.0.1
JDK versions
Adopt OpenJDK 8
Adopt OpenJDK 11
Scala versions2.12.10, 2.11.12, 2.13.1
JPMS module nameakka.stream.alpakka.solr
License
Readiness level
Since 0.17, 2018-02-19
Home pagehttps://doc.akka.io/docs/alpakka/current
API documentation
Forums
Release notesIn the documentation
IssuesGithub issues
Sourceshttps://github.com/akka/alpakka

Artifacts

sbt
val AkkaVersion = "2.5.31"
libraryDependencies ++= Seq(
  "com.lightbend.akka" %% "akka-stream-alpakka-solr" % "2.0.1",
  "com.typesafe.akka" %% "akka-stream" % AkkaVersion
)
Maven
<properties>
  <akka.version>2.5.31</akka.version>
  <scala.binary.version>2.12</scala.binary.version>
</properties>
<dependency>
  <groupId>com.lightbend.akka</groupId>
  <artifactId>akka-stream-alpakka-solr_${scala.binary.version}</artifactId>
  <version>2.0.1</version>
</dependency>
<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-stream_${scala.binary.version}</artifactId>
  <version>${akka.version}</version>
</dependency>
Gradle
versions += [
  AkkaVersion: "2.5.31",
  ScalaBinary: "2.12"
]
dependencies {
  compile group: 'com.lightbend.akka', name: "akka-stream-alpakka-solr_${versions.ScalaBinary}", version: '2.0.1',
  compile group: 'com.typesafe.akka', name: "akka-stream_${versions.ScalaBinary}", version: versions.AkkaVersion
}

The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.

Direct dependencies
OrganizationArtifactVersion
com.typesafe.akkaakka-stream_2.122.5.31
org.apache.solrsolr-solrj7.7.2
org.scala-langscala-library2.12.10
Dependency tree
com.typesafe.akka    akka-stream_2.12    2.5.31
    com.typesafe.akka    akka-actor_2.12    2.5.31
        com.typesafe    config    1.3.3
        org.scala-lang.modules    scala-java8-compat_2.12    0.8.0
    com.typesafe.akka    akka-protobuf_2.12    2.5.31
    com.typesafe    ssl-config-core_2.12    0.3.8
        com.typesafe    config    1.3.3
        org.scala-lang.modules    scala-parser-combinators_2.12    1.1.2
    org.reactivestreams    reactive-streams    1.0.2
org.apache.solr    solr-solrj    7.7.2
    commons-io    commons-io    2.5
    org.apache.commons    commons-math3    3.6.1
    org.apache.httpcomponents    httpclient    4.5.6
    org.apache.httpcomponents    httpcore    4.4.10
    org.apache.httpcomponents    httpmime    4.5.6
    org.apache.zookeeper    zookeeper    3.4.14
    org.codehaus.woodstox    stax2-api    3.1.4
    org.codehaus.woodstox    woodstox-core-asl    4.4.1
    org.noggit    noggit    0.8
    org.slf4j    jcl-over-slf4j    1.7.24
    org.slf4j    slf4j-api    1.7.30
org.scala-lang    scala-library    2.12.10

Set up a Solr client

Sources, Flows and Sinks provided by this connector need a prepared SolrClient (eg. CloudSolrClient) to access to Solr.

Scala
final val zookeeperPort = 9984
final val zookeeperHost = s"127.0.0.1:$zookeeperPort/solr"
implicit val solrClient: CloudSolrClient =
  new CloudSolrClient.Builder(Arrays.asList(zookeeperHost), Optional.empty()).build
Java
private static final int zookeeperPort = 9984;
private static final String zookeeperHost = "127.0.0.1:" + zookeeperPort + "/solr";

  CloudSolrClient solrClient =
      new CloudSolrClient.Builder(Arrays.asList(zookeeperHost), Optional.empty()).build();

Reading from Solr

Create a Solr TupleStream (eg. via CloudSolrStream) and use SolrSource.fromTupleStream (APIAPI) to create a source.

Scala
val factory = new StreamFactory().withCollectionZkHost(collection, zookeeperHost)
val solrClientCache = new SolrClientCache()
val streamContext = new StreamContext()
streamContext.setSolrClientCache(solrClientCache)

val expression =
  StreamExpressionParser.parse(s"""search($collection, q=*:*, fl="title,comment", sort="title asc")""")
val stream: TupleStream = new CloudSolrStream(expression, factory)
stream.setStreamContext(streamContext)

val source = SolrSource
  .fromTupleStream(stream)
Java
StreamFactory factory = new StreamFactory().withCollectionZkHost(collection, zookeeperHost);
SolrClientCache solrClientCache = new SolrClientCache();
StreamContext streamContext = new StreamContext();
streamContext.setSolrClientCache(solrClientCache);

String expressionStr =
    String.format("search(%s, q=*:*, fl=\"title,comment\", sort=\"title asc\")", collection);
StreamExpression expression = StreamExpressionParser.parse(expressionStr);
TupleStream stream = new CloudSolrStream(expression, factory);
stream.setStreamContext(streamContext);

Source<Tuple, NotUsed> source = SolrSource.fromTupleStream(stream);

Writing to Solr

Alpakka Solr batches updates to Solr by sending all updates of the same operation type at once to Solr. These batches are extracted from the elements within one collection sent to a Solr flow or sink. Updates of different types may be contained in a single collection sent, though. In case streams don’t have natural batches of updates, you may use the groupedWithin operator to create count or time-based batches.

Alpakka Solr offers three styles for writing to Apache Solr:

  1. Using SolrInputDocument (via SolrSink.documents, SolrFlow.documents and SolrFlow.documentsWithPassThrough)
  2. Annotated Java Bean classes supported by Solr’s DocumentObjectBinder (via SolrSink.beans, SolrFlow.beans and SolrFlow.beansWithPassThrough)
  3. Typed streams with document binders to translate to SolrInputDocument (via SolrSink.typeds, SolrFlow.typeds and SolrFlow.typedsWithPassThrough)

In all variations the data is wrapped into WriteMessages.

Committing and configuration for updates

Data sent to Solr is not searchable until it has been committed to the index. These are the major options for handling commits:

  1. The Solr installation can be configured to use auto-commit.
  2. Specify commit-within in SolrUpdateSettings to trigger commits after every write through Alpakka Solr.
  3. Use explicit committing via the SolrClient.commit methods on stream completion as most examples show. As commit is a blocking operation, choose an appropriate execution context (preferably not system.dispatcher).

Configuration of Solr committing is described in UpdateHandlers in SolrConfig.

Available settings

Parameter Default Description
commitWithin -1 Max time (in ms) before a commit will happen, -1 for explicit committing
Scala
import akka.stream.alpakka.solr.SolrUpdateSettings

val settings = SolrUpdateSettings()
  .withCommitWithin(-1)
Java
import akka.stream.alpakka.solr.SolrUpdateSettings;
SolrUpdateSettings settings = SolrUpdateSettings.create().withCommitWithin(-1);

Writing SolrInputDocuments

Use SolrSink.document to stream SolrInputDocument to Solr.

Defining mappings

Scala
case class Book(title: String, comment: String = "", routerOpt: Option[String] = None)

val bookToDoc: Book => SolrInputDocument = { b =>
  val doc = new SolrInputDocument
  doc.setField("title", b.title)
  doc.setField("comment", b.comment)
  b.routerOpt.foreach { router =>
    doc.setField("router", router)
  }
  doc
}

val tupleToBook: Tuple => Book = { t =>
  val title = t.getString("title")
  Book(title, t.getString("comment"))
}
Java
public static class Book {
  public String title;

  public String comment;

  public String router;

  public Book() {}

  public Book(String title) {
    this.title = title;
  }

  public Book(String title, String comment) {
    this.title = title;
    this.comment = comment;
  }

  public Book(String title, String comment, String router) {
    this.title = title;
    this.comment = comment;
    this.router = router;
  }
}

Function<Book, SolrInputDocument> bookToDoc =
    book -> {
      SolrInputDocument doc = new SolrInputDocument();
      doc.setField("title", book.title);
      doc.setField("comment", book.comment);
      if (book.router != null) doc.setField("router", book.router);
      return doc;
    };

Function<Tuple, Book> tupleToBook =
    tuple -> {
      String title = tuple.getString("title");
      return new Book(title, tuple.getString("comment"));
    };

Use SolrSink.documents, SolrFlow.documents or SolrFlow.documentsWithPassThrough to stream SolrInputDocuments to Solr.

A SolrClient must be provided to SolrSink implicitly. SolrSink.

Scala
val copyCollection = SolrSource
  .fromTupleStream(stream)
  .map { tuple: Tuple =>
    val book: Book = tupleToBook(tuple)
    val doc: SolrInputDocument = bookToDoc(book)
    WriteMessage.createUpsertMessage(doc)
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink.documents(collectionName, SolrUpdateSettings())
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> copyCollection =
    SolrSource.fromTupleStream(stream)
        .map(
            tuple -> {
              Book book = tupleToBook.apply(tuple);
              SolrInputDocument doc = bookToDoc.apply(book);
              return WriteMessage.createUpsertMessage(doc);
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.documents(collectionName, SolrUpdateSettings.create(), solrClient),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

Writing Java beans

Firstly, create a POJO.

Scala
import org.apache.solr.client.solrj.beans.Field

import scala.annotation.meta.field
case class BookBean(@(Field @field) title: String)
Java
class BookBean {
  @Field("title")
  public String title;

  public BookBean(String title) {
    this.title = title;
  }
}

Use SolrSink.beans, SolrFlow.beans or SolrFlow.beansWithPassThrough to stream POJOs to Solr.

Scala
val copyCollection = SolrSource
  .fromTupleStream(stream)
  .map { tuple: Tuple =>
    val title = tuple.getString("title")
    WriteMessage.createUpsertMessage(BookBean(title))
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink.beans[BookBean](collectionName, SolrUpdateSettings())
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> copyCollection =
    SolrSource.fromTupleStream(stream)
        .map(
            tuple -> {
              String title = tuple.getString("title");
              return WriteMessage.createUpsertMessage(new BookBean(title));
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.beans(
                collectionName, SolrUpdateSettings.create(), solrClient, BookBean.class),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

Writing arbitrary classes via custom binding

Use SolrSink.typeds, SolrFlow.typeds or SolrFlow.typedsWithPassThrough to stream messages with custom binding to Solr.

Scala
val copyCollection = SolrSource
  .fromTupleStream(stream)
  .map { tuple: Tuple =>
    val book: Book = tupleToBook(tuple)
    WriteMessage.createUpsertMessage(book)
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink
      .typeds[Book](
        collectionName,
        SolrUpdateSettings(),
        binder = bookToDoc
      )
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> copyCollection =
    SolrSource.fromTupleStream(stream)
        .map(tuple -> WriteMessage.createUpsertMessage(tupleToBook.apply(tuple)))
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.typeds(
                collectionName, SolrUpdateSettings.create(), bookToDoc, solrClient, Book.class),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

Using a flow with custom binding

You can also build flow stages with SolrFlow. SolrFlow.

Scala
val copyCollection = SolrSource
  .fromTupleStream(stream)
  .map { tuple: Tuple =>
    val book: Book = tupleToBook(tuple)
    WriteMessage.createUpsertMessage(book)
  }
  .groupedWithin(5, 10.millis)
  .via(
    SolrFlow
      .typeds[Book](
        collectionName,
        SolrUpdateSettings(),
        binder = bookToDoc
      )
  )
  .runWith(Sink.seq)
  // explicit commit when stream ended
  .map { seq =>
    solrClient.commit(collectionName)
    seq
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> copyCollection =
    SolrSource.fromTupleStream(stream)
        .map(tuple -> WriteMessage.createUpsertMessage(tupleToBook.apply(tuple)))
        .groupedWithin(5, Duration.ofMillis(10))
        .via(
            SolrFlow.typeds(
                collectionName, SolrUpdateSettings.create(), bookToDoc, solrClient, Book.class))
        .runWith(Sink.ignore(), materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

Passing data through SolrFlow

All flow types (documents, beans, typeds) exist with pass-through support: Use SolrFlow.documentsWithPassThrough, SolrFlow.beansWithPassThrough or SolrFlow.typedsWithPassThrough.

When streaming documents from Kafka, you might want to commit to Kafka AFTER the document has been written to Solr. This scenario uses implicit committing via the commit within setting.

Scala
// Note: This code mimics Alpakka Kafka APIs
val copyCollection = kafkaConsumerSource
  .map { kafkaMessage: CommittableMessage =>
    val book = kafkaMessage.book
    // Transform message so that we can write to solr
    WriteMessage.createUpsertMessage(book).withPassThrough(kafkaMessage.committableOffset)
  }
  .groupedWithin(5, 10.millis)
  .via( // write to Solr
    SolrFlow.typedsWithPassThrough[Book, CommittableOffset](
      collectionName,
      // use implicit commits to Solr
      SolrUpdateSettings().withCommitWithin(5),
      binder = bookToDoc
    )
  ) // check status and collect Kafka offsets
  .map { messageResults =>
    val offsets = messageResults.map { result =>
      if (result.status != 0)
        throw new Exception("Failed to write message to Solr")
      result.passThrough
    }
    CommittableOffsetBatch(offsets)
  }
  .mapAsync(1)(_.commitScaladsl())
  .runWith(Sink.ignore)
Java
// Note: This code mimics Alpakka Kafka APIs
CompletionStage<Done> completion =
    kafkaConsumerSource // Assume we get this from Kafka
        .map(
            kafkaMessage -> {
              Book book = kafkaMessage.book;
              // Transform message so that we can write to elastic
              return WriteMessage.createUpsertMessage(book)
                  .withPassThrough(kafkaMessage.committableOffset);
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .via(
            SolrFlow.typedsWithPassThrough(
                collectionName,
                // use implicit commits to Solr
                SolrUpdateSettings.create().withCommitWithin(5),
                bookToDoc,
                solrClient,
                Book.class))
        .map(
            messageResults ->
                messageResults.stream()
                    .map(
                        result -> {
                          if (result.status() != 0) {
                            throw new RuntimeException("Failed to write message to Solr");
                          }
                          return result.passThrough();
                        })
                    .collect(Collectors.toList()))
        .map(ConsumerMessage::createCommittableOffsetBatch)
        .mapAsync(1, CommittableOffsetBatch::commitJavadsl)
        .runWith(Sink.ignore(), materializer);

Excluding messages

Failure to deserialize a kafka message is a particular case of conditional message processing. It is also likely that we would have no message to produce to SolR when we encounter messages that fail to deserialize. The solr flow will not let us pass through the corresponding committable offset without doing a request to solr.

Use WriteMessage.createPassThrough to exclude this message without doing any change on solr inside a flow.

Scala
// Note: This code mimics Alpakka Kafka APIs
val copyCollection = kafkaConsumerSource
  .map { offset: CommittableOffset =>
    // Transform message so that we can write to solr
    WriteMessage.createPassThrough(offset).withSource(new SolrInputDocument())
  }
  .groupedWithin(5, 10.millis)
  .via( // write to Solr
    SolrFlow.documentsWithPassThrough[CommittableOffset](
      collectionName,
      // use implicit commits to Solr
      SolrUpdateSettings().withCommitWithin(5)
    )
  ) // check status and collect Kafka offsets
  .map { messageResults =>
    val offsets = messageResults.map { result =>
      if (result.status != 0)
        throw new Exception("Failed to write message to Solr")
      result.passThrough
    }
    CommittableOffsetBatch(offsets)
  }
  .mapAsync(1)(_.commitScaladsl())
  .runWith(Sink.ignore)
Java
// Note: This code mimics Alpakka Kafka APIs
CompletionStage<Done> completion =
    kafkaConsumerSource // Assume we get this from Kafka
        .map(
            kafkaMessage -> {
              // Transform message so that we can write to elastic
              return WriteMessage.createPassThrough(kafkaMessage)
                  .withSource(new SolrInputDocument());
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .via(
            SolrFlow.documentsWithPassThrough(
                collectionName,
                // use implicit commits to Solr
                SolrUpdateSettings.create().withCommitWithin(5),
                solrClient))
        .map(
            messageResults ->
                messageResults.stream()
                    .map(
                        result -> {
                          if (result.status() != 0) {
                            throw new RuntimeException("Failed to write message to Solr");
                          }
                          return result.passThrough();
                        })
                    .collect(Collectors.toList()))
        .map(ConsumerMessage::createCommittableOffsetBatch)
        .mapAsync(1, CommittableOffsetBatch::commitJavadsl)
        .runWith(Sink.ignore(), materializer);

Update documents

With WriteMessage.createUpdateMessage documents can be updated atomically. All flow and sink types (documents, beans, typeds) support atomic updates.

Scala
val updateCollection = SolrSource
  .fromTupleStream(stream2)
  .map { tuple: Tuple =>
    val id = tuple.fields.get("title").toString
    val comment = tuple.fields.get("comment").toString
    WriteMessage.createUpdateMessage[SolrInputDocument](
      idField = "title",
      idValue = id,
      updates = Map(
        "comment" ->
        Map("set" -> (comment + " It is a good book!!!"))
      )
    )
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink.documents(collectionName, SolrUpdateSettings())
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> updateCollection =
    SolrSource.fromTupleStream(stream2)
        .map(
            t -> {
              String id = t.fields.get("title").toString();
              String comment = t.fields.get("comment").toString();
              Map<String, Object> m2 = new HashMap<>();
              m2.put("set", (comment + " It's is a good book!!!"));
              Map<String, Map<String, Object>> updates = new HashMap<>();
              updates.put("comment", m2);
              return WriteMessage.<SolrInputDocument>createUpdateMessage("title", id, updates);
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.documents(collectionName, SolrUpdateSettings.create(), solrClient),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

If a collection contains a router field, use the WriteMessage.createUpdateMessage(...).withRoutingFieldValue(..) to set the router field.

Delete documents by ids

With WriteMessage.createDeleteMessage(id) documents may be deleted by ID. All flow and sink types (documents, beans, typeds) support deleting.

Scala
val deleteDocuments = SolrSource
  .fromTupleStream(stream2)
  .map { tuple: Tuple =>
    val id = tuple.fields.get("title").toString
    WriteMessage.createDeleteMessage[SolrInputDocument](id)
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink.documents(collectionName, SolrUpdateSettings())
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> deleteDocuments =
    SolrSource.fromTupleStream(stream2)
        .map(
            t -> {
              String id = tupleToBook.apply(t).title;
              return WriteMessage.<SolrInputDocument>createDeleteMessage(id);
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.documents(collectionName, SolrUpdateSettings.create(), solrClient),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });

Delete documents by query

With WriteMessage.createDeleteByQueryMessage(query) documents matching a query may be deleted. All flow and sink types (documents, beans, typeds) support deleting.

Scala
val deleteByQuery = SolrSource
  .fromTupleStream(stream2)
  .map { tuple: Tuple =>
    val title = tuple.fields.get("title").toString
    WriteMessage.createDeleteByQueryMessage[SolrInputDocument](
      s"""title:"$title" """
    )
  }
  .groupedWithin(5, 10.millis)
  .runWith(
    SolrSink.documents(collectionName, SolrUpdateSettings())
  )
  // explicit commit when stream ended
  .map { _ =>
    solrClient.commit(collectionName)
  }(commitExecutionContext)
Java
CompletionStage<UpdateResponse> deleteByQuery =
    SolrSource.fromTupleStream(stream2)
        .map(
            t -> {
              String id = t.fields.get("title").toString();
              return WriteMessage.<SolrInputDocument>createDeleteByQueryMessage(
                  "title:\"" + id + "\"");
            })
        .groupedWithin(5, Duration.ofMillis(10))
        .runWith(
            SolrSink.documents(collectionName, SolrUpdateSettings.create(), solrClient),
            materializer)
        // explicit commit when stream ended
        .thenApply(
            done -> {
              try {
                return solrClient.commit(collectionName);
              } catch (Exception e) {
                throw new IllegalStateException(e);
              }
            });
Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.