GSP603

Overview

Today’s financial world is complex, and the old technology used for constructing financial data pipelines isn’t keeping up. With multiple financial exchanges operating around the world and global user demand, these data pipelines have to be fast, reliable and scalable.

Currently, using an econometric approach—applying models to financial data to forecast future trends—doesn’t work for real-time financial predictions. And data that’s old, inaccurate or from a single source doesn’t translate into dependable data for financial institutions to use.

But building pipelines with Google Cloud can solve some of these key challenges. In this lab, you’ll describe how to build a pipeline to predict financial trends in microseconds, set up and configure a pipeline for ingesting real-time, time-series data from various financial exchanges, and how to design a suitable data model, which facilitates querying and graphing at scale.

You’ll find a tutorial below on setting up and deploying the proposed architecture using Google Cloud, particularly these products:

• Cloud Dataflow for scalable data ingestion system that can handle late data
• Cloud Bigtable, our scalable, low-latency time series database that’s reached 40 million transactions per second on 3500 nodes. Bonus: Scalable ML pipeline using Tensorflow eXtended, while not part of this tutorial, is a logical next step.

The tutorial will explain how to establish a connection to multiple exchanges, subscribe to their trade feed, and extract and transform these trades into a flexible format to be stored in Cloud Bigtable and be available to be graphed and analyzed.

This will also set the foundation for ML online learning predictions at scale. You’ll see how to graph the trades, volume, and time delta from trade execution until it reaches your system (an indicator of how close to real time you can get the data). You can find more details on GitHub too.

Requirements / Solutions

Architectural overview

The typical requirement for trading systems is low latency data ingestion, and for this lab is extended with near real-time data storage and querying at scale. You will learn the following from this lab:

1. Ingest real-time trading data with low latency from globally scattered datasources / exchanges. Possibility to adopt data ingest worker pipeline location. Easily add additional trading pairs / exchanges. Solution: Dataflow + Xchange Reactive Websockets Framework
2. Demonstrate an unbounded streaming source code that is runnable with multiple runners. Solution: Apache BEAM
3. Strong consistency + linear scalability + super low latency for querying the trading data. Solution: Bigtable
4. Querying and visualization — Execute time series queries on Bigtable visualize it in on the webpage. Solution: Python Flask + Vis.js + Google Bigtable Python Client

Architecture / How it works

The source code is written in Java 8, Python 3.7, JavaScript; and Maven, PIP for dependency/build management.

The code can be divided into five main framework units:

1. Data ingestion — The XChange Stream framework (Github link) Java library provides a simple and consistent streaming API for interacting with Bitcoin and other cryptocurrency exchanges via WebSocket protocol. XChange library is providing new interfaces for streaming API. Users can subscribe for live updates via reactive streams of RxJava library. We use this JAVA 8 framework to connect and configure some exchanges (BitFinex, Poloniex, BitStamp, OkCoin, Gemini, HitBTC, Binance...). Link to the exchange / trading pair configuration code
2. Parallel processing — Apache Beam (Github link) Apache Beam is an open source unified programming model to define and execute data processing pipelines, including ETL, batch, and stream (continuous) processing. Supported runners: Apache Apex, Apache Flink, Apache Gearpump, Apache Samza, Apache Spark, and Google Cloud Dataflow. You will learn how to create an unbounded streaming source/reader and manage basic watermarking, checkpointing, and record ID for data ingestion. Link to the bridge between BEAM and XChange Stream framework
3. Bigtable sink — Cloud Bigtable with Beam using the HBase API. (Github link) Connector and writer to Bigtable. You will see how to create a row key and create a Bigtable mutation function prior to writing to Bigtable. Link to the Bigtable key creation / mutation function
4. Realtime API endpoint — Flask web server at port:5000 + Bigtable client (GitHub link) will be used to query the Bigtable and serve as API endpoint. Link to the Bigtable query builder + results retrieval and sampling
5. JavaScript Visualization — Vis.JS Flask template that will query the real-time API endpoint every 500ms. Link to the HTML template file

Flask web server will be run in the Google Cloud VM instance

Pipeline definition

For every exchange + trading pair, a different pipeline instance is created. The pipeline consists of 3 steps:

1. UnboundedStreamingSource that contains ‘Unbounded Streaming Source Reader' (bitStamp2)
2. Pre-writing mutation and key definition
   (ETH-USD Mut2)
3. Bigtable write step (ETH-USD2)

Bigtable row key design decisions

The DTO for this lab looks like this:

The row key structure is formulated in the following way:

TradingCurrency#Exchange#SystemTimestampEpoch#SystemNanosTime

E.g: a row key might look like BTC/USD#Bitfinex#1546547940918#63187358085

• BTC/USD — Trading Pair

• Bitfinex — Exchange

• 1546547940918 — Epoch timestamp ( more info)

• 63187358085 — System Nano time ( more info)

Why is nanotime added at the key end?

Nanotime is used to avoid multiple versions per row for different trades. Two DoFn mutations might execute in the same Epoch ms time if there is a streaming sequence of TradeLoad DTOs. NanoTime at the end will split Millisecond to an additional one million.

In your own environment, if this is not enough, you can hash the volume / price ratio and attach the hash at the end of the row key.

Row cells will contain an exact schema replica of the exchange TradeLoad DTO (see earlier in the table above). This choice will help you go from a specific (trading pair) — (exchange) to less specific (timestamp — nanotime) and avoid hotspots when you query the data.

Setup and requirements

Before you click the Start Lab button

Read these instructions. Labs are timed and you cannot pause them. The timer, which starts when you click Start Lab, shows how long Google Cloud resources will be made available to you.

This hands-on lab lets you do the lab activities yourself in a real cloud environment, not in a simulation or demo environment. It does so by giving you new, temporary credentials that you use to sign in and access Google Cloud for the duration of the lab.

To complete this lab, you need:

• Access to a standard internet browser (Chrome browser recommended).

Note: Use an Incognito or private browser window to run this lab. This prevents any conflicts between your personal account and the Student account, which may cause extra charges incurred to your personal account.

• Time to complete the lab---remember, once you start, you cannot pause a lab.

Note: If you already have your own personal Google Cloud account or project, do not use it for this lab to avoid extra charges to your account.

How to start your lab and sign in to the Google Cloud console

1. Click the Start Lab button. If you need to pay for the lab, a pop-up opens for you to select your payment method. On the left is the Lab Details panel with the following:

   • The Open Google Cloud console button
   • Time remaining
   • The temporary credentials that you must use for this lab
   • Other information, if needed, to step through this lab

2. Click Open Google Cloud console (or right-click and select Open Link in Incognito Window if you are running the Chrome browser).

   The lab spins up resources, and then opens another tab that shows the Sign in page.

   Tip: Arrange the tabs in separate windows, side-by-side.

   Note: If you see the Choose an account dialog, click Use Another Account.

3. If necessary, copy the Username below and paste it into the Sign in dialog.

   {{{user_0.username | "Username"}}}

   You can also find the Username in the Lab Details panel.

4. Click Next.

5. Copy the Password below and paste it into the Welcome dialog.

   {{{user_0.password | "Password"}}}

   You can also find the Password in the Lab Details panel.

6. Click Next.

   Important: You must use the credentials the lab provides you. Do not use your Google Cloud account credentials. Note: Using your own Google Cloud account for this lab may incur extra charges.

7. Click through the subsequent pages:

   • Accept the terms and conditions.
   • Do not add recovery options or two-factor authentication (because this is a temporary account).
   • Do not sign up for free trials.

After a few moments, the Google Cloud console opens in this tab.

Note: To view a menu with a list of Google Cloud products and services, click the Navigation menu at the top-left.

Activate Cloud Shell

Cloud Shell is a virtual machine that is loaded with development tools. It offers a persistent 5GB home directory and runs on the Google Cloud. Cloud Shell provides command-line access to your Google Cloud resources.

1. Click Activate Cloud Shell at the top of the Google Cloud console.

When you are connected, you are already authenticated, and the project is set to your Project_ID, . The output contains a line that declares the Project_ID for this session:

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud is the command-line tool for Google Cloud. It comes pre-installed on Cloud Shell and supports tab-completion.

2. (Optional) You can list the active account name with this command:

gcloud auth list

3. Click Authorize.

Output:

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

4. (Optional) You can list the project ID with this command:

gcloud config list project

Output:

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} Note: For full documentation of gcloud, in Google Cloud, refer to the gcloud CLI overview guide.

Task 1. Create your lab resources

You need a virtual machine to perform the creation of the pipeline and use as your website.

• In Cloud Shell, run the following command:

gcloud beta compute instances create crypto-driver \ --zone=us-east1-c \ --machine-type=n1-standard-1 \ --subnet=default \ --network-tier=PREMIUM \ --maintenance-policy=MIGRATE \ --service-account=$(gcloud iam service-accounts list --format='value(email)' --filter="compute") \ --scopes=https://www.googleapis.com/auth/cloud-platform \ --image-family=debian-11 \ --image-project=debian-cloud \ --boot-disk-size=20GB \ --boot-disk-type=pd-standard \ --boot-disk-device-name=crypto-driver

