MongoDB
The MongoDB connector allows you to read and save documents. You can query a stream of documents from MongoSource
or update documents in a collection with MongoSink
.
This connector is based on the MongoDB Java Driver, which is compatible with MongoDB versions 2.6 through 4.4.
Another MongoDB connector is available - ReactiveMongo. It is a Scala driver that provides fully non-blocking and asynchronous I/O operations. Please read more about it in the ReactiveMongo documentation.
Project Info: Alpakka MongoDB | |
---|---|
Artifact | com.lightbend.akka
akka-stream-alpakka-mongodb
9.0.0
|
JDK versions | Eclipse Temurin JDK 11 Eclipse Temurin JDK 17 |
Scala versions | 2.13.12 |
JPMS module name | akka.stream.alpakka.mongodb |
License | |
Readiness level |
Since 0.15, 2017-12-06
|
Home page | https://doc.akka.io/libraries/alpakka/current |
API documentation | |
Forums | |
Release notes | GitHub releases |
Issues | Github issues |
Sources | https://github.com/akka/alpakka |
Artifacts
The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.
Additionally, add the dependencies as below.
- sbt
val AkkaVersion = "2.10.0" libraryDependencies ++= Seq( "com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "9.0.0", "com.typesafe.akka" %% "akka-stream" % AkkaVersion )
- Maven
- Gradle
The table below shows direct dependencies of this module and the second tab shows all libraries it depends on transitively.
- Direct dependencies
Organization Artifact Version com.typesafe.akka akka-stream_2.13 2.10.0 org.mongodb.scala mongo-scala-driver_2.13 4.11.4 org.scala-lang scala-library 2.13.12 - Dependency tree
Initialization
In the code examples below we will be using Mongo’s support for automatic codec derivation for POJOs. For Scala we will be using a case class and a macro based codec derivation. For Java a codec for POJO is derived using reflection.
- Scala
- Java
-
source
public final class Number { private Integer _id; public Number() {} public Number(Integer _id) { this._id = _id; } public void setId(Integer _id) { this._id = _id; } public Integer getId() { return _id; } }
For codec support, you first need to setup a CodecRegistry.
- Scala
- Java
-
source
PojoCodecProvider codecProvider = PojoCodecProvider.builder().register(Number.class).build(); CodecRegistry codecRegistry = CodecRegistries.fromProviders(codecProvider, new ValueCodecProvider());
Sources provided by this connector need a prepared collection to communicate with the MongoDB server. To get a reference to a collection, let’s initialize a MongoDB connection and access the database.
- Scala
- Java
-
source
client = MongoClients.create("mongodb://localhost:27017"); db = client.getDatabase("MongoSourceTest"); numbersColl = db.getCollection("numbers", Number.class).withCodecRegistry(codecRegistry);
We will also need an ActorSystem
.
Source
Let’s create a source from a Reactive Streams Publisher.
- Scala
- Java
-
source
final Source<Number, NotUsed> source = MongoSource.create(numbersColl.find(Number.class));
And then run it.
Here we used a basic sink to complete the stream by collecting all of the stream elements to a collection. The power of streams comes from building larger data pipelines which leverage backpressure to ensure efficient flow control. Feel free to edit the example code and build more advanced stream topologies.
Flow and Sink
Each of these sink factory methods have a corresponding factory in MongoFlow
which will emit the written document or result of the operation downstream.
Insert
We can use a Source of documents to save them to a mongo collection using MongoSink.insertOne
or MongoSink.insertMany
.
- Scala
- Java
-
source
List<Number> testRangeObjects = testRange.stream().map(Number::new).collect(toList()); final CompletionStage<Done> completion = Source.from(testRangeObjects).runWith(MongoSink.insertOne(numbersColl), system);
Insert Many
Insert many can be used if you have a collection of documents to insert at once.
- Scala
- Java
-
source
final List<Number> testRangeObjects = testRange.stream().map(Number::new).collect(toList()); final CompletionStage<Done> completion = Source.from(testRangeObjects).grouped(2).runWith(MongoSink.insertMany(numbersColl), system);
Update
We can update documents with a Source of DocumentUpdate
which is a filter and an update definition. Use either MongoSink.updateOne
or MongoSink.updateMany
if the filter should target one or many documents.
- Scala
- Java
-
source
final Source<DocumentUpdate, NotUsed> source = Source.from(testRange) .map( i -> DocumentUpdate.create( Filters.eq("value", i), Updates.set("updateValue", i * -1))); final CompletionStage<Done> completion = source.runWith(MongoSink.updateOne(numbersDocumentColl), system);
Delete
We can delete documents with a Source of filters. Use either MongoSink.deleteOne
or MongoSink.deleteMany
if the filter should target one or many documents.