Re: [akka-user] Re: Akka persistence query examples

[Akka Team]

One option would be to use a persistent actor as your read side as well,
store the last offset you have seen and stream from after that offset when
the actor has recovered completely, this way you can also provide snapshots
(and possible delete the actor to replay the entire history into a view
after adding functionality for example).

Another could be to persist the read side in some other way, a relational
database for example and do queries against that.

--
Johan
Akka Team

On Wed, Nov 9, 2016 at 5:34 PM, Juan Carlos Roig  wrote:

> Hi everyone. Just waking up this thread (once again) with a related
> challenge/question
>
>
> I'm used to use PersistentView for my read model before they got
> deprecated.
>
> I have a CQRS architecture where both write and read sides are cluster
> singleton. In write side I use persistent actors to validate and store the
> events and then those events get replayed to the persistent views on the
> other sharded cluster. Those persistent views process the events and keep
> the read model in memory, so queries are sent to them and they respond
> accordingly. This is very easy to do as akka persistence takes care of
> actors initialisation both at persistent actors and views so that in case
> any actor gets restarted or rebalanced by the cluster it will be started
> and initialized recovering its previous state by using snapshots and events
> replay before processing any new incoming message.
>
> I'm trying to do the same using persistent query but I'm having some
> problems.
>
> I'm creating the persistent query inside the actor view and send the
> events to itself.
>
> Example:
>
> class DefaultAccountView extends Actor with ActorLogging {
>   implicit val mat = ActorMaterializer()(context.system)
>   implicit val ec = context.dispatcher
>
>   val aggregateId = context.self.path.name
>
>   def persistenceId: String = s"AccountAggregate$aggregateId"
>
>   // Configure Persistence Query
>   val queries = PersistenceQuery(context.system).readJournalFor[
> LeveldbReadJournal](LeveldbReadJournal.Identifier)
>
>   val events = queries.eventsByPersistenceId(persistenceId, 0L,
> Long.MaxValue)
>   .map(eventEnvelope => eventEnvelope.event)
>   .to(Sink.actorRef(self, StreamCompleted))
>   .run()
>
>
>   def handleAccountQueries: Recevice = {
> case Query1 =>
>   // Process query 1
> case Query 2 =>
>   // Process query 2
>   }
>
>
>   def handleEvents: Receive = {
> case Event1 =>
>   // process event 1 update state
>
>   case Event2 =>
> // process event 2 update state
>
>   }
>
>
>   override def receive: Receive = handleAccountQueries orElse handleEvents
> }
>
> In that case the stream will be created every time the actor gets started
> and will replay all events. The problem is that being an actor belonging to
> a sharded cluster it will be started the first time it gets a message, that
> is, a query. What happens is the first message always gets processed before
> all events are replayed and hence it fails to give correct response as the
> state is not initialised, in other words, there's no initialisation
> lyfecycle like PersistentView has. I tried to use a custom object to signal
> the end of the stream (StreamCompleted) but obviously it is never sent as
> the stream never finishes (it keeps opened sending events generated in real
> time, which is what we need to keep the view updated). On a side note, it
> will be very nice to be able to easily use snapshots here too as replay all
> events every time the view starts might take a lot of time for high load
> actors.
>
> Is there any way to do this overcoming these problems?
>
> Shall I continue using PersistentView instead although they're deprecated?
>
> How are you guys propose to implement this use case if persistent views
> are deprecated?
>
> Regards,
>
> J. Carlos
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send 

Re: [akka-user] Re: Akka persistence query examples

[Juan Carlos Roig]

Hi everyone. Just waking up this thread (once again) with a related 
challenge/question


I'm used to use PersistentView for my read model before they got 
deprecated. 