The compute engine service account is used with the cloud API scope. This provides the necessary permissions to create the necessary resources for your environment.

Wait for the instance to start.

Click Check my progress to verify the objective.

Create a virtual machine to perform the creation of the pipeline and use as your website.

Connect to the instance via SSH

1. In the Cloud Platform Console, on the Navigation menu, click Compute Engine > VM Instances.

2. For the instance called crypto-driver , click SSH.

A window will open and you will be automatically logged into the instance. You will run all commands for the remainder of this lab in the SSH window.

3. Run the following commands to uninstall the existing google-cloud-cli and install all the necessary tools (such as java, git, maven, pip, python and cloud bigtable command line tool cbt):

sudo apt-get remove google-cloud-cli -y sudo -s apt-get update -y sudo apt install python3-pip -y sudo pip3 install -U virtualenv virtualenv -p python3 venv source venv/bin/activate sudo apt -y --allow-downgrades install openjdk-11-jdk git maven google-cloud-sdk=349.0.0-0 google-cloud-sdk-cbt=349.0.0-0

This will also install virtualenv for the Python environment. All the Python related code will be executed in the Virtualenv.

4. Now create the Bigtable resource. The first gcloud command will enable the required Bigtable and Dataflow API in the project. The next command will create a Bigtable Cluster called “cryptorealtime-c1” with one instance called “cryptorealtime”. The instance type is Development, and therefore it will be a one node instance. And finally, using the cbt command you are creating a table called “cryptorealtime” with one column family called “market” in the Bigtable instance.

export PROJECT=$(gcloud info --format='value(config.project)') export ZONE=$(curl "http://metadata.google.internal/computeMetadata/v1/instance/zone" -H "Metadata-Flavor: Google"|cut -d/ -f4) gcloud services enable bigtable.googleapis.com \ bigtableadmin.googleapis.com \ dataflow.googleapis.com \ --project=${PROJECT} gcloud bigtable instances create cryptorealtime \ --cluster=cryptorealtime-c1 \ --cluster-zone=${ZONE} \ --display-name=cryptorealtime \ --cluster-storage-type=HDD \ --instance-type=DEVELOPMENT cbt -instance=cryptorealtime createtable cryptorealtime families=market

Click Check my progress to verify the objective.

Create the Bigtable instance

For this lab, one column family called market is used to simplify the schema design. For more on that you can read this link.

5. Run the following command to create a bucket:

gsutil mb -p ${PROJECT} gs://realtimecrypto-${PROJECT}

This bucket will be used by the Dataflow job as a staging area for the Jar files.

Click Check my progress to verify the objective.

Create a Cloud Storage bucket

6. Now clone the application source code repository:

git clone https://github.com/GoogleCloudPlatform/professional-services

7. Run the following to build the software:

cd professional-services/examples/cryptorealtime mvn clean install

Output:

[INFO] ---------------------------------------------------------------- [INFO] BUILD SUCCESS [INFO] ---------------------------------------------------------------- [INFO] Total time: 5.862 s [INFO] Finished at: 2018-12-13T00:29:30+00:00 [INFO] Final Memory: 49M/253M [INFO]-----------------------------------------------------------------

The code for this lab is written in Java, which needs to be compiled and packaged as a Jar file using Maven build tool. The build will take a couple of minutes, please wait until you see the BUILD SUCCESS message.

8. Once the build is finished, start the pipeline:

./run.sh ${PROJECT} \ cryptorealtime gs://realtimecrypto-${PROJECT}/temp \ cryptorealtime market

If you check the logs, don't worry about the errors you see. It is safe to ignore illegal thread pool exceptions.

Wait for the command to complete.

run.sh script is a wrapper to submit the crypto tracking Dataflow job on Google Cloud. It takes the following arguments:

• Project_Name,
• Bigtable Instance name,
• Cloud Storage bucket to use as staging area,
• Bigtable Table name and
• Column family name to write the output of the pipeline.

When the pipeline starts, it will create two worker nodes. At this point, you will see two additional VMs in your project.

9. Wait a couple of minutes, then run the following to observe the incoming trades by peeking into Bigtable. You can do this by using Cloud Bigtable CLI tool (called cbt). If the pipeline is successfully executing, you should see new data appearing in the cryptorealtime table.

cbt -instance=cryptorealtime read cryptorealtime

