| = Section 6: Projection for queries |
| :page-supergroup-java-scala: Language |
| |
| include::ROOT:partial$include.adoc[] |
| |
| Next, we will create an {akka-projection}/[Akka Projection {tab-icon}, window="tab"] from the events emitted by the `ShoppingCart` entity. The Projection will update counts in the database to track item popularity. Then, we can query the database to find how popular an item is. Since `ShoppingCart` entities can only be addressed by individual cart identifiers, we can find a particular cart, but we can't find all carts that contain a particular item. |
| |
| ifdef::review[REVIEWERS: I don't follow the logic at the end of the last sentence in the prev para. Is it that each shopping cart only knows about its own items?] |
| |
| |
| [caption=""] |
| image::example-projection-query.svg[Example query] |
| |
| This piece of the xref:overview.adoc[full example] focuses on the `ItemPopularityProjection` and a query representation in the database. On this page you will learn how to: |
| |
| * implement a Projection |
| * distribute the Projection instances over the nodes in the Akka Cluster |
| * work with the Projection JDBC API |
| |
| The xref:concepts:cqrs.adoc[CQRS] section explains why it is a good practice to build a Projection from entity events that can be queried. The {akka-blog}/news/2020/09/10/akka-projection-intro-video[Introduction to Akka Projections video {tab-icon}, window="tab"] is also a good starting point for learning about Akka Projections. |
| |
| This example is using PostgreSQL for storing the Projection result, and the Projection offset. An alternative is described in xref:how-to:cassandra-alternative.adoc[]. |
| |
| === Akka Workshop |
| The third video of the https://info.lightbend.com/akka-platform-workshop-part-3-on-demand-recording.html[Akka Workshop Series {tab-icon}, window="tab"] covers both Projections for queries and CQRS. It provides some solid guidance to aid you in understanding this section of this guide. |
| |
| == Source downloads |
| |
| If you prefer to simply view and run the example, download a zip file containing the completed code: |
| |
| [.tabset] |
| Java:: |
| + |
| **** |
| * link:_attachments/3-shopping-cart-event-sourced-complete-java.zip[Source] that includes all previous tutorial steps and allows you to start with the steps on this page. |
| * link:_attachments/4-shopping-cart-projection-java.zip[Source] with the steps on this page completed. |
| **** |
| |
| Scala:: |
| + |
| **** |
| * link:_attachments/3-shopping-cart-event-sourced-complete-scala.zip[Source] that includes all previous tutorial steps and allows you to start with the steps on this page. |
| * link:_attachments/4-shopping-cart-projection-scala.zip[Source] with the steps on this page completed. |
| **** |
| |
| :sectnums: |
| == Process events in a Projection |
| |
| To process events in a projection, we will: |
| |
| * encapsulate database access with `ItemPopularityRepository`, which can have a stubbed implementation for tests |
| * add Repository implementation for JDBC |
| * implement the event processing of the Projection in a `Handler` |
| |
| [.group-java] |
| **** |
| This example uses Spring Data and Hibernate to access the database. You may use any other JDBC library to achieve this. |
| |
| The template provides a few classes for integration with Spring Data: `HibernateJdbcSession`, `SpringConfig`, `SpringIntegration` and `CreateTableTestUtils`. How these classes work won't be covered by this tutorial. Consult their respective source code in the template for more information on how they provide the necessary glue code. |
| **** |
| [.group-scala] |
| **** |
| This example uses the link:http://scalikejdbc.org/[ScalikeJDBC library {tab-icon}, window="tab"] to access the database.. You may use any other JDBC library to achieve this. |
| |
| The template provides a few classes for integration with ScalikeJDBC, for instance: `ScalikeJdbcSetup`, `ScalikeJdbcSession` and `CreateTableTestUtils`. How these classes work won't be covered by this tutorial. Consult their respective source code in the template for more information on how they provide the necessary glue code. |
| **** |
| |
| Follow these steps to process events in a Projection: |
| |
| |
| . Add the `ItemPopularityRepository`: |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/repository/ItemPopularityRepository.java: |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/repository/ItemPopularityRepository.java[] |
| ---- |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/repository/ItemPopularityRepository.scala: |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/repository/ItemPopularityRepository.scala[tag=trait] |
| ---- |
| |
| . [.group-java]#Add the `ItemPopularity`:# |
| + |
| [.group-java] |
| .src/main/java/shopping/cart/ItemPopularity.java: |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ItemPopularity.java[] |
| ---- |
| |
| . [.group-java]#Use the Spring `ApplicationContext` class to retrieve the implementation for the repository. This will be added later to the `Main` class.# [.group-scala]#Add the implementation for PostgresSQL by using the `ScalikeJdbcSession`:# |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/Main.java: |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/Main.java[tag=repo-instance] |
| ---- |
| + |
| <1> Initialize a Spring application context using the ActorSystem. |
| <2> Get a Spring generated implementation for the `ItemPopularityRepository`. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/repository/ItemPopularityRepository.scala: |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/repository/ItemPopularityRepository.scala[tag=impl] |
| ---- |
| |
| |
| . Add a class `ItemPopularityProjectionHandler`: |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/ItemPopularityProjectionHandler.java: |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ItemPopularityProjectionHandler.java[tags=handler] |
| ---- |
| + |
| <1> Extends `akka.projection.javadsl.JdbcHandler`. |
| <2> The `process` method to implement. |
| <3> Match events and increment or decrement the count via the `ItemPopularityRepository`, which encapsulates the database access. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/ItemPopularityProjectionHandler.scala: |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/ItemPopularityProjectionHandler.scala[tags=handler] |
| ---- |
| + |
| <1> Extends `akka.projection.scaladsl.JdbcHandler`. |
| <2> The `process` method to implement. |
| <3> Match events and increment or decrement the count via the `ItemPopularityRepository`. |
| |
| == Initialize the Projection |
| |
| We want to connect the events from the `ShoppingCart` with the Projection. Several instances of the Projection may run on different nodes of the Akka Cluster. Each Projection instance will consume a slice of the events to distribute the load. All events from a specific entity (cart id) will always be processed by the same Projection instance so that it can build a stateful model from the events if needed. |
| |
| [#tagging] |
| === Create tags |
| |
| To connect the events from the entities with the Projection we need to tag the events. We should use several tags, each with a slice number, to distribute the events over several Projection instances. The tag is selected based on the modulo of the entity id's hash code (stable identifier) and the total number of tags. Each entity instance will tag its events using one of those tags, and the entity instance will always use the same tag. |
| |
| |
| Create tags as follows: |
| |
| . Edit [.group-scala]#`ShoppingCart.scala`# [.group-java]#`ShoppingCart.java`# to include the following: |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/ShoppingCart.java |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ShoppingCart.java[tag=tagging] |
| ---- |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/ShoppingCart.scala |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/ShoppingCart.scala[tags=importEntityContext;tagging] |
| ---- |
| |
| + |
| One of the tags is selected based on the `cartId`, which is the `entityContext.entityId`. The tag is assigned to the `EventSourcedBehavior`. |
| |
| . [.group-scala]#In the `ShoppingCart.apply` method, add the `projectionTag` parameter and pass it to `.withTagger`:# [.group-java]#In the `ShoppingCart` constructor, add the `projectionTag` parameter and use it to override the `tagsFor` method:# |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/ShoppingCart.java |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ShoppingCart.java[tag=withTagger] |
| ---- |
| + |
| <1> Use `tagsFor` to assign the `projectionTag`. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/ShoppingCart.scala |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/ShoppingCart.scala[tag=withTagger] |
| ---- |
| + |
| <1> Use `withTagger` to assign the `projectionTag`. |
| |
| |
| NOTE: In this example, we use five different tags. Tagging is not easy to change later without system downtime. Before going live in production you should consider how many tags to use, see {akka-projection}/running.html[Akka Projections reference documentation {tab-icon}, window="tab"] for more information. |
| |
| [#projection] |
| === Create Projection |
| |
| To create the Projection: |
| |
| . Place the initialization code of the Projection in an `ItemPopularityProjection` [.group-scala]#object# [.group-java]#class#: |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/ItemPopularityProjection.java: |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ItemPopularityProjection.java[tag=projection] |
| ---- |
| + |
| <1> `ShardedDaemonProcess` will manage the Projection instances. It ensures the Projection instances are always running and distributes them over the nodes in the Akka Cluster. |
| <2> The `tag` is selected based on the Projection instance's index, corresponding to *carts-0* to *carts-3* as explained in the tagging in the `ShoppingCart`. |
| <3> The source of the Projection is `EventSourcedProvider.eventsByTag` with the selected tag. |
| <4> Using the JDBC event journal. |
| <5> Using JDBC for offset storage of the Projection using `exactly-once` strategy. Offset and projected model will be persisted transactionally. |
| <6> Define a `HibernateJdbcSession` factory. The JDBC connection are create by the projection and used to save the offset and the projected model. |
| <7> Define a Projection `Handler` factory for the handler we wrote in the beginning of this part. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/ItemPopularityProjection.scala: |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/ItemPopularityProjection.scala[tag=projection] |
| ---- |
| + |
| <1> `ShardedDaemonProcess` will manage the Projection instances. It ensures the Projection instances are always running and distributes them over the nodes in the Akka Cluster. |
| <2> The `tag` is selected based on the Projection instance's index, corresponding to *carts-0* to *carts-3* as explained in the tagging in the `ShoppingCart`. |
| <3> The source of the Projection is `EventSourcedProvider.eventsByTag` with the selected tag. |
| <4> Using the JDBC event journal. |
| <5> Using JDBC for offset storage of the Projection using `exactly-once` strategy. Offset and projected model will be persisted transactionally. |
| <6> Creating the Projection `Handler` that we wrote in the beginning of this part. |
| |
| . Call the `ItemPopularityProjection.init` from `Main`: |
| + |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/Main.java |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/Main.java[tag=ItemPopularityProjection] |
| ---- |
| + |
| <1> Initialize a Spring application context using the ActorSystem. |
| <2> Get a Spring generated implementation for the `ItemPopularityRepository`. |
| <3> Get a Spring `JpaTransactionManager` to pass to the Projection init method. |
| <4> Initialize the Projection passing all necessary dependencies. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/Main.scala |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/Main.scala[tag=ItemPopularityProjection] |
| ---- |
| + |
| <1> Call `ScalikeJdbcSetup.init` method to initiate the connection pool for the read-side. The connection pool will be closed when the actor system terminates. |
| <2> Instantiate the repository. |
| <3> Call the initialization of the Projection. |
| |
| == Query |
| |
| To expose the item popularity to the outside of the service we can add an operation in the gRPC `ShoppingCartService`. Follow these steps: |
| |
| . Add a new `GetItemPopularity` operation to the `ShoppingCartService.proto`: |
| + |
| .src/main/protobuf/ShoppingCartService.proto |
| [source,protobuf,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/protobuf/ShoppingCartService.proto[tag=GetItemPopularity] |
| ---- |
| |
| . Generate code from the new Protobuf specification by compiling the project: |
| + |
| [.tabset] |
| Java:: |
| + |
| [source,shell] |
| ---- |
| mvn compile |
| ---- |
| |
| Scala:: |
| + |
| [source,shell] |
| ---- |
| sbt compile |
| ---- |
| |
| . Add the `getItemPopularity` method to the `ShoppingCartServiceImpl`: |
| + |
| |
| For this you have to add the `ItemPopularityRepository` as a constructor parameter to the `ShoppingCartServiceImpl`. The `ItemPopularityRepository` instance is created in [.group-scala]#`Main.scala`# [.group-java]#`Main.java`# so pass that instance as parameter to `ShoppingCartServiceImpl`. |
| |
| [.tabset] |
| Java:: |
| + |
| .src/main/java/shopping/cart/ShoppingCartServiceImpl.java |
| [source,java,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/src/main/java/shopping/cart/ShoppingCartServiceImpl.java[tag=getItemPopularity] |
| ---- |
| + |
| <1> Add the `ItemPopularityRepository` to the service implementation constructor. |
| <2> Use the `ActorSystem` to get an instance of an `Executor` tailored for running JDBC blocking operations. |
| <3> Implement `getItemPopularity` by calling the repository to find the projected model by id and wrap it with `CompletableFuture.supplyAsync`. |
| |
| Scala:: |
| + |
| .src/main/scala/shopping/cart/ShoppingCartServiceImpl.scala |
| [source,scala,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/src/main/scala/shopping/cart/ShoppingCartServiceImpl.scala[tag=getItemPopularity] |
| ---- |
| + |
| <1> Add the `ItemPopularityRepository` to the service implementation constructor. |
| <2> Use the `ActorSystem` to get an instance of an `ExecutionContext` tailored for running JDBC blocking operations. |
| <3> Implement `getItemPopularity` by calling the repository to find the projected model by id and wrap it in a `Future` running on the executor for JDBC operations. |
| |
| [.group-java] |
| [IMPORTANT] |
| ==== |
| Calls to the repository are blocking therefore we wrap it with `CompletableFuture.supplyAsync` and we use the JDBC blocking executor to run it. This is a pre-configured dispatcher provided by Akka Projections. Its pool size can be configured with settings `akka.projection.jdbc.blocking-jdbc-dispatcher.thread-pool-executor.fixed-pool-size` (see src/main/resources/persistence.conf). |
| |
| Make sure to follow that pattern whenever you interact with a DB repository. |
| ==== |
| |
| [.group-scala] |
| [IMPORTANT] |
| ==== |
| Calls to the repository are blocking therefore we run them in a `Future` with the JDBC blocking execution context. This is a pre-configured dispatcher provided by Akka Projections. Its pool size can be configured with settings `akka.projection.jdbc.blocking-jdbc-dispatcher.thread-pool-executor.fixed-pool-size` (see src/main/resources/persistence.conf). |
| |
| Make sure to follow that pattern whenever you interact with a DB repository. |
| ==== |
| |
| == Run locally |
| |
| Try your solution by running locally: |
| |
| . Start the docker containers, unless they are already running: |
| + |
| [source,shell script] |
| ---- |
| docker-compose up -d |
| ---- |
| |
| . Create the item popularity table by creating a `ddl-scripts/create_user_tables.sql` file and adding the SQL statement below. |
| + |
| [.group-java] |
| [source,sql,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-java/ddl-scripts/create_user_tables.sql[] |
| ---- |
| |
| [.group-scala] |
| [source,sql,indent=0] |
| ---- |
| include::example$04-shopping-cart-service-scala/ddl-scripts/create_user_tables.sql[] |
| ---- |
| |
| . Load the file into Postgres: |
| + |
| [source,shell script] |
| ---- |
| docker exec -i shopping-cart-service_postgres-db_1 psql -U shopping-cart -t < ddl-scripts/create_user_tables.sql |
| ---- |
| + |
| include::partial$docker-important.adoc[] |
| |
| . Run the service with: |
| + |
| [.group-java] |
| [source,shell script] |
| ---- |
| # make sure to compile before running exec:exec |
| mvn compile exec:exec -DAPP_CONFIG=local1.conf |
| ---- |
| + |
| [.group-scala] |
| [source,shell script] |
| ---- |
| sbt -Dconfig.resource=local1.conf run |
| ---- |
| |
| === Exercise the service |
| |
| // # tag::exercise[] |
| Use `https://github.com/fullstorydev/grpcurl[grpcurl]` to exercise the service: |
| |
| . Add 5 hoodies to a cart: |
| + |
| [source,shell script] |
| ---- |
| grpcurl -d '{"cartId":"cart3", "itemId":"hoodie", "quantity":5}' -plaintext 127.0.0.1:8101 shoppingcart.ShoppingCartService.AddItem |
| ---- |
| |
| . Check the popularity of the item: |
| + |
| [source,shell script] |
| ---- |
| grpcurl -d '{"itemId":"hoodie"}' -plaintext 127.0.0.1:8101 shoppingcart.ShoppingCartService.GetItemPopularity |
| ---- |
| |
| . Add 2 hoodies to another cart: |
| + |
| [source,shell script] |
| ---- |
| grpcurl -d '{"cartId":"cart5", "itemId":"hoodie", "quantity":2}' -plaintext 127.0.0.1:8101 shoppingcart.ShoppingCartService.AddItem |
| ---- |
| |
| . Check that the popularity count increased to 7: |
| + |
| [source,shell script] |
| ---- |
| grpcurl -d '{"itemId":"hoodie"}' -plaintext 127.0.0.1:8101 shoppingcart.ShoppingCartService.GetItemPopularity |
| ---- |
| // # end::exercise[] |
| |
| === Stop the service |
| |
| When finished, stop the service with `ctrl-c`. Leave PostgresSQL running for the next set of steps, or stop it with: |
| |
| [source,shell script] |
| ---- |
| docker-compose stop |
| ---- |
| NOTE: The following steps for cloud deployment are optional. If you are only running locally, you can skip to the next section of the tutorial. |
| |
| [#kubernetes] |
| == Run in Kubernetes |
| |
| Create a xref:deployment:aws-install.adoc[Kubernetes cluster and install the Akka Operator] if you haven't already. |
| |
| === Create a Item Popularity table |
| |
| Create the item popularity table using the `ddl-scripts/create_user_tables.sql` SQL script. Follow the instructions in xref:deployment:jdbc.adoc[JDBC integration] to connect to your PostgreSQL instance and load the script. |
| |
| [source,shell script] |
| ---- |
| kubectl run -i rds-mgmt --image=postgres \ |
| --restart=Never --rm --env "PGPASSWORD=<password>" -- \ |
| psql -h <rds endpoint> -U postgres -t < ddl-scripts/create_user_tables.sql |
| ---- |
| |
| Make sure that `ddl-scripts/create_tables.sql` is also loaded as previously described. |
| |
| [NOTE] |
| ==== |
| If you created your database with the xref:deployment:aws-install-quickstart.adoc[] then you can reference the postgres username, password, and hostname using the `pulumi output` command (you must be in the Pulumi working directory for this to work), and reference the ddl script with an absolute path. |
| See the xref:deployment:aws-install-quickstart.adoc#connect-aurora-database[Connect to the Aurora RDS database {tab-icon}, window="tab"] section of the quick start for an example. |
| ==== |
| |
| Before following the steps below, create Kubernetes cluster and install the Akka Operator. Used the instructions below for: |
| |
| === Build Docker image |
| |
| Create a Docker repository and authenticate Docker. |
| |
| [.tabset] |
| GCP:: |
| + |
| Follow the instructions in https://cloud.google.com/container-registry/docs/using-with-google-cloud-platform[Using Container Registry with Google Cloud {tab-icon}, window="tab"] to deploy Docker images on GCP's container registry. |
| |
| AWS:: |
| + |
| Follow the instructions in xref:deployment:aws-ecr.adoc[Amazon Elastic Container Registry] to deploy Docker images on AWS's container registry. |
| |
| === Additional steps for Docker and AWS |
| |
| If you are using AWS, you will also need to complete the following procedures. |
| |
| Rebuild the Docker image. |
| |
| include::partial$build-docker-for-kube.adoc[] |
| |
| === Update the deployment descriptor |
| |
| Update the `kubernetes/shopping-cart-service-cr.yml` deployment descriptor with the new image tag as previously described. |
| |
| === Apply to Kubernetes |
| |
| Re-apply the `shopping-cart-service-cr.yml` to Kubernetes: |
| |
| [source,shell script] |
| ---- |
| kubectl apply -f kubernetes/shopping-cart-service-cr.yml |
| ---- |
| |
| You can see progress by viewing the status: |
| |
| [source,shell script] |
| ---- |
| kubectl get akkamicroservices/shopping-cart-service |
| ---- |
| |
| See xref:deployment:troubleshooting.adoc[troubleshooting deployment status] for more details. |
| |
| === Exercise the service in Kubernetes |
| |
| include::partial$prepare-to-exercise-in-kube.adoc[] |
| |
| include::projection-query.adoc[tag=exercise] |
| |
| [NOTE] |
| ==== |
| If you decide to compile and run your tests you may encounter aan error `reason: actual and formal argument lists differ in length`. If this should occur you may bypass the error by adding an arbitrary tag to the shopping cart creation within the test. |
| ==== |
| |
| :!Sectnums: |
| == Learn more |
| |
| * xref:concepts:cqrs.adoc[CQRS concepts]. |
| * {akka-projection}/[Akka Projection reference documentation {tab-icon}, window="tab"]. |
| * {akka}/typed/cluster-sharded-daemon-process.html[Akka Sharded Daemon Process reference documentation {tab-icon}, window="tab"]. |
| * xref:how-to:cassandra-alternative.adoc[] |