I have a CQRS architecture where both write and read sides are cluster 
singleton. In write side I use persistent actors to validate and store the 
events and then those events get replayed to the persistent views on the 
other sharded cluster. Those persistent views process the events and keep 
the read model in memory, so queries are sent to them and they respond 
accordingly. This is very easy to do as akka persistence takes care of 
actors initialisation both at persistent actors and views so that in case 
any actor gets restarted or rebalanced by the cluster it will be started 
and initialized recovering its previous state by using snapshots and events 
replay before processing any new incoming message.

I'm trying to do the same using persistent query but I'm having some 
problems. 

I'm creating the persistent query inside the actor view and send the events 
to itself. 

Example:

class DefaultAccountView extends Actor with ActorLogging {
  implicit val mat = ActorMaterializer()(context.system)
  implicit val ec = context.dispatcher

  val aggregateId = context.self.path.name

  def persistenceId: String = s"AccountAggregate$aggregateId"

  // Configure Persistence Query
  val queries = 
PersistenceQuery(context.system).readJournalFor[LeveldbReadJournal](LeveldbReadJournal.Identifier)

  val events = queries.eventsByPersistenceId(persistenceId, 0L, 
Long.MaxValue)
  .map(eventEnvelope => eventEnvelope.event)
  .to(Sink.actorRef(self, StreamCompleted))
  .run()


  def handleAccountQueries: Recevice = {
case Query1 =>
  // Process query 1
case Query 2 =>
  // Process query 2
  }


  def handleEvents: Receive = {
case Event1 =>
  // process event 1 update state

  case Event2 =>
// process event 2 update state

  }


  override def receive: Receive = handleAccountQueries orElse handleEvents
}

In that case the stream will be created every time the actor gets started 
and will replay all events. The problem is that being an actor belonging to 
a sharded cluster it will be started the first time it gets a message, that 
is, a query. What happens is the first message always gets processed before 
all events are replayed and hence it fails to give correct response as the 
state is not initialised, in other words, there's no initialisation 
lyfecycle like PersistentView has. I tried to use a custom object to signal 
the end of the stream (StreamCompleted) but obviously it is never sent as 
the stream never finishes (it keeps opened sending events generated in real 
time, which is what we need to keep the view updated). On a side note, it 
will be very nice to be able to easily use snapshots here too as replay all 
events every time the view starts might take a lot of time for high load 
actors.

Is there any way to do this overcoming these problems?

Shall I continue using PersistentView instead although they're deprecated? 

How are you guys propose to implement this use case if persistent views are 
deprecated?

Regards,

J. Carlos

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Patrik Nordwall]

There is a offset: Long that the query journal can implement in a way that
is suitable. Easiest is to use a timestamp, or a global sequence number if
you have that luxury (e.g. a sql database).

The Cassandra plugin is using a timeuuid (a Cassandra data type) column and
therefore supports queries with timestamp offset

or uuid offset

.

Cheers,
Patrik
​

On Fri, Feb 12, 2016 at 8:56 AM, Andrew Easter 
wrote:

> Hi everyone. Just waking up this thread with a related challenge/question.
>
> One of the objectives of Persistence Query was to address the limitation
> that a PersistentView was only able to project from a single persistence
> id. As we all know, this meant that trying to do CQRS in a DDD based system
> (actor per instance of an aggregate) wasn't possible without workarounds.
> The workaround I used was to have all my aggregate actors send (using ALOD
> trait) events on to an intermediate persistent actor, one per aggregate
> type. This meant I could create a PersistentView that could see a stream of
> all events from all instances of an aggregate type.
>
> Obviously, the above solution is a little cumbersome. Downsides such as
> having to essentially duplicate all events in the intermediate persistent
> actor.
>
> The Persistence Query introduces us to the concept of querying events by
> tag - the docs suggest this as the most appropriate method for implementing
> CQRS in a DDD based actor implementation. However, I'm struggling to
> understand how this is really fit for purpose given that there are no order
> guarantees in an eventsByTag stream, and accordingly no sequence number I
> can refer back to such that I know how far through the stream I've read.
>
> The discussion here is about persisting an offset in the view datastore,
> but how can I do that when there is no reliable seq number?
>
> Sorry if I'm not making sense! Would love to hear some ideas about how to
> deal with this.
>
> Thanks, Andrew
>
> --
> >>  Read the docs: http://akka.io/docs/
> >>  Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>  Search the archives:
> https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>



