sql >> Database teknologi >  >> NoSQL >> MongoDB

bruge akka-strømme til at gå over mongo-samlingen

Løsning 1:

def changeModelAndInsertToNewCollection(person:Person) : Future[Boolean] ={
//Todo : call mongo api to update the person
???
}

def processPeople()(implicit m: Materializer): Future[Done] = {
val numberOfConcurrentUpdate = 10

val peopleSource: Source[Person, Future[State]] =
  collection
    .find(json())
    .cursor[Person]()
    .documentSource()

peopleSource
  .mapAsync(numberOfConcurrentUpdate)(changeModelAndInsertToNewCollection)
  withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider))
  .runWith(Sink.ignore)}

Løsning 2: bruger Alpakka som akka stream stik til mongo

val source: Source[Document, NotUsed] =
MongoSource(collection.find(json()).cursor[Person]().documentSource())

source.runWith(MongoSink.updateOne(2, collection))



  1. Aggreger gruppe efter dato med sommertid offset

  2. Hvordan udfører man Persistence Store i Redis?

  3. Hvordan opretter jeg en MongoDB-dump af min database?

  4. Sådan skifter du et element i array i mongoDB