BatchServiceis in the Real Estate IT Industry

BatchService

batchservice

The journey to implementing artificial intelligence and machine learning solutions requires solving a lot of common challenges that routinely crop up in digital systems: updating legacy systems, eliminating batch processes, and using innovative technologies that are grounded in AI/ML to improve the customer experience in ways that seemed like science fiction just a few years ago. To illustrate this evolution, let’s follow a hypothetical contractor who was hired to help implement AI/ML solutions at a big-box retailer. This is the first in a series of articles that will detail important aspects of the journey to AI/ML. The Problem: A Brittle Batch Process It’s the first day at BigBoxCo on the “Infrastructure” team. After working through the obligatory human resources activities, I received my contractor badge and made my way over to my new workspace. After meeting the team, I was told that we have a meeting with the “Recommendations” team this morning. My system access isn’t quite working yet, so hopefully, IT will get that squared away while we’re in the meeting. In the meeting room, it’s just a few of us: my manager and two other engineers from my new team, and one engineer from the Recommendations team. We start off with some introductions and then move on to discuss an issue from the week prior. Evidently, there was some kind of overnight batch failure last week, and they’re still feeling the effects of that. It seems like the current product recommendations are driven by data collected from customer orders. With each order, there’s a new association between the products ordered, which is recorded. When customers view product pages, they can get recommendations based on how many other customers bought the current product alongside different products. The product recommendations are served to users on bigboxco.com via a microservice layer in the cloud. The microservice layer uses a local (cloud) data center deployment of to serve up the results. Apache Cassandra How the results are collected and served, though, is a different story altogether. Essentially, the results of associations between products (purchased together) are compiled during a MapReduce job. This is the batch process that failed last week. While this batch process has never been fast, it has become slower and more brittle over time. In fact, sometimes the process takes two or even three days to run. Improving the Experience After the meeting, I check my computer, and it looks like I can finally log in. As I’m looking around, our principal engineer (PE) comes by and introduces himself. I tell him about the meeting with the Recommendations team, and he gives me a little more of the history behind the Recommendation service. It sounds like that batch process has been in place for about 10 years. The engineer who designed it has moved on, not many people in the organization really understand it, and nobody wants to touch it. The other problem, I begin to explain, is that the dataset driving each recommendation is almost always a couple of days old. While this might not be a big deal in the grand scheme of things, if the recommendation data could be made to be more up-to-date, it would benefit the short-term promotions that marketing runs. He nods in agreement and says that he’s definitely open to suggestions on how to improve the system. Maybe It’s a Graph Problem? At the onset, this sounds to me like a graph problem. We have customers who log on to the site and buy products. Before that, when they look at a product or add it to the cart, we can show recommendations in the form of “Customers who bought X also bought Y.” The site has this today, in that the recommendations service does exactly this: It returns the top four additional products that are frequently purchased together. But we’d have to have some way to “rank” the products because the mapping of one product to every other purchased at the same time by any of our 200 million customers is going to get big, fast. So we can rank them by the number of times they appear in an order. A graph of this system might look something like what is shown below in Figure 1. After modeling this out and running it on our graph database with real volumes of data, I quickly realized that this isn’t going to work. The traversal from one product to nearby customers to their products and computing the products that appear most takes somewhere in the neighborhood of 10 seconds. Essentially, we’ve “punted” on the two-day batch problem, to have each lookup putting the traversal latency precisely where we don’t want it: in front of the customer. But perhaps, that graph model isn’t too far off from what we need to do here? In fact, the approach described above is a machine learning (ML) technique known as “collaborative filtering.” Essentially, collaborative filtering is an approach that examines the similarity of certain data objects based on activity with other users, and it enables us to make predictions based on that data. In our case, we will be implicitly collecting cart/order data from our customer base, and we will use it to make better product recommendations to increase online sales. Implementation First of all, let’s look at data collection. Adding an extra service call to the shopping “place order” function isn’t too big of a deal. In fact, it already exists; it’s just that data gets stored in a database and processed later. Make no mistake: We still want to include the batch processing. But we’ll also want to process that cart data in real-time, so we can feed it right back into the online data set and use it immediately afterward. We’ll start out by putting in an event streaming solution like . That way, all new cart activity is put on a , where it is consumed and sent to the underlying batch database as well as to help train our real-time ML model. Apache Pulsar Pulsar topic both As for the latter, our Pulsar consumer will write to a Cassandra table (shown in Figure 2) designed simply to hold entries for each product in the order. The product then has a row for all of the other products from that and other orders: CREATE TABLE order_products_mapping (
    id text,
    added_product_id text,
    cart_id uuid,
    qty int,
    PRIMARY KEY (id, added_product_id, cart_id)
) WITH CLUSTERING ORDER BY (added_product_id ASC, cart_id ASC); We can then query this table for a particular product (“DSH915” in this example), like this: SELECT added_product_id, SUM(qty)
FROm order_products_mapping
WHERE id='DSH915'
GROUP BY added_product_id;

 added_product_id | system.sum(qty)