-- 

Patrik Nordwall
Typesafe  -  Reactive apps on the JVM
Twitter: @patriknw

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Andrew Easter]

Hi Patrik. Thanks for this. I reckon I get it now :-)

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Andrew Easter]

Hi everyone. Just waking up this thread with a related challenge/question.

One of the objectives of Persistence Query was to address the limitation that a 
PersistentView was only able to project from a single persistence id. As we all 
know, this meant that trying to do CQRS in a DDD based system (actor per 
instance of an aggregate) wasn't possible without workarounds. The workaround I 
used was to have all my aggregate actors send (using ALOD trait) events on to 
an intermediate persistent actor, one per aggregate type. This meant I could 
create a PersistentView that could see a stream of all events from all 
instances of an aggregate type.

Obviously, the above solution is a little cumbersome. Downsides such as having 
to essentially duplicate all events in the intermediate persistent actor.

The Persistence Query introduces us to the concept of querying events by tag - 
the docs suggest this as the most appropriate method for implementing CQRS in a 
DDD based actor implementation. However, I'm struggling to understand how this 
is really fit for purpose given that there are no order guarantees in an 
eventsByTag stream, and accordingly no sequence number I can refer back to such 
that I know how far through the stream I've read.

The discussion here is about persisting an offset in the view datastore, but 
how can I do that when there is no reliable seq number?

Sorry if I'm not making sense! Would love to hear some ideas about how to deal 
with this.

Thanks, Andrew

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Konrad Malawski]

On Thu, Dec 10, 2015 at 7:53 AM, Alan Johnson  wrote:

> I'm new to this space, but yes, it seems to me like the read store should
> store lastProcessEventNr someplace, ideally updated in the same transaction
> that updates the view.


Welcome to the community then :-)
Yup, that's exactly the idea, the offset has to be maintained somewhere.


-- 
Cheers,
Konrad 'ktoso' Malawski
Akka  @ Typesafe 

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Adam Dohnal]

Hello,

now I better understand queries like allPersistenceIds() and 
eventsByPersistenceId(persistenceId). What can I do if persistent query 
implementation I use does not support live stream of events? Should I 
implement polling myself (schedule message to self and after receive call 
query again)? But every time this query will be called, every event will be 
processed again and again. Should I store somewhere lastProcessedEventNr 
and use it as offset in eventsByPersistenceId?

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: Akka persistence query examples

[Bruno Filippone]

What you want to do is process all the events coming from the 
PersistenceQuery stream and create a view that can be queried, for example: 
transform all the events into normalized data and insert them into 
ElasticSearch.

A quick example would look like the following (using EventStoreReadJournal):

val queries = PersistenceQuery(system).readJournalFor
[EventStoreReadJournal](EventStoreReadJournal.Identifier)

queries.allPersistenceIds()

.runForeach { persistenceId =>


queries.eventsByPersistenceId(persistenceId).map(_.event).runForEach(elasticSearchIndexer
 
! _)

}



Hope this helps.

On a side note, one of the things I'm struggling with instead is how to 
process the event stream if I have a cluster of nodes on my read side 
without having all the nodes processing all the events coming from the 
events source, potentially I would like to have each node indexing a subset 
of the events into my ElasticSearch database.

Does anyone have any guidance on how to achieve this?

The problem is that I have no way to chop the stream up in my read nodes 
based on my cluster shards for example, the PersistentView played well with 
cluster sharding, but what about PersistenceQuery?


