{"id":16493,"date":"2018-04-03T15:07:00","date_gmt":"2018-04-03T15:07:00","guid":{"rendered":"http:\/\/www.gamasutra.com\/view\/news\/316041"},"modified":"2018-04-03T15:07:00","modified_gmt":"2018-04-03T15:07:00","slug":"blog-building-a-fully-managed-game-analytics-pipeline","status":"publish","type":"post","link":"https:\/\/sickgaming.net\/blog\/2018\/04\/03\/blog-building-a-fully-managed-game-analytics-pipeline\/","title":{"rendered":"Blog: Building a fully-managed game analytics pipeline"},"content":{"rendered":"

The following blog post, unless otherwise noted, was written by a member of Gamasutra\u2019s community.
The thoughts and opinions expressed are those of the writer and not Gamasutra or its parent company.<\/small><\/em><\/strong><\/p>\n

\n

Gathering usage\u00a0data such as player progress in games is invaluable for teams. Typically, entire teams have been dedicated to building and maintaining data pipelines for collecting and storing tracking data for games. However, with many new serverless tools available, the barriers to building an analytics pipeline for collecting game data have been significantly reduced. Managed tools such as Google\u2019s PubSub, DataFlow, and BigQuery have made it possible for a small team to set up analytics pipelines that can scale to a huge volume of events, while requiring minimal operational overhead. This post describes how to build a lightweight game analytics pipeline on the Google Cloud platform (GCP) that is fully-managed (serverless) and auto scales to meet demand.\u00a0<\/p>\n

I was inspired by Google\u2019s\u00a0reference architecture<\/a>\u00a0for mobile game analytics. The goal of this post is to show that it\u2019s possible for a small team to build and maintain a\u00a0data pipeline<\/a>\u00a0that scales to large event volumes, provides a data lake for data science tasks, provides a query environment for analytics teams, and has extensibility for additional components such as an experiment framework for applications.\u00a0<\/p>\n

The core piece of technology I\u2019m using to implement this data pipeline is Google\u2019s DataFlow, which is now integrated with the\u00a0Apache Beam<\/a>\u00a0library. DataFlow tasks define a graph of operations to perform on a collection of events, which can be streaming data sources. This post presents a DataFlow task implemented in Java that streams tracking events from a PubSub topic to a data lake and to BigQuery. An introduction to DataFlow and it\u2019s concepts is available in\u00a0Google\u2019s documentation<\/a>. While DataFlow tasks are portable, since they are now based on Apache Beam, this post focuses on how to use DataFlow in conjunction with additional managed services on GCP to build a simple, serverless, and scalable data pipeline for storing game events.<\/p>\n

<\/p>\n

My lightweight implementation of the GCP Reference Architecture for Analytics.<\/em><\/p>\n

The data pipeline that performs all of this functionality is relatively simple. The pipeline reads messages from PubSub and then transforms the events for persistence: the BigQuery portion of the pipeline converts messages to TableRow objects and streams directly to BigQuery, while the AVRO portion of the pipeline batches events into discrete windows and then saves the events to Google Storage. The graph of operations is shown in the figure below.<\/p>\n

<\/p>\n

The streaming pipeline deployed to Google\u00a0Cloud<\/em><\/p>\n

Setting up the Environment<\/strong><\/h2>\n

The first step in building a game data pipeline is setting up the dependencies necessary to compile and deploy the project. I used the following maven dependencies to set up environments for the tracking API that sends events to the pipeline, and the data pipeline that processes events.\u00a0
\u00a0<\/p>\n

Dependencies for the Tracking API<\/em> ->
\u00a0 com.google.cloud
\u00a0 google-cloud-pubsub
\u00a0 0.32.0-beta
\u00a0 <\/p>\n

\u200b<!– Dependencies for the data pipeline<\/em> ->
<dependency>
\u00a0 <groupId>com.google.cloud.dataflow<\/groupId>
\u00a0 <artifactId>google-cloud-dataflow-java-sdk-all<\/artifactId>
\u00a0 <version>2.2.0<\/version>
<\/dependency><\/p>\n<\/div>\n

I used Eclipse to author and compile the code for this tutorial, since it is open source. However, other IDEs such as\u00a0IntelliJ<\/a>\u00a0provide additional features for deploying and monitoring DataFlow tasks. Before you can deploy jobs to Google Cloud, you\u2019ll need to set up a service account for both PubSub and DataFlow. Setting up these credentials is outside the scope of this post, and more details are available in the\u00a0Google documentation<\/a>.<\/p>\n

An additional prerequisite for running this data pipeline is setting up a PubSub topic on GCP. I defined a\u00a0raw-events<\/em>\u00a0topic that is used for publishing and consuming messages for the data pipeline. Additional details on creating a PubSub topic are available\u00a0here<\/a>.<\/p>\n

To deploy this data pipeline, you\u2019ll need to set up a java environment with the maven dependencies listed above, set up a GCP\u00a0project and enable billing, enable billing on the storage and BigQuery services, and create a PubSub topic for sending and receiving messages. All of these managed services do cost money, but there is a free tier that can be used for prototyping a data pipeline.\u00a0<\/p>\n

<\/p>\n

Sending events from a\u00a0game server to a PubSub\u00a0topic<\/em><\/p>\n

Publishing Events<\/h2>\n

In order to build a usable data pipeline, it\u2019s useful to build APIs that encapsulate the details of sending game events. The\u00a0Tracking API<\/a>\u00a0class provides this functionality, and can be used to send generated event data to the data pipeline. The code below shows the method signature for sending game events, and shows how to generate sample data.\u00a0
\u00a0<\/p>\n

\/** Event Signature for the Tracking API *\/
\/\/ sendEvent(String eventType, String eventVersion, HashMap attributes);<\/p>\n

\/\/ send a batch of events\u00a0\u00a0\u00a0
for (int i=0; i<10000; i++) {<\/p>\n

\u00a0 \/\/ generate event names\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 String eventType = Math.random() < 0.5 ?
\u00a0\u00a0\u00a0\u00a0\u00a0 “Session” : (Math.random() < 0.5 ? “Login” : “MatchStart”);<\/p>\n

\u00a0 \/\/ create attributes to send\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 HashMap<String, String> attributes = new HashMap<String,String>();
\u00a0 attributes.put(“userID”, “” + (int)(Math.random()*10000));
\u00a0 attributes.put(“deviceType”, Math.random() < 0.5 ?
\u00a0\u00a0\u00a0\u00a0\u00a0 “Android” : (Math.random() < 0.5 ? “iOS” : “Web”));<\/p>\n

\u00a0 \/\/ send the event\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 tracking.sendEvent(eventType, “V1”, attributes);\u00a0\u00a0\u00a0\u00a0\u00a0
}<\/p>\n<\/div>\n

The tracking API establishes a connection to a PubSub topic, passes game events as a JSON format, and implements a callback for notification of delivery failures. The code used to send events is provided below, and is based on Google\u2019s PubSub example provided\u00a0here<\/a>.
\u00a0<\/p>\n

\/\/ Setup a PubSub connection
TopicName topicName = TopicName.of(projectID, topicID);
Publisher publisher = Publisher.newBuilder(topicName).build();<\/p>\n

\/\/ Specify an event to send
String event = {\\”eventType\\”:\\”session\\”,\\”eventVersion\\”:\\”1\\”}”;<\/p>\n

\/\/ Convert the event to bytes\u00a0\u00a0\u00a0
ByteString data = ByteString.copyFromUtf8(event.toString());<\/p>\n

\/\/schedule a message to be published\u00a0\u00a0\u00a0
PubsubMessage pubsubMessage =
\u00a0 PubsubMessage.newBuilder().setData(data).build();<\/p>\n

\u200b\/\/ publish the message, and add this class as a callback listener
ApiFuture<String> future = publisher.publish(pubsubMessage);\u00a0\u00a0\u00a0
ApiFutures.addCallback(future, this);<\/p>\n<\/div>\n

The code above enables games\u00a0to send events to a PubSub topic. The next step is to process this events in a fully-managed environment that can scale as necessary to meet demand.<\/p>\n

Storing Events<\/h2>\n

One of the key functions of a game data pipeline is to make instrumented events available to data science and analytics teams for analysis. The data sources used as endpoints should have low latency and be able to scale up to a massive volume of events. The data pipeline defined in this tutorial shows how to output events to both BigQuery and a data lake that can be used to support a large number of analytics business users.\u00a0<\/p>\n

<\/p>\n

Streaming event data from PubSub to\u00a0DataFlow<\/em><\/p>\n

The first step in this data pipeline is reading events from a PubSub topic and passing ingested messages to the DataFlow process. DataFlow provides a PubSub connector that enables streaming of PubSub messages to other DataFlow components. The code below shows how to instantiate the data pipeline, specify streaming mode, and to consume messages from a specific PubSub topic. The output of this process is a collection of PubSub messages that can be stored for later analysis.
\u00a0<\/p>\n

\/\/ set up pipeline options\u00a0\u00a0\u00a0
Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);\u00a0\u00a0\u00a0
options.setStreaming(true);\u00a0\u00a0\u00a0
Pipeline pipeline = Pipeline.create(options);<\/p>\n

\u200b\/\/ read game events from PubSub\u00a0\u00a0\u00a0
PCollection<PubsubMessage> events = pipeline
\u00a0 .apply(PubsubIO.readMessages().fromTopic(topic));<\/p>\n<\/div>\n

The first way we want to store game events is in a columnar format that can be used to build a data lake. While this post doesn\u2019t show how to utilize these files in downstream ETLs, having a data lake is a great way to maintain a copy of your data set in case you need to make changes to your database. The data lake provides a way to backload your data if necessary due to changes in schemas or data ingestion issues. The portion of the data pipeline allocated to this process is shown below.<\/p>\n

<\/p>\n

Batching events to AVRO format and saving to Google\u00a0Storage<\/em><\/p>\n

For AVRO, we can\u2019t use a direct streaming approach. We need to group events into batches before we can save to flat files. The way this can be accomplished in DataFlow is by applying a windowing function that groups events into fixed batches. The code below applies transformations that convert the PubSub messages into String objects, group the messages into 5 minute intervals, and output the resulting batches to AVRO files on Google Storage.
\u00a0<\/p>\n

\/\/ AVRO output portion of the pipeline\u00a0\u00a0\u00a0
events.apply(“To String”, ParDo.of(new DoFn() {
\u00a0 @ProcessElement\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 public void processElement(ProcessContext c) throws Exception {
\u00a0\u00a0\u00a0 String message = new String(c.element().getPayload());
\u00a0\u00a0\u00a0 c.output(message);\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 }\u00a0\u00a0\u00a0\u00a0\u00a0
}))<\/p>\n

\/\/ Batch events into 5 minute windows\u00a0\u00a0\u00a0\u00a0\u00a0
.apply(“Batch Events”, Window.<String>into(\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0 FixedWindows.of(Duration.standardMinutes(5)))\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 .triggering(AfterWatermark.pastEndOfWindow())\u00a0\u00a0\u00a0\u00a0
\u00a0 .discardingFiredPanes()\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 .withAllowedLateness(Duration.standardMinutes(5)))<\/p>\n

\u200b\/\/ Save the events in ARVO format\u00a0\u00a0\u00a0\u00a0\u00a0
.apply(“To AVRO”, AvroIO.write(String.class)
\u00a0 .to(“gs:\/\/your_gs_bucket\/avro\/raw-events.avro”)
\u00a0 .withWindowedWrites()
\u00a0 .withNumShards(8)
\u00a0 .withSuffix(“.avro”));<\/p>\n<\/div>\n

To summarize, the above code batches game events into 5 minute windows and then exports the events to AVRO files on Google Storage.\u00a0<\/p>\n

The result of this portion of the data pipeline is a collection of AVRO files on google storage that can be used to build a data lake. A new AVRO output is generated every 5 minutes, and downstream ETLs can parse the raw events into processed event-specific table schemas. The image below shows a sample output of AVRO files.<\/p>\n

<\/p>\n

AVRO files saved to Google\u00a0Storage<\/em><\/p>\n

In addition to creating a data lake, we want the events to be immediately accessible in a query environment. DataFlow provides a BigQuery connector which serves this functionality, and data streamed to this endpoint is available for analysis after a short duration. This portion of the data pipeline is shown in the figure below.\u00a0<\/p>\n

<\/p>\n

Streaming events from DataFlow to\u00a0BigQuery<\/em><\/p>\n

The data pipeline converts the PubSub messages into TableRow objects, which can be directly inserted into BigQuery. The code below consists of two apply methods: a data transformation and a IO writer. The transform step reads the message payloads from PubSub, parses the message as a JSON object, extracts the\u00a0eventType<\/em>\u00a0and\u00a0eventVersion<\/em>\u00a0attributes, and creates a TableRow object with these attributes in addition to a timestamp and the message payload. The second apply method tells the pipeline to write the records to BigQuery and to append the events to an existing table.\u00a0
\u00a0<\/p>\n

\/\/ parse the PubSub events and create rows to insert into BigQuery\u00a0\u00a0\u00a0
events.apply(“To Table Rows”, new
\u00a0 PTransform, PCollection>() {
\u00a0\u00a0\u00a0 public PCollection expand(
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 PCollection input) {\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 return input.apply(“To Predictions”, ParDo.of(new\u00a0
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 DoFn() {\u00a0\u00a0<\/p>\n

\u00a0\u00a0\u00a0 @ProcessElement\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0 public void processElement(ProcessContext c) throws Exception {
\u00a0\u00a0\u00a0\u00a0\u00a0 String message = new String(c.element().getPayload());<\/p>\n

\u00a0\u00a0\u00a0\u00a0\u00a0 \/\/ parse the json message for attributes
\u00a0\u00a0\u00a0\u00a0\u00a0 JsonObject jsonObject =
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 new JsonParser().parse(message).getAsJsonObject();
\u00a0\u00a0\u00a0\u00a0\u00a0 String eventType = jsonObject.get(“eventType”).getAsString();
\u00a0\u00a0\u00a0\u00a0\u00a0 String eventVersion = jsonObject.
\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 get(“eventVersion”).getAsString();\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0\u00a0 String serverTime = dateFormat.format(new Date());<\/p>\n

\u00a0\u00a0\u00a0\u00a0 \/\/ create and output the table row\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0 TableRow record = new TableRow();\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0 record.set(“eventType”, eventType);\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0 record.set(“eventVersion”, eventVersion);\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0 record.set(“serverTime”, serverTime);
\u00a0\u00a0\u00a0\u00a0 record.set(“message”, message);\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0\u00a0\u00a0\u00a0 c.output(record);\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 }}));\u00a0\u00a0\u00a0\u00a0\u00a0
}})<\/p>\n

\/\/stream the events to Big Query\u00a0\u00a0\u00a0
.apply(“To BigQuery”,BigQueryIO.writeTableRows()\u00a0\u00a0
\u00a0 .to(table)\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0
\u00a0 .withSchema(schema)
\u00a0 .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
\u00a0 .withWriteDisposition(WriteDisposition.WRITE_APPEND));<\/p>\n<\/div>\n

To summarize the above code, each game event\u00a0that is consumed from PubSub is converted into a TableRow object with a timestamp and then streamed to BigQuery for storage.\u00a0<\/p>\n

The result of this portion of the data pipeline is that game events will be streamed to BigQuery and will be available for analysis in the output table specified by the DataFlow task. In order to effectively use these events for queries, you\u2019ll need to build additional ETLs for creating processed event tables with schematized records, but you now have a data collection mechanism in place for storing events.\u00a0<\/p>\n

<\/p>\n

Game event records queried from the raw-events table in\u00a0BigQuery<\/em><\/p>\n

Deploying and Auto\u00a0Scaling<\/strong><\/h2>\n

With DataFlow you can test the data pipeline locally or deploy to the cloud. If you run the code samples without specifying additional attributes, then the data pipeline will execute on your local machine. In order to deploy to the cloud and take advantage of the auto scaling capabilities of this data pipeline, you need to specify a new runner class as part of your runtime arguments. In order to run the data pipeline, I used the following runtime arguments:<\/p>\n

–runner=org.apache.beam.runners.dataflow.DataflowRunner
–jobName=game-analytics
–project=your_project_id
–tempLocation=gs:\/\/temp-bucket<\/p>\n

Once the job is deployed, you should see a message that the job has been submitted. You can then click on the\u00a0DataFlow console<\/a>\u00a0to see the task:\u00a0<\/p>\n

<\/p>\n

The steaming data pipeline running on Google\u00a0Cloud<\/em><\/p>\n

The runtime configuration specified above will not default to an auto scaling configuration. In order to deploy a job that scales up based on demand, you\u2019ll need to specify additional attributes, such as:
\u00a0<\/p>\n

–autoscalingAlgorithm=THROUGHPUT_BASED
–maxNumWorkers=30<\/p>\n

Additional details on setting up a DataFlow task to scale to heavy workload conditions are available in\u00a0this Google article<\/a>\u00a0and\u00a0this post<\/a>\u00a0from Spotify. The image below shows how DataFlow can scale up to meet demand as necessary.<\/p>\n

<\/p>\n

An example of Dataflow auto scaling<\/em><\/p>\n

Conclusion\u00a0<\/strong><\/h2>\n

There is now a variety of tools available that make it possible to set up an analytics pipeline for a game or web application with minimal effort. Using managed resources enables small teams to take advantage of serverless and autoscaling infrastructure to scale up to massive event volumes with minimal infrastructure management. Rather than using a data vendor\u2019s off-the-shelf solution for collecting data, you can record all relevant data for your app.\u00a0<\/p>\n

The goal of this post was to show how a data lake and query environment can be set up using the GCP stack. While the approach presented here isn\u2019t directly portable to other clouds, the Apache Beam library used to implement the core functionality of this data pipeline is portable and similar tools can be leveraged to build scalable data pipelines on other cloud providers.<\/p>\n

This architecture is a minimal implementation of an event collection system that is useful for analytics and data science teams. In order to meet the demands of most analytics teams, the raw events will need to be transformed into processed and cooked events in order to meet business needs. This discussion is outside the scope of this post, but the analytics foundation should now be in place for building out a highly effective data platform.\u00a0<\/p>\n

The full source code for this sample pipeline is available on Github<\/a>.\u00a0Ben Weber<\/a>\u00a0is the lead data scientist at\u00a0Windfall Data<\/a>, where our mission is to build the most accurate and comprehensive model of net worth.<\/p>\n","protected":false},"excerpt":{"rendered":"

The following blog post, unless otherwise noted, was written by a member of Gamasutra\u2019s community.The thoughts and opinions expressed are those of the writer and not Gamasutra or its parent company. Gathering usage\u00a0data such as player progress in games is invaluable for teams. Typically, entire teams have been dedicated to building and maintaining data pipelines […]<\/p>\n","protected":false},"author":2,"featured_media":16494,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[20],"tags":[],"class_list":["post-16493","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-news"],"_links":{"self":[{"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/posts\/16493","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/comments?post=16493"}],"version-history":[{"count":0,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/posts\/16493\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/media\/16494"}],"wp:attachment":[{"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/media?parent=16493"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/categories?post=16493"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/sickgaming.net\/blog\/wp-json\/wp\/v2\/tags?post=16493"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}