------------------+-----------------
            APC30 |               7
           ECJ112 |               1
            LN355 |               2
            LS534 |               4
           RCE857 |               3
          RSH2112 |               5
           TSD925 |               1

(7 rows) We can then take the top four results and put them into the product recommendations table, ready for the recommendation service to query by ` `: product_id SELECT * FROM product_recommendations
WHERE product_id='DSH915';

 product_id | tier | recommended_id | score
------------+------+----------------+-------
     DSH915 |    1 |          APC30 |     7
     DSH915 |    2 |        RSH2112 |     5
     DSH915 |    3 |          LS534 |     4
     DSH915 |    4 |         RCE857 |     3

(4 rows) In this way, the new recommendation data is constantly being kept up to date. Also, all of the infrastructure assets described above are located in the local data center. Therefore, the process of pulling product relationships from an order, sending them through a Pulsar topic, and processing them into recommendations stored in Cassandra happens in less than a second. With this simple data model, Cassandra is capable of serving the requested recommendations in single-digit milliseconds. Conclusions and Next Steps We’ll want to be sure to examine how our data is being written to our Cassandra tables in the long term. This way, we can get ahead of potential problems related to things like unbound row growth and in-place updates. Some additional heuristic filters may be necessary to add as well, like a “do not recommend” list. This is because there are some products that our customers will buy either once or infrequently, and recommending them will only take space away from other products that they are much more likely to buy on impulse. For example, recommending a purchase of something from our appliance division such as a washing machine is not likely to yield an “impulse buy.” Another future improvement would be to implement a real-time AI/ML platform like to handle both the product relationship streaming and to serve the recommendation data to the service directly. Kaskada Fortunately, we did come up with a way to augment the existing, sluggish batch process using Pulsar to feed the cart-add events to be processed in real-time. Once we get a feel for how this system performs in the long run, we should consider shutting down the legacy batch process. The PE acknowledged that we made good progress with the new solution, and, better yet, that we have also begun to lay the groundwork to eliminate some technical debt. In the end, everyone feels good about that. In an upcoming article, we’ll take a look at improving product promotions with . vector search Learn how DataStax enables real-time AI By Aaron Ploetz, DataStax Also published here

Implementing AI in Retail With Apache Cassandra and Apache Pulsar

Implementación de IA en el comercio minorista con Apache Cassandra y Apache Pulsar

Implémentation de l'IA dans le commerce de détail avec Apache Cassandra et Apache Pulsar

अपाचे कैसेंड्रा और अपाचे पल्सर के साथ रिटेल में एआई को लागू करना

Apache Cassandra と Apache Pulsar を使用した小売業への AI の実装

Implementando AI no varejo com Apache Cassandra e Apache Pulsar

Triển khai AI trong bán lẻ với Apache Cassandra và Apache Pulsar

使用 Apache Cassandra 和 Apache Pulsar 在零售业实施人工智能