You will see something like the following if the data is flowing. If you see nothing, please wait for a minute and try again.

Click Check my progress to verify the objective.

Run the Dataflow pipeline

Task 2. Examine the Dataflow pipeline

1. In the Cloud Platform Console, on the Navigation menu, click Dataflow.

2. Click the name of the existing pipeline.

You should see the status as Running for the listed jobs.

Note: Ignore the javalang error type and move ahead in the lab.

Task 3. Visualizing the data

You will configure the environment and run the python Flask frontend server visualization in these next steps.

1. Go back to the SSH session and run the following command to open firewall port 5000 for visualization:

gcloud compute --project=${PROJECT} firewall-rules create crypto-dashboard \ --direction=INGRESS \ --priority=1000 \ --network=default \ --action=ALLOW \ --rules=tcp:5000 \ --source-ranges=0.0.0.0/0 \ --target-tags=crypto-console \ --description="Open port 5000 for crypto visualization tutorial"

2. Now link the VM with the firewall rule:

gcloud compute instances add-tags crypto-driver --tags="crypto-console" --zone=${ZONE}

Click Check my progress to verify the objective.

Open firewall port 5000 for visualization

3. Next, navigate to the frontend directory:

cd frontend/ pip install -r requirements.txt pip uninstall Flask Jinja2

If prompted Proceed (Y/n)?, press Y.

pip install Flask Jinja2 python app.py ${PROJECT} cryptorealtime cryptorealtime market

app.py is a Python application to visualize the Crypto currency data stored in the Bigtable table “cryptorealtime”.

4. Open another SSH session and run the following command to find your external IP address for the crypto-driver instance:

gcloud compute instances list --format='value(EXTERNAL_IP)' --filter="name:crypto-driver"

5. Copy the EXTERNAL IP address displayed to use for the next command.

6. Open a new tab in your web browser and use the following URL to see the visualization, replacing <external-ip> with the IP address returned from the previous command:

http://<external-ip>:5000/stream

You should see the visualization of aggregated ASK/BID pair on several exchanges (without predictor part).

You have created a real-time "periscope" multi-exchange observer.

Task 4. Clean up

When you end this lab, the lab platform deletes the resources you used. In your own environment, it is useful to know how to save costs by cleaning up your unused, or no longer required, resources.

1. You can stop the pipeline in either the Console or in the SSH session. It takes a few minutes with either method.

• Console: Return to Dataflow page, click on the name of your job and click Stop. Select Cancel, then Stop Job.

• Inside the second SSH session run:

  gcloud dataflow jobs cancel \ $(gcloud dataflow jobs list \ --format='value(id)' \ --filter="name:runthepipeline*" \ --region="us-central1")

  When the pipeline stops, return to the command line in the SSH session by pressing CTRL+C.

2. Inside the SSH session run the following commands to empty and delete the bucket. Replace with your your Project ID:

export PROJECT=<Your Project ID> gsutil -m rm -r gs://realtimecrypto-${PROJECT}/* gsutil rb gs://realtimecrypto-${PROJECT}

3. Inside the SSH session run the following command to delete the Bigtable instance:

gcloud bigtable instances delete cryptorealtime

4. If prompted, Do you want to continue (Y/n), press Y.

5. Close the SSH console.

6. In the Cloud Platform Console, on the Navigation menu, click Compute Engine > VM Instances.

7. Check the box next to the crypto-driver instance then click Delete, then Delete again to confirm your action.

Congratulations!

You have learned how to do the following:

• Set up and configure a pipeline for ingesting real-time time-series data from various crypto exchanges.

• Design a suitable data model, which facilitates querying and graphing at scale.

• Set up and deploy the proposed architecture using Google Cloud.

You established a connection to multiple exchanges, subscribed to their trade feed, then extracted and transformed these trades into a flexible format to be stored in Bigtable to be graphed and analyzed.

Next steps/ Learn more

This lab is based on this Medium article by Ivo Galic.

Google Cloud training and certification

...helps you make the most of Google Cloud technologies. Our classes include technical skills and best practices to help you get up to speed quickly and continue your learning journey. We offer fundamental to advanced level training, with on-demand, live, and virtual options to suit your busy schedule. Certifications help you validate and prove your skill and expertise in Google Cloud technologies.

Manual Last Tested January 22, 2024

Manual Last Updated January 22, 2024

Copyright 2024 Google LLC All rights reserved. Google and the Google logo are trademarks of Google LLC. All other company and product names may be trademarks of the respective companies with which they are associated.