On Sunday, November 8, 2015 at 5:36:57 PM UTC, Adam Dohnal wrote:
>
> Hello,
>
> recently I read through new akka-persistence-query-experimental module and 
> would like to see some tutorials and examples to see how it can work. 
> Documentation is really abstract for me. Unfortunately, I'm not able to 
> find any activator templates nor tutorials. 
>
> What I understand is how can I get a stream of events for my aggregate 
> root. But how can I construct my read model. Should I subscribe to that 
> event stream and for each event I should update ma read model and store it 
> to read database? I think I am little confused.
>
> Can you guys share your experience with that module? Some blog posts, 
> mini-projects on github etc... to show how can it be used to implement 
> query side of CQRS? Thank you
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Re: Akka persistence query examples

[Bruno Filippone]

Thanks for that, I'm not very familiar with the Akka streams API yet, so 
any feedback is very welcome.

I guess the cluster singleton could work for my use case, it could just 
communicate to sharded workers that do the data normalization so that it 
does not become a bottleneck.

On Friday, November 20, 2015 at 6:21:34 PM UTC, Patrik Nordwall wrote:
>
>
>
> 20 nov. 2015 kl. 18:50 skrev Bruno Filippone  >:
>
>
> What you want to do is process all the events coming from the 
> PersistenceQuery stream and create a view that can be queried, for example: 
> transform all the events into normalized data and insert them into 
> ElasticSearch.
>
>
> Right
>
>
> A quick example would look like the following (using 
> EventStoreReadJournal):
>
> val queries = PersistenceQuery(system).readJournalFor
> [EventStoreReadJournal](EventStoreReadJournal.Identifier)
>
> queries.allPersistenceIds()
>
> .runForeach { persistenceId =>
>
> 
> queries.eventsByPersistenceId(persistenceId).map(_.event).runForEach(elasticSearchIndexer
>  
> ! _)
>
> }
>
>
> You should use mapAsync + ask or some of the other facilities in akka 
> streams to integrate with actors without loosing the backpressure to the 
> stream. The problem with foreach + tell is that if elasticSearchIndexer is 
> slower than the stream you will fill up the mailbox snd eventually OOME.
>
>
> Hope this helps.
>
> On a side note, one of the things I'm struggling with instead is how to 
> process the event stream if I have a cluster of nodes on my read side 
> without having all the nodes processing all the events coming from the 
> events source, potentially I would like to have each node indexing a subset 
> of the events into my ElasticSearch database.
>
> Does anyone have any guidance on how to achieve this?
>
> The problem is that I have no way to chop the stream up in my read nodes 
> based on my cluster shards for example, the PersistentView played well with 
> cluster sharding, but what about PersistenceQuery?
>
>
> Perhaps you can use a cluster singleton for the allPersistenceIds query, 
> which delegates to a sharded entity actor for each persistenceId?
>
> Thanks for sharing your thoughts about this.
>
> Cheers,
> Patrik
>
>
>
> On Sunday, November 8, 2015 at 5:36:57 PM UTC, Adam Dohnal wrote:
>>
>> Hello,
>>
>> recently I read through new akka-persistence-query-experimental module 
>> and would like to see some tutorials and examples to see how it can work. 
>> Documentation is really abstract for me. Unfortunately, I'm not able to 
>> find any activator templates nor tutorials. 
>>
>> What I understand is how can I get a stream of events for my aggregate 
>> root. But how can I construct my read model. Should I subscribe to that 
>> event stream and for each event I should update ma read model and store it 
>> to read database? I think I am little confused.
>>
>> Can you guys share your experience with that module? Some blog posts, 
>> mini-projects on github etc... to show how can it be used to implement 
>> query side of CQRS? Thank you
>>
> -- 
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: 
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> --- 
> You received this message because you are subscribed to the Google Groups 
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an 
> email to akka-user+...@googlegroups.com .
> To post to this group, send email to akka...@googlegroups.com 
> .
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.