State synchronization across connected virtual-worlds Multiplayer immersive game development is one of the most interesting projects for indie creators and teams alike, presenting challenges covering a wide-spectrum from user experience to distributed computing.From Minecraft to WoW open ended have entertained and educated us and created communities for gamers and trolls alike and also pushed the boundaries of our and . MMOs collective creativity imagination King’s landing recreated in Minecraft [ ] Source In recent years, MMO game-dev has seen a transition from intricate worlds created via hundreds of thousands of lines-of-code to player-defined worlds which provide building blocks for creating persistant artifacts and experiences. As the cutting edge in game economics, storage and UX technologies hit mainstream applications, it’s not overtly optimistic to assume that we’ll see some truly amazing MMO gameplay and immersive virtual worlds in the time to come.. For some perspective, WoW at its peak had monthly active players. LoL which is the largest online Multiplayer game has over a monthly active players (Although it is not really a MMO and maintains rooms of 10 players hosted on a single shard). By contrast, Youtube has monthly active users and Facebook has monthly active users. Last year during Oculus conference, Mark Zuckerberg gave a of things to come… 12 million 100 million 1.5 Billion 2.2 billion teaser Why you s̶h̶o̶u̶l̶d̶n̶’̶t̶ absolutely should be creating a MMO — Client SDKs, game-networking and cloud-native database stacks provide standardized solutions for most common access-patterns. It has historically never been a better time for creative professionals and programmers to come together and create MMO prototypes for emerging clients. Crowd-funds and ICOs have often removed the barriers the for indie game developers and small teams to go from prototype to production. Here are some major technological and social factors that will propel immersive world applications into the future. 1. Liquidity for game tokens in the real world Liquid game tokens can be created on the blockchain It’s far too long been the case that the only multiplayer games outside of big production studios making money were the ones which provide services with such as Poker or Betting games..Game currencies, such as (the currency of Second Life) while pioneering struggled to gain credibility and acceptance.. mostly because they were created out of thin air by game studios…many recently published white-papers and proof-of-concepts illustrate that game token and are one of the most natural use-cases of the blockchain. For instance, if the currency earned through industry in the virtual world can then be used in the real world and vice-versa and we’ll see many more interesting artifacts and development in immersive massively multiplayer landscapes. imagine you being able to hold an auction for the house or the bridge you built in Minecraft in game tokens and being able to convert the game tokens immediately to . Fiat Linden Dollars virtual world currencies Fiat 2. Production ready VR SDKs Good news everyone ..and not just through netsuits Virtual Reality and Augmented reality have finally crossed the peak of inflated expectations and are almost past the trough of disillusionment of the the G as real world usages are being tested and factories are creating production lines. From Unity and Unreal Engine major game engines are providing VR app client SDKs natively and vendors are already churning out VR devices for niche use-cases. artner hype cycle 3. Micro-services enabled seamless gameplay across clients The most played MMOs will provide cross-client experiences When the app middleware is built as micro-services, it allows you-the game developer to support gameplay for multiple game clients and progressively enhance game play for more advanced clients. Imagine an MMO where players can do the full spectrum of actions such as participate in battles via the oculus rift client but are also able to chat with other players or water their garden via the android client. 4. Cloud-native, auto-sharded distributed storage engines Cloud native databases provide consistent low latency reads and writes across regions and zones Network latency has been the issue with playing online games even at moderately high FPS. This is because any network traffic over the public internet will incur latency resulting in considerable lag. Using dedicated networking provided by cloud service providers, we can limit the interaction with public internet to the last few hops, drastically improving P99 latencies over a large number of reads and writes. Advances in distributed database technology supporting multiple access patterns and providing fast reads and writes means that a lot of game networking and load-balancing logic can move from the middleware, down to the data layer. Game middleware architecture 1. Authoritative servers and dumb clients To the user the client is the game, clients will render all the graphics and allow user actions, the game client can be as rich as or or as simple as . The client should not however authoritatively execute any logic based on user action and must act on values provided by the server. The reason for this is that we must always assume that the client is an attacker when developing a game in multiplayer contexts. For instance if the client were allowed to dictate it’s position and propagate it to other clients, it may become trivially easy for a malicious player to hack their coordinates and set it to be on the enemy flag and propagate it’s position to other very confused clients. It’s also easy to see how an authoritative server architecture scales better than authoritative client architectures. Runescape WoW slitherio Instead for every user action, the client must encode the state into a data structure and pass it down to the server which then interprets it against the rules of the game to see if the action makes sense, it then sends the position back to client which then regards it as it’s final position.Additionally the server might take a batch of such inputs and run it against an analytics and inference engine to verify player actions. For instance did the player just hit 10 out his last 10 shots, are the mouse clicks on the screen differing from input coordinates? 2. Network latency and state propagation Having authoritative server means that clients cannot cheat but it also means that the when you press the key W, the client sends the request and waits for the server response to tell it to move forward, when this happens over the internet it will visibly cause lag in the game experience. To get over this game clients use several smoothening techniques such as client side prediction, server reconciliation and interpolation. This article does not go into the details of client side compensation techniques, you can check out this here which also includes this . excellent explanation demo On the server side, we can reduce network latency by limiting the number of hops made over the public internet and by using high speed network peering provided by IaaS providers as much as possible. supports single click deployment and scale-up/down operations on primary nodes and read replicas in both AWS and GCP, while leveraging VPC peering to enable high speed networking and private traffic across regions and data centers. YugaByte 3. Server latency and linear scale-out models So how come we can have millions of people viewing the same webpage but only thousands on the same shard in an MMO? In the article titled , the author explains how the network and server load multiplies exponentially on a MMO shard and the problems in handling state consensus in the application layer. MMO Population Problems Scaling out from 25 to 50 nodes doubles read and write throughput Source DB solves the problem of scale by providing models that linearly scale-out instead of scaling up. For instance doubling the size of a 25 node cluster to 50 nodes causes the performance and throughput to double. You read the benchmark details . YugaByte here 4. Automatic sharding and load balancing When creating game databases, teams are often faced with tough questions about the physical and logical partition of the data servers like — how many databases should I have per node? How should I decide which nodes I need to scale faster than others. This creates a chicken and egg problem in which you need to know access patterns of game interaction before you have actually built the game. simplifies this for game developers by providing auto-sharding data models from the outset that let’s you elastically scale out your cluster as your game grows. YugaByte The game servers could simply sit behind dumb load-balancers serving requests from nearest clients, which will in turn return the most recent consistent snapshot of the data maintained via . Load balancing is handled on the data layer through which automatically splits tables based on game models into multiple shards. This allows your team to not have to manage load-balancing in the application layer and your architecture would look more and more like if you were developing a single player game. MVCC auto-sharding Sharding of data into tablets. Source 5. Read replicas for high-speed read access patterns MMOs access patterns usually follow frequent reads (for all the actors in the grid) and infrequent writes (for player actions)In many access patterns where the actor is not writing but only reading, timeline consistent reads from async replicas may provide lower latencies. For instance if the player is not doing anything and only needs to read the data of all actors in their grid of reference, they can get timeline consistent copies from a Read-Replica. Read-Replicas may also be used to run cluster aware analytics jobs on live data to infer real time intelligence from player actions without straining the primary nodes. Efficiency of modeling APIs Games are computationally taxing applications and often test the limits of clients, networks and servers (on a scale), which is also the reason why efficiency matters most just just in the space it takes store the dirty state but also the code needed to serialize/deserialize it. Most MMO games follow access patterns dictating fast reads for the state of all objects in the player’s grid and fast write for the player’s information when the player does an action. Redis API provides simple constructs to store game state objects with minimal serialization/deserialization costs which can be used to build more complex structures. Often in browser games, the game state is passed down as a JSON document via HTTP. 1. Redis Modeling Redis API provides developers with a set of very useful data structures such as sets, sorted sets, and hash maps that can be used to build more advanced structures that model most state transitions in MMO objects efficiently. However traditional Redis implementations as a pure in-memory cache has its own limitations, the first is that the entire data-set needs to fit in-memory which might be a challenge for game servers with high user activity in a single shard. Another layer of complication is that implementing Redis + DB models requires further access pattern logic integrated into the application layer. solves this problem by providing a persistant Redis with Block-caches serving hot reads while providing sub-ms latency even for workloads that . You can read more about implementation . YugaByte don’t fit the cache YugaByte Redis here 2. CQL Modeling Fast access patterns for large datasets or those requiring strongly consistent distributed transactions such as marketplace transactions can be modeled over the CQL API. Moreover access patterns requiring range scans or queries that intuitively models the column family structure can be modeled via CQL API over the DocDB storage engine. YugaByte also supports Graph API access patterns via and APIs for time series metrics. Janusgraph KairosDB 3. JSON Modeling MMO game states especially for web-based games are also often intuitively modeled as the JSON documents, Support for JSON type with Postgres API compatibility is in the works.You can find details about the PR . here Real-time Analytics over game state Any commercially viable virtual world will be running real-time analytics jobs on game data to gather intelligence on game play improvements, targeted advertising, cheating detection etc. Some examples of questions that you’ll likely be asking are — Is the matchmaker service matching players according to their skill levels? Are some players cheating? If so how are they cheating? What can I infer about player’s likes/dislikes etc. from gameplay patterns? Are players clicking on in-game advertisements? Real time analytics using the allows the services to be designed on top of the data layers which lets you infer meaningful knowledge from player actions. SKY stack Simplifying the MMO data and operations stack Multiplayer game networking and state propagation is hard and definitely not something your team wants to focus on. While staying within the realm of what is feasible you want your models to be defined by providing the best user experience and not necessarily for minimizing data packets. DB can simplify a game engine server stack by combining the caching and persistance layer into one, pushing the load-balancing and sharding logic down from the application layer to the data layer and use the speed of Redis with persistance guarantees for in-game state and CQL APIs for Player data and web-service models. YugaByte About me — I am a developer at where we build data solutions for distributed, high-density applications and games. If you or your team are building immersive virtual-world experiences and exploring scale-out , you can check out the docs . YugaByte data-models here Meanwhile you can head over to the Github project and take a deep dive into the code base.

Fast-paced Authoritative MMO Data-Models

Recently, I faced with lack of documentation when I wanted to use Consul as a host resolver in gRPC connections. That’s why I want to fill the hole for engineers who struggle with the same problem. So, let’s start with a little bit of theory. Why gRPC is a trendy RPC framework developed by Google. It reached its popularity cause of several reasons: gRPC it’s extremely fast because the Protocol Buffers (protobuf), a powerful binary serialization toolset and language it generates the client/server code by a single protobuf contract. it works with HTTP/2 and supports BI-directional data streams Why Consul is a robust service mesh developed by HashiCorp. There are a lot of use-cases for Consul: Consul service discovery config storage key-value storage In our project, Consul keeps info about all microservice hosts. So when a microservice instance goes down, or there is a new one — Consul knows it immediately. So we want to use the information to resolve a host for any gRPC interaction to ensure that microservices are connected to actual healthy hosts. First non-optimal solution The main problem is that there is no official documentation on resolving hosts by Consul in gRPC, so we tried to Google a solution. There was a suggestion to implement a host resolver by ourselves. OK, let’s do it: Golang — is the primary language in our tech stack, so all written code will be in Go. package grpcclient

import (
	"errors"
	"fmt"
	"net/url"
	"strings"

	"utils/pkg/connection"
	"utils/pkg/consul"

	"github.com/hashicorp/consul/api"
	"github.com/sirupsen/logrus"
	"google.golang.org/grpc/resolver"
)

const (
	resolverSchemeConsul = "consul"
)

var (
	errUnknownScheme = errors.New("unknown scheme. Only 'consul' is applicable")
)

// consulBuilder builds the address resolver for gRPC dialer.
type consulBuilder struct {
	consul         *consul.Cluster
	serviceWatcher connection.Watcher
}

// newConsulBuilder is a constructor.
func newConsulBuilder(consul *consul.Cluster, serviceWatcher connection.Watcher) *consulBuilder {
	return &consulBuilder{
		consul:         consul,
		serviceWatcher: serviceWatcher,
	}
}

// Builds the consul address resolver for gRPC.
func (c consulBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	if target.Scheme != resolverSchemeConsul {
		return nil, errUnknownScheme
	}

	serviceName, tag, err := parseTarget(target)
	if err != nil {
		return nil, fmt.Errorf("parse target: %w", err)
	}

	cr := newConsulResolver(c.serviceWatcher, cc)
	err = cr.watch(serviceName, []string{tag})
	if err != nil {
		return nil, fmt.Errorf("watch: %w", err)
	}

	return cr, nil
}

// parses the target and returns service name and tag.
// example of target endpoint: 127.0.0.1:8500/service?tag=master
func parseTarget(target resolver.Target) (string, string, error) {
	u, err := url.Parse(fmt.Sprintf("%s://%s", resolverSchemeConsul, target.Endpoint))
	if err != nil {
		return "", "", fmt.Errorf("parse: %w", err)
	}

	return strings.Trim(u.Path, "/"), u.Query().Get("tag"), nil
}

// Scheme returns the consul resolver scheme.
func (c consulBuilder) Scheme() string {
	return resolverSchemeConsul
}

type consulResolver struct {
	cc             resolver.ClientConn
	serviceWatcher connection.Watcher
	unwatchFunc    func() error
}

func newConsulResolver(serviceWatcher connection.Watcher, cc resolver.ClientConn) *consulResolver {
	return &consulResolver{
		serviceWatcher: serviceWatcher,
		cc:             cc,
	}
}

func (c *consulResolver) onServiceChanged(entries []*api.ServiceEntry) {
	addresses := make([]resolver.Address, len(entries))
	for i, e := range entries {
		addresses[i] = resolver.Address{
			Addr: connection.BuildAddr(e),
		}
	}

	err := c.cc.UpdateState(resolver.State{
		Addresses: addresses,
	})
	if err != nil {
		logrus.WithError(err).Error("update gRPC consul resolver addresses")
	}
}

func (c *consulResolver) watch(serviceName string, tags []string) error {
	var err error
	c.unwatchFunc, err = c.serviceWatcher.WatchService(serviceName, tags, true, c.onServiceChanged)
	if err != nil {
		return fmt.Errorf("watch service: %w", err)
	}

	return nil
}

// ResolveNow we don't need to anything here because all addresses are updated on change in consul.
func (c *consulResolver) ResolveNow(options resolver.ResolveNowOptions) {}

// Close is an interface method. Gracefully closes the consulResolver.
func (c *consulResolver) Close() {
	err := c.unwatchFunc()
	if err != nil {
		logrus.WithError(err).Error("unwatch service in gRPC consul resolver")
	}
} In that case, we need to specify the custom host resolver on dialing: import (
  ...
  "google.golang.org/grpc"
)
...

// Dial dials to a service through gRPC and returns a new connection.
func Dial(serviceName, tag string, timeout time.Duration) (grpc.ClientConnInterface, error) {
    cfg := consul.Config()
    target := fmt.Sprintf("%s:///%s/%s?tag=%s", resolverSchemeConsul, cfg.Address, serviceName, tag)

	return grpc.Dial(
		target,
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
		grpc.WithResolvers(newConsulBuilder(consul, consul.ServiceWatcher())),
		grpc.WithInsecure(),
	)
} The implementation works, but we wrote some tricky code and must maintain it. It looks like there is a better way to do that. The undocumented, optimal solution don’t contain Consul info, so I found a good solution accidentally while doing some job. The official gRPC name resolution docs As it turned out, we don’t need a custom host resolver for Consul, and gRPC has the built integration. To use it, we need to specify a dial target correctly: import (
  ...
  "google.golang.org/grpc"
  _ "github.com/mbobakov/grpc-consul-resolver"
)
...

// Dial dials a service through gRPC and returns a new connection.
func Dial(serviceName, tag string, timeout time.Duration) (grpc.ClientConnInterface, error) {
    cfg := consul.Config()
    target := fmt.Sprintf("consul://%s:%s@%s/%s?tag=%s", cfg.User, cfg.Password, cfg.Address, serviceName, tag)

	return grpc.Dial(
		target,
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
		grpc.WithInsecure(),
	)
} The dialing target will be something like: . consul://user:passsword@127.0.0.1:8500/service_name?tag=service_tag And that’s it. No need to write any code. Conclusion I want to summarize some conclusions based on the story: the first obvious solution from Google to a problem sometimes isn’t the best if a lib is very popular, it can still have a lack of docs, so carefully explore the library internals before implementing your improvements but it’s still OK to come up with a non-optimal solution at the first because all great products are built iteratively from scratch

Stories, News & Mentions about BATCHSERVICE

Batchservice’s Stories on HackerNoon