Data Science on the Google Cloud Platform

SECOND EDITION

Implementing End-to-End Real-Time Data Pipelines: From Ingest to Machine Learning

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

Valliappa Lakshmanan

Data Science on the Google Cloud Platform

by Valliappa Lakshmanan

Copyright © 2022 Google LLC. All rights reserved.

Printed in the United States of America.

Published by O’Reilly Media, Inc., 1005 Gravenstein Highway North, Sebastopol, CA 95472.

O’Reilly books may be purchased for educational, business, or sales promotional use. Online editions are also available for most titles (http://oreilly.com). For more information, contact our corporate/institutional sales department: 800-998-9938 or corporate@oreilly.com.

- Editors: Michele Cronin and Jessica Haberman

- Production Editor: Katherine Tozer

- Interior Designer: David Futato

- Cover Designer: Karen Montgomery

- Illustrator: Kate Dullea

- January 2023: Second Edition

Revision History for the Early Release

- 2021-10-26: First Release

See http://oreilly.com/catalog/errata.csp?isbn=9781098118952 for release details.

The O’Reilly logo is a registered trademark of O’Reilly Media, Inc. Data Science on the Google Cloud Platform, the cover image, and related trade dress are trademarks of O’Reilly Media, Inc.

The views expressed in this work are those of the author(s) and do not represent the publisher’s views. While the publisher and the author(s) have used good faith efforts to ensure that the information and instructions contained in this work are accurate, the publisher and the author(s) disclaim all responsibility for errors or omissions, including without limitation responsibility for damages resulting from the use of or reliance on this work. Use of the information and instructions contained in this work is at your own risk. If any code samples or other technology this work contains or describes is subject to open source licenses or the intellectual property rights of others, it is your responsibility to ensure that your use thereof complies with such licenses and/or rights.

978-1-098-11895-2

Chapter 1. Making Better Decisions Based on Data

A NOTE FOR EARLY RELEASE READERS

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 1st chapter of the final book. Please note that the GitHub repo will be made active later on.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at mcronin@oreilly.com.

The primary purpose of data analysis is to make better decisions. There is rarely any need for us to spend time analyzing data if we aren’t under pressure to make a decision based on the results of that analysis. When you are purchasing a car, you might ask the seller what year the car was manufactured and the odometer reading. Knowing the age of the car allows you to estimate the potential value of the car. Dividing the odometer reading by the age of the car allows you to discern how hard the car has been driven, and whether it is likely to last the five years you plan to keep it. Had you not cared about purchasing the car, there would have been no need for you to do this data analysis.

In fact, we can go further—the purpose of collecting data is, in many cases, only so that you can later perform data analysis and make decisions based on that analysis (See Figure 1-1). When you asked the seller the age of the car and its mileage, you were collecting data to carry out your data analysis. But it goes beyond your data collection. The car has an odometer in the first place because many people, not just potential buyers, will need to make decisions based on the mileage of the car. The odometer reading needs to support many decisions—should the manufacturer pay for a failed transmission? Is it time for an oil change? The analysis for each of these decisions is different, but they all rely on the fact that the mileage data has been collected.

Figure 1-1. The purpose of collecting data is to make decisions with it.

If you are in the business of making a lot of decisions using car mileage, it makes sense to store the data that you have collected so that future decisions are easier to make. Collecting data takes time and effort, whereas storing it is relatively inexpensive. Of course, you have to plan on storing the data in a way that you will know what it means when you need it later. This is called capturing the semantics of the data and is an important aspect of data governance, to ensure that data is useful for decision making.

Collecting data in a form that enables decisions to be made places requirements on the collecting infrastructure and the security of such infrastructure. How does the insurance company that receives an accident claim and needs to pay its customer the car’s value know that the odometer reading is accurate? How are odometers calibrated? What kinds of safeguards are in place to ensure that the odometer has not been tampered with? What happens if the tampering is inadvertent, such as installing tires whose size is different from what was used to calibrate the odometer? The auditability of data is important whenever there are multiple parties involved, and ownership and use of the data are separate. When data is unverifiable, markets fail, optimal decisions cannot be made, and the parties involved need to resort to signaling and screening.1

Not all data is as expensive to collect and secure as the odometer reading of a car.2 The cost of sensors has dropped dramatically in recent decades, and many of our daily processes produce so much data that we find ourselves in possession of data that we had no intention of explicitly collecting. As the hardware to collect, ingest, and store the data has become cheaper, we often default to retaining the data indefinitely, keeping it around for no discernable reason. However, if we’re to perform analysis on all of this data that we somehow managed to collect and store, we better have a purpose for it. Labor remains expensive.

Another reason to be purposeful about the data we collect and store is that a lot of it is about people. Knowing the mileage of the car that someone drives gives us a lot of information about them and this is information that they may not want to share other than for the specific purpose of estimating the market price of their car. Privacy and confidentiality need to be considered even before any data is collected, so that appropriate decisions can be made about what data to collect, how to control access to it, and how long to retain it.

Data analysis is usually triggered because a decision needs to be made. To move into a market or not? To pay a commission or not? How high to bid up the price? How many bags to purchase? Whether to buy now or wait a week? The decisions keep multiplying, and because data is so ubiquitous now, we no longer need to make those decisions based on heuristic rules of thumb. We can now make those decisions in a data-driven manner.

Of course, we don’t need to create the systems and tools to make every data-driven decision ourselves. The use case of estimating the value of a car that has been driven a certain distance is common enough that there are several companies that provide this as a service—they will verify that an odometer is accurate, confirm that the car hasn’t been in an accident, and compare the asking price against the typical selling price of cars in your market. The real value, therefore, comes not in making a data-driven decision once, but in being able to do it systematically and provide it as a service. This also allows companies to specialize, and continuously improve the accuracy of the decisions that can be made.

Many Similar Decisions

Because of the low costs associated with sensors and storage, there are many industries and use cases that have the potential to support data-driven decision making. If you are working in such an industry, or you want to start a company that will address such a use case, the possibilities for supporting data-driven decision making have just become wider. In some cases, you will need to collect the data. In others, you will have access to data that was already collected, and, in many cases, you will need to supplement the data you have with other datasets that you will need to hunt down. In all these cases, being able to carry out data analysis to support decision making systematically on behalf of users is a good skill to possess.

In this book, I will take a decision that needs to be made and apply different statistical and machine learning methods to gain insight into making that decision. However, we don’t want to make that decision just once, even though we might occasionally pose it that way. Instead, we will look at how to make the decision in a systematic manner so that we use the same algorithm to make the decision many, many, many times. Our ultimate goal will be to provide this decision-making capability as a service to our customers—they will tell us the things they reasonably can be expected to know, and we will either know or infer the rest (because we have been systematically collecting data). Based on this data, we will suggest the optimal decision.

Whether or not a decision is a one-off is the primary difference between data analytics and data science. Data analytics is about manually analyzing data to make a single decision or answer a single question. Data science is about developing a technique (called a model or algorithm) so that similar decisions can be made in a systematic way. Often, data science is about automating and optimizing the decision making process that was first determined through data analysis.3

When we are collecting the data, we will need to look at how to make the data secure. This will include how to ensure not only that the data has not been tampered with, but also that users’ private information is not compromised—for example, if we are systematically collecting odometer mileage and know the precise mileage of the car at any point in time, this knowledge becomes extremely sensitive information. Given enough other information about the customer (such as the home address and traffic patterns in the city in which the customer lives), the mileage is enough to be able to infer that person’s location at all times.4 So, the privacy implications of hosting something as seemingly innocuous as the mileage of a car can become enormous. Security implies that we need to control access to the data, and we need to maintain immutable audit logs on who has viewed or changed the data.

It is not enough to simply collect the data or use it as is. We must understand the data. Just as we needed to know the kinds of problems associated with odometer tampering to understand the factors that go into estimating a vehicle’s value based on mileage, our analysis methods will need to consider how the data was collected in real time, and the kinds of errors that could be associated with that data. Intimate knowledge of the data and its quirks is invaluable when it comes to doing data science—often the difference between a data-science startup idea that works and one that doesn’t is whether the appropriate nuances have all been thoroughly evaluated and taken into account.

When it comes to providing the decision-support capability as a service, it is not enough to simply have a way to do it in some offline system somewhere. Enabling it as a service implies a whole host of other concerns. The first set of concerns is about the quality of the decision itself—how accurate is it typically? What are the typical sources of errors? In what situations should this system not be used? The next set of concerns, however, is about the quality of service. How reliable is it? How many queries per second can it support? What is the latency between some piece of data being available, and it being incorporated into the model that is used to provide systematic decision making? In short, we will use this single use case as a way to explore many different facets of practical data science.

The Role of Data Scientists

“Wait a second,” I imagine you saying, “I never signed up for queries-per-second of a web service. We have people who do that kind of stuff. My job is to write SQL queries and create reports. I don’t recognize this thing you are talking about. It’s not what I do at all.” Or perhaps the first part of the discussion was what puzzled you. “Decision making? That’s for the business people. Me? What I do is to design data processing systems. I can provision infrastructure, tell you what our systems are doing right now, and keep it all secure. Data science sure sounds fancy, but I do engineering. When you said Data Science on the Google Cloud Platform, I was thinking that you were going to talk about how to keep the systems humming and how to offload bursts of activity to the cloud.” A third set of people are wondering, “How is any of this data science? Where’s the discussion of different types of models and of how to make statistical inferences and evaluate them? Where’s the math? Why are you talking to data analysts and engineers? Talk to me, I’ve got a PhD.” These are fair points—I seem to be mixing up the jobs done by different sets of people in your organization.

In other words, you might agree with the following:

- Data analysis is there to support decision making

- Decision making in a data-driven manner can be superior to heuristics

- The accuracy of the decision models depends on your choice of the right statistical or machine learning approach

- Nuances in the data can completely invalidate your modeling, so understanding the data and its quirks is crucial

- There are large market opportunities in supporting decision making systematically and providing it as a service

- Such services require ongoing data collection and model updates

- Ongoing data collection implies robust security and auditing

- Customers of the service require reliability, accuracy, and latency assurances

What you might not agree with is whether these aspects are all things that you, personally and professionally, need to be concerned about. Instead, you might think of yourself as a data analyst, a data engineer, or a data scientist and not care about how the other roles do whatever it is that they do.

There are three answers to this objection:

1. In a scrappy, innovative company or startup, you will find yourself playing all these roles, so learn the full lifecycle.

2. The public cloud makes it relatively easy to learn all the roles, so why not be a full stack data scientist?

3. Even if you work in a large company where these tasks are carried out by different roles, it is helpful to understand the end-to-end process and concerns at each stage. This will help you collaborate with other teams better.

Let’s take these answers one by one.

Scrappy Environment

At Google, we look at the role of a data engineer quite expansively. Just as we refer to all our technical staff as engineers, we look at data engineers as an inclusive term for anyone who can “shape business outcomes by performing data analysis”. To perform data analysis, you begin by preparing the data so that you can analyze it at scale. It is not enough to simply count and sum and graph the results using SQL queries and charting software—you must understand the nuances of the data and the statistical framework within which you are interpreting the results. This ability to prepare the data and carry out statistically valid data analysis to solve specific business problems is of paramount importance—the queries, the reports, the graphs are not the end goal. A verifiably accurate decision is.

Of course, it is not enough to do one-off data analysis. That data analysis needs to scale. In other words, the accurate decision-making process must be repeatable and be capable of being carried out by many users, not just you. The way to scale up one-off data analysis is to make it automated. After a data engineer has devised the algorithm, they should be able to make it systematic and repeatable. Just as it is a lot easier when the folks in charge of systems reliability can make code changes themselves,5 it is considerably easier when people who understand statistics and machine learning can code those models themselves. A data engineer, Google believes, should be able to go from building statistical and machine learning models to automating them. They can do this only if they are capable of designing, building, and troubleshooting data processing systems that are secure, reliable, fault-tolerant, scalable, and efficient.

This desire to have engineers who know data science and data scientists who can code is not Google’s alone—it’s common at many startups and small companies. When a scrappy company advertises for data engineers or for data scientists, what they are looking for is a person who can do all the three tasks – data preparation, data analysis, and automation – that are needed to make repeatable, scalable decisions on the basis of data.

How realistic is it for companies to expect a Renaissance man, a virtuoso in different fields? Can they reasonably expect to hire data engineers who can do data science? How likely is it that they will find someone who can design a database schema, write SQL queries, train machine learning models, code up a data processing pipeline, and figure out how to scale it all up? Surprisingly, this is a very reasonable expectation, because the amount of knowledge you need in order to do these jobs has become a lot less than what you needed a few years ago.

Full Stack Cloud Data Scientists

Because of the ongoing movement to the cloud, data scientists can do the job that used to be done by several people with different sets of skills. With the advent of autoscaling, serverless, managed infrastructure that is easy to program, there are more and more people who can build scalable systems. Therefore, it is now reasonable to expect to be able to hire data scientists who are capable of creating holistic data-driven solutions to your thorniest problems. You don’t need to be a polymath to be a full stack data scientist—you simply need to learn how to do data science on the cloud.

Saying that the cloud is what makes full stack data scientists possible seems like a very tall claim. This hinges on what I mean by “cloud”—I don’t mean simply migrating workloads that run on-premises to infrastructure that is owned by a public cloud vendor. I’m talking, instead, about truly autoscaling, managed services that automate a lot of the infrastructure provisioning, monitoring, and management—services such as Google BigQuery, Vertex AI, Cloud Dataflow, and Cloud Run on Google Cloud Platform. When you consider that the scaling and fault-tolerance of many data analysis and processing workloads can be effectively automated, provided the right set of tools is being used, it is clear that the amount of IT support that a data scientist needs dramatically reduces with a migration to the cloud.

At the same time, data science tools are becoming simpler and simpler to use. The wide availability of frameworks like Spark, scikit-learn, and Pandas has made data science and data science tools extremely accessible to the average developer—no longer do you need to be a specialist in data science to create a statistical model or train a random forest. This has opened up the field of data science to people in more traditional IT roles.

Similarly, data analysts and database administrators today can have completely different backgrounds and skill sets because data analysis has usually involved serious SQL wizardry, and database administration has typically involved deep knowledge of database indices and tuning. With the introduction of tools like BigQuery, in which tables are denormalized and the administration overhead is minimal, the role of a database administrator is considerably diminished. The growing availability of turnkey visualization tools like Tableau and Looker that connect to all the data stores within an enterprise makes it possible for a wider range of people to directly interact with enterprise warehouses and pull together compelling reports and insights.

The reason that all these data-related roles are merging together, then, is because the infrastructure problem is becoming less intense and the data analysis and modeling domain is becoming more democratized.

If you think of yourself today as a data scientist, or a data analyst, or a database administrator, or a systems programmer, this is either totally exhilarating or totally unrealistic. It is exhilarating if you can’t wait to do all the other tasks that you’ve considered beyond your ken if the barriers to entry have fallen as low as I claim they have. If you are excited and raring to learn the things you will need to know in this new world of data, welcome! This book is for you.

If my vision of a blend of roles strikes you as an unlikely dystopian future, hear me out. The vision of autoscaling services that require very little in the form of infrastructure management might be completely alien to your experience if you are in an enterprise environment that is notoriously slow moving—there is no way, you might think, that data roles are going to change as dramatically as all that by the time you retire.

Well, maybe. I don’t know where you work, and how open to change your organization is. What I believe, though, is that more and more organizations and more and more industries are going to be like the tech industry in San Francisco. There will be increasingly more openings for full stack data scientists, and data engineers will be as sought after as data scientists are today. This is because data engineers will be people who can do data science and know enough about infrastructure so as to be able to run their data science workloads on the public cloud. It will be worthwhile for you to learn data science terminology and data science frameworks, and make yourself more valuable for the next decade.

Collaboration

Even if you work in a company with strict separation of responsibilities, it can be helpful to know how the other teams do their work. This is because there are many artifacts that they create that you will use, or that you will create and they will use. Knowing their requirements and constraints will help you be more effective at communicating across organizational boundaries.

The various job roles related to data and machine learning are shown in Figure 1-2. All these roles collaborate in creating a production machine learning model. Between data ingestion and the end-user interface, there are multiple hand-offs. Every such hand-off presents an opportunity for misunderstanding the requirements of the next stage or for an inability to take over what’s been created at the previous stage.

Figure 1-2. There are many job roles that need to collaborate to take a data science solution from idea to production. Every handoff carries a risk of failure.

Understanding the adjacent roles, the tools they work with, and the infrastructure that they use can help you reduce the chances of the baton getting dropped.

That said, it is very difficult to get a clean separation of responsibilities – the best organizations that I know, the ones that have hundreds to thousands of machine learning models in production, employ full stack data scientists that work on a problem from inception to production. They may have specialist data analysts, data engineers, data scientists, or ML engineers but mostly in a maintenance capacity – the innovation tends to be done by the full stack folks.

Target audience for the book

The target audience for this book is anyone who does computing with data. If you are a data analyst, database administrator, data engineer, data scientist, or systems programmer today, this book is for you. Your role will evolve to require both creating data science models and implementing them at scale in a production-ready system that has reliability and security considerations, or working closely with teams that do these things. In this book, we’ll talk about all aspects of data-based services because you will be involved from the designing of those services, to the development of the statistical and machine learning models, to the scalable production of those services in real time.

Best Practices

This entire book consists of an extended case study. Solving a real-world, practical problem will help cut through all the hype that surrounds big data, machine learning, cloud computing, and so on. Pulling a case study apart and putting it together in multiple ways illuminates the capabilities and shortcomings of the various big data and machine learning tools that are available to you. A case study can help you identify the kinds of data-driven decisions that you can make in your business and illuminate the considerations behind the data you need to collect and curate, and the kinds of statistical and machine learning models you can use. I will attempt, throughout this book, to apply current best practice.

Simple to Complex Solutions

One of the ways that this book will mirror practice is that I will use a real-world dataset to solve a realistic scenario and address problems as they come up. So, I will begin with a decision that needs to be made and apply different statistical and machine learning methods to gain insight into making that decision in a data-driven manner. This will give you the ability to explore other problems and the confidence to solve them from first principles. As with most things, I will begin with simple solutions and work my way to more complex ones. Starting with a complex solution will only obscure details about the problem that are better understood when solving it in simpler ways. Of course, the simpler solutions will have drawbacks, and these will help to motivate the need for additional complexity.

One thing that I do not do, however, is to go back and retrofit earlier solutions based on knowledge that I gain in the process of carrying out more sophisticated approaches. In your practical work, however, I strongly recommend that you maintain the software associated with early attempts at a problem, and that you go back and continuously enhance those early attempts with what you learn along the way. Parallel experimentation is the name of the game. Due to the linear nature of a book, I don’t do it, but I heartily recommend that you continue to actively maintain several models. Given the choice of two models with similar accuracy measures, you can then choose the simpler one—it makes no sense to use more complex models if a simpler approach can work with some modifications. Another reason to have multiple models is that a drop-in replacement is useful to have if you discover that the current production model drops in accuracy or is discovered to have unwanted behaviors.

Cloud Computing

Before I joined Google, I was a research scientist working on machine learning algorithms for weather diagnosis and prediction. The machine learning models involved multiple weather sensors, but were highly dependent on weather radar data. A few years ago, when we undertook a project to reanalyze historical weather radar data using the latest algorithms, it took us four years to do. However, more recently, my team was able to build rainfall estimates off the same dataset, but were able to traverse the dataset in about two weeks. You can imagine the pace of innovation that results when you take something that used to take four years and make it doable in two weeks.

Four years to two weeks. The reason was that much of the work as recently as five years ago involved moving data around. We’d retrieve data from tape drives, stage it to disk, process it, and move it off to make way for the next set of data. Figuring out what jobs had failed was time consuming, and retrying failed jobs involved multiple steps including a human in the loop. We were running it on a cluster of machines that had a fixed size. The combination of all these things meant that it took incredibly long periods of time to process the historical archive. After we began doing everything on the public cloud, we found that we could store all of the radar data on cloud storage, and as long as we were accessing it from virtual machines (VMs) in the same region, data transfer speeds were fast enough. We still had to stage the data to disks, carry out the computation, and bring down the VMs, but this was a lot more manageable. Simply lowering the amount of data movement between tape and disk and running the processes on many more machines enabled us to carry out processing much faster; to the credit of elasticity in the public cloud.

Was it more expensive to run the jobs on 10 times more machines than we did when we did the processing on-premises? No, because the economics are in favor of renting on demand rather than buying the processing power outright especially if you will not be using the machines 24x7. Whether you run 10 machines for 10 hours or 100 machines for 1 hour, the cost remains the same. Why not, then, get your answers in an hour rather than 10 hours?

In this book, we will do all our data science on Google Cloud in order to take advantage of the near-infinite scale that the public cloud offers.

Serverless

As it turns out, though, we were still not taking full advantage of what the cloud has to offer. We could have completely foregone the process of spinning up VMs, installing software on them, and looking for failed jobs—what we should have done was to use an autoscaling data processing pipeline such as Cloud Dataflow. Had we done that, we would have been able to run our jobs on thousands of machines and might have brought our processing time from two weeks to a few hours. Not having to manage any infrastructure is itself a huge benefit when it comes to trawling through terabytes of data. Having the data processing, analysis, and machine learning autoscale to thousands of machines is a bonus.

The key benefit of performing data engineering in the cloud is the amount of time that it saves you. You shouldn’t need to wait days or months—instead, because many jobs are embarrassingly parallel, you can get your results in minutes to hours by having them run on thousands of machines. You might not be able to afford permanently owning so many machines, but it is definitely possible to rent them for minutes at a time. These time savings make autoscaled services on a public cloud the logical choice to carry out data processing.

Running data jobs on thousands of machines for minutes at a time requires fully managed services. Storing the data locally on the virtual machines or persistent disks as with the Apache Hadoop cluster doesn’t scale unless you know precisely what jobs are going to be run, when, and where. You will not be able to downsize the cluster of machines if you don’t have automatic retries for failed jobs and more importantly shuffle the data around in remaining data nodes (assuming there is enough free space). The total computation time will be the time taken by the most overloaded worker unless you have dynamic task shifting among the nodes in the cluster. All of these point to the need for autoscaling services that dynamically resize the cluster, split jobs down into tasks, move tasks between compute nodes, and can rely on highly efficient networks to move data to the nodes that are doing the processing.

On Google Cloud Platform, the key autoscaling, fully managed, “serverless” services are BigQuery (for data analytics), Cloud Spanner (for databases), Cloud Dataflow (for data processing pipelines), Google Cloud Pub/Sub (for message-driven systems), Google Cloud Bigtable (for high-throughput ingest), Cloud Run / Functions (for applications, tasks), and Vertex AI (for machine learning). Using autoscaled services like these makes it possible for a data engineer to begin tackling more complex business problems because they have been freed from the world of managing their own machines and software installations whether in the form of bare hardware, virtual machines, or containers. Given the choice between a product that requires you to first configure a container, server, or cluster, and another product that frees you from those considerations, choose the serverless one. You will have more time to solve the problems that actually matter to your business.

A Probabilistic Decision

Imagine that you are about to take a flight and just before the flight takes off from the runway (and you are asked to switch off your phone), you have the opportunity to send one last text message. It is past the published departure time and you are a bit anxious. Figure 1-3 presents a graphic view of the scenario.

Figure 1-3. A graphic illustration of the case study: if the flight departs late, should the road warrior cancel the meeting?

The reason for your anxiety is that you have scheduled an important meeting with a client at its offices. As befits a rational data scientist,6 you scheduled things rather precisely. You have taken the airline at its word with respect to when the flight would arrive, accounted for the time to hail a taxi, and used an online mapping tool to estimate the time to the client’s office. Then, you added some leeway (say 30 minutes) and told the client what time you’d meet them. And now, it turns out that the flight is departing late. So, should you send a text informing your client that you will not be able to make the meeting because your flight will be late or should you not?

This decision could be made in many ways, including by gut instinct and using heuristics. Being very rational people, we (you and I) will make this decision informed by data. Also, we see that this is a decision made by many of the road warriors in our company day in and day out. It would be a good thing if we could do it in a systematic way and have a corporate server send out an alert to travelers about anticipated delays if we see events on their calendar that they are likely to miss. Let’s build a data framework to solve this problem.

Probabilistic Approach

If we decide to make the decision in a data-driven way, there are several approaches we can take. Should we cancel the meeting if there is greater than a 30% chance that you will miss it? Or should we assign a cost to postponing the meeting (the client might go with our competition before we get a chance to demonstrate our great product) versus not making it to a scheduled meeting (the client might never take our calls again) and minimize our expected loss in revenue? The probabilistic approach translates to risk, and many practical decisions hinge on risk. In addition, the probabilistic approach is more general because if we know the probability and the monetary loss associated with missing the meeting, it is possible to compute the expected value of any decision that we make. For example, suppose the chance of missing the meeting is 20% and we decide to not cancel the meeting (because 20% is less than our decision threshold of 30%). But there is only a 25% chance that the client will sign the big deal (worth a cool million bucks) for which you are meeting them. Because there is an 80% chance that we will make the meeting, the expected upside value of not canceling the meeting is 0.8 * 0.25 * 1 million, or $200,000. The downside value of not canceling is that we do miss the meeting. Assuming that the client is 90% likely to blow us off in the future if we miss a meeting with them, the expected downside value is 0.2 * 0.9 * 0.25 * 1 million, or $45,000. This yields an expected value of $155,000 in favor of not canceling the meeting. We can adjust these numbers to come up with an appropriate probabilistic decision threshold.

Another advantage of a probabilistic approach is that we can directly take into account human psychology. You might feel frazzled if you arrive at a meeting only two minutes before it starts and, as a result, might not be able to perform at your best. It could be that arriving only two minutes early to a very important meeting doesn’t feel like being on time. This obviously varies from person to person, but let’s say that this time interval that you need to settle down is 15 minutes. You want to cancel a meeting for which you cannot arrive 15 minutes early. You could also treat this time interval as your personal risk aversion threshold, a final bit of headroom if you will. Thus, you want to arrive at the client’s site 15 minutes before the meeting and you want to cancel the meeting if there is a less than 70% chance of doing that. This, then, is our decision criterion:

Cancel the client meeting if the probability of arriving 15 minutes early is 70% or less.

I’ve explained the 15 minutes, but I haven’t explained the 70%. Surely, you can use the aforementioned model diagram (Figure 1-3, in which we modeled our journey from the airport to the client’s office), plug in the actual departure delay, and figure out what time you will arrive at the client’s offices. If that is less than 15 minutes before the meeting starts, you should cancel! Where does the 70% come from?

Probability Density Function

It is important to realize that the model diagram (Figure 1-3) of times is not exact. The probabilistic decision framework gives you a way to treat this in a principled way. For example, although the airline company says that the flight duration is 127 minutes and publishes an arrival time, not all flights are exactly 127 minutes long. If the plane happens to take off with the wind, catch a tail wind, and land against the wind, the flight might take only 90 minutes. Flights for which the winds are all precisely wrong might take 127 minutes (i.e., the airline might be publishing worst-case scenarios for the route). Google Maps predicts journey times based on historical data, and the actual journeys by taxi will be centered around those times. Your estimate of how long it takes to walk from the airport gate to the taxi stand might be predicated on landing at a specific gate, and actual times may vary. So, even though the model depicts a certain time between airline departure and your arrival at the client site, this is not an exact number. The actual time between departure and arrival might have a distribution that looks like that shown in Figure 1-4.

Figure 1-4. There are many possible values for the time differences between aircraft departure and your arrival at a client site, and the distribution of that value is called the probability distribution function

The curve in Figure 1-4 is referred to as the probability density function (abbreviated as the PDF). In fact, the PDF can be (and often is) greater than one. In order to get a probability, you will need to integrate the probability density function. A simple way to do this integration is provided by the Cumulative Distribution Function (CDF).

Cumulative Distribution Function

The cumulative probability distribution function of a value x is the probability that the observed value X is less than the threshold x. For example, you can get the Cumulative Distribution Function (CDF) for 227 minutes by finding the fraction of flights for which the time difference is less than 227 minutes, as shown in Figure 1-5.

Figure 1-5. The CDF is easier to understand and keep track of than the PDF. In particular, it is bounded between 0 and 1, whereas the PDF could be greater than 1.

Let’s interpret the graph in Figure 1-5. What does a CDF(227 minutes) = 0.8 mean? It means that 80% of flights will arrive such that we will make it to the client’s site in less than 227 minutes—this includes both the situation in which we can make it in 100 minutes and the situation in which it takes us 226 minutes. The CDF, unlike the PDF, is bounded between 0 and 1. The y-axis value is a probability, just not the probability of an exact value. It is, instead, the probability of observing all values less than that value.

Because the time to get from the arrival airport to the client’s office is unaffected by the flight’s departure delay, we can ignore it in our modeling. We can similarly ignore the time to walk through the airport, hail the taxi, and get ready for the meeting. So, we need only to find the probability of the arrival delay being more than 15 minutes. If that probability is 0.3 or more, we will need to cancel the meeting. In terms of the CDF, that means that the probability of arrival delays of less than 15 minutes has to be at least 0.7, as presented in Figure 1-6.

Thus, our decision criteria translate to the following:

Cancel the client meeting if the CDF of an arrival delay of 15 minutes is less than 70%.

Figure 1-6. Our decision criterion is to cancel the meeting if the CDF of an arrival delay of 15 minutes is less than 70%. Loosely speaking, we want to be 70% sure of the aircraft arriving no more than 15 minutes late.

The rest of this book is going to be about building data pipelines that enable us to compute the CDF of arrival delays using statistical and machine learning models. From the computed CDF of arrival delays, we can look up the CDF of a 15-minute arrival delay and check whether it is less than 70%.

Data and Tools

What data do we need to predict the probability of a specific flight delay? What tools shall we use? Should we use Hadoop? BigQuery? Should we do it on my laptop or should we do it in the public cloud? The question about data is easily answered—we will use historical flight arrival data published by the US Bureau of Transportation Statistics, analyze it, and use it to inform our decision. Often, a data scientist would choose the best tool based on their experience and just use that one tool to help make the decision, but here, I will take you on a tour of several ways that we could carry out the analysis. This will also allow us to model best practice in the sense of picking the simplest tool and analysis that suffices.

On a cursory examination of the data, we discover that there were more than 30.6 million flights in 2015-2019.7 My laptop, nice as it is, is not going to cut it. We will do the data analysis on the public cloud. Which cloud? We will use the Google Cloud Platform (GCP). Although some of the tools we use in this book (notably Hadoop, Spark, Beam, TensorFlow, etc.) are available on other cloud platforms, the managed services I use (BigQuery, Cloud Dataproc, Cloud Dataflow, Vertex AI, etc.) are specific to the GCP. Using GCP will allow me to avoid fiddling around with virtual machines and machine configuration and focus solely on the data analysis. Also, I do work at Google, so this is the platform I know best.

This book is not an exhaustive look at data science—there are other books (often based on university courses) that do that. Instead, the information it contains allows you to look over my shoulder as I solve one particular data science problem using a variety of methods and tools. I promise to be quite chatty and tell you what I am thinking and why I am doing what I am doing. Instead of presenting you with fully formed solutions and code, I will show you intermediate steps as I build up to a solution.

This learning material is presented to you in three forms:

1. This book that you are reading.

2. The code that is referenced throughout the book is on GitHub at https://github.com/GoogleCloudPlatform/data-science-on-gcp. Note in particular, the README.md file in each folder of the GitHub repository.

3. Labs with instructions that allow you to try the code of this book in a sandbox environment are available at https://qwiklabs.com

Rather than simply read this book cover to cover, I strongly encourage you to follow along with me by also taking advantage of the code. After reading each chapter, try to repeat what I did, referring to the code if something’s not clear.

Getting Started with the Code

To begin working with the code, follow these steps:

1. Visit https://console.cloud.google.com and sign up for the free trial if you haven’t already done so. Otherwise, use your existing GCP account.

2. Create a new project. You can name it data-science-on-gcp. GCP will assign a unique project ID to your project. You will need to provide that ID whenever you do anything that is billable. Once you are finished working through this book, simply delete the project to stop getting billed.

3. Open CloudShell, your terminal access to GCP. To open CloudShell, on the menu bar, click the Activate Cloud Shell icon, as shown in Figure 1-7. Even though the web console is very nice, I typically prefer to script things rather than go through a graphical user interface (GUI). To me, web GUIs are great for occasional and/or first-time use, but for repeatable tasks, nothing beats the terminal.

Figure 1-7. Activate Cloud Shell by clicking on the highlighted icon in the GCP web console.

Cloud Shell

Cloud Shell is a micro-VM that is alive for the duration of the browser window and gives you terminal access to the micro-VM. Close the browser window, and the micro-VM goes away. The CloudShell VM is free and comes loaded with many of the tools that developers on Google Cloud Platform will need. For example, it has Python, Git, the Google Cloud SDK and Orion (a web-based code editor) installed on it. Although the CloudShell VM is ephemeral, it is attached to a persistent disk that is tied to your user account. Files that you store in your home directory are saved across different CloudShell sessions.

In the CloudShell window, git clone my repository by typing the following:

git clone \

https://github.com/GoogleCloudPlatform/data-science-on-gcp

cd data-science-on-gcp

Because the Git repository was checked out to the home directory of the Cloud Shell micro-VM, it will be persistent across browser sessions.

DEVELOPING LOCALLY

If you prefer to do development on your local machine (rather than in Cloud Shell), you will need to install three pieces of software (all three are already present on Cloud Shell, so this is only if you wish to develop on your own laptop):

1. Python version 3.6 or higher.

2. The Google Cloud SDK. Installation instructions are available at https://cloud.google.com/sdk/docs/install

3. The version control software git

That’s it. You are now ready to follow along with me. As you do, remember that you need to change my project ID (cloud-training-demos) to the ID of your project (you can find this on the dashboard of the Google Cloud web console) and bucket-name (gs://cloud-training-demos-ml/) to your bucket on Cloud Storage (you will create this in Chapter 2; we’ll introduce buckets at that time).

Summary

A key goal of data analysis is to be able to provide data-driven guidance toward making accurate decisions systematically. Ideally, this guidance can be provided as a service, and providing as a service gives rise to questions of service quality—both in terms of the accuracy of the guidance and the reliability, latency, and security of the implementation.

A data engineer needs to be able to go from designing data-based services and building statistical and machine learning models to implementing them as reliable, high-quality services. This has become easier with the advent of cloud services that provide an autoscaling, serverless, managed infrastructure. Also, the wide availability of data science tools has made it so that you don’t need to be a specialist in data science to create a statistical or machine learning model. As a result, the ability to work with data will spread throughout an enterprise—no longer will it be a restricted skill.

Our case study involves a traveler who needs to decide whether to cancel a meeting depending on whether the flight they are on will arrive late. The decision criterion is that the meeting should be canceled if the probability of an arrival delay of 15 minutes is less than 70%. To estimate the probability of this arrival delay, we will use historical data from the US Bureau of Transportation Statistics.

To follow along with me throughout the book, create a project on Google Cloud Platform and a clone of the GitHub repository of the source code listings in this book. Alternatively, try the code of this book in a sandbox environment using Qwiklabs. The folder for each of the chapters in GitHub contains a README.md file that lists the steps to be able to replicate what I do in the chapters. So, if you get stuck, refer to those README files.

TIP

Incidentally, the footnotes in this book are footnotes because they break the flow of the chapter. Some readers of the first edition noted that they realized only towards the middle of the book that many of the footnotes contained useful information. So, this might be a good time to read the footnotes if you have been skipping them.

1   The classic paper on this is George Akerlof’s 1970 paper titled “The Market for Lemons.” Akerlof, Michael Spence (who explained signaling), and Joseph Stiglitz (who explained screening) jointly received the 2001 Nobel Prize in Economics for describing this problem.

2   The odometer itself might not be all that expensive, but collecting that information and ensuring that it is correct has considerable costs. The last time I sold a car, I had to sign a statement that I had not tampered with the odometer, and that statement had to be notarized by a bank employee with a financial guarantee. This was required by the company that was loaning the purchase amount on the car to the buyer. Every auto mechanic is supposed to report odometer tampering, and there is a state government agency that enforces this rule. All of these costs are significant.

3   Contrary to what you may hear, it is not about whether you use SQL or Python! You can do data science in SQL – we will see BigQuery ML later on in the book, and you can use Python for one-off data analysis.

4   In 2014, New York city officials released a public dataset of New York city trips in response to a Freedom of Information request. However, because it was improperly anonymized, a brute force attack was able to find out the trips associated with any specific driver. It got worse. Privacy researchers were able to cross-reference paparazzi photos (which revealed the exact location of celebrities at specific times) and figure out which celebrities don’t tip. It got even worse. By looking at people who picked up a taxi cab at the same location every morning, and correlating it with the location from where they got dropped back, they were able to identify New Yorkers who frequented strip clubs.

5   Google invented the role of Site Reliability Engineers (SREs) – these are folks in charge of keeping systems running. Unlike traditional IT, though, they know the software they are operating and are quite capable of making changes to it.

6   Perhaps I’m simply rationalizing my own behavior—if I’m getting to the departure gate with more than 15 minutes to spare at least once in about five flights, I decide that I must be getting to the airport too early and adjust accordingly. Fifteen minutes and 20% tend to capture my risk aversion. If you are wondering why my risk aversion threshold is not simply 15 minutes but includes an associated probabilistic threshold, read on.

7   Yes, this is the second edition of the book, published in 2021. I didn’t update the book to use more timely data because 2020 was the year of the COVID pandemic, and flights were rather spotty. So, we’ll use 2015-2019 data in this book.

Chapter 2. Ingesting Data into the Cloud

A NOTE FOR EARLY RELEASE READERS

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 2nd chapter of the final book. Please note that the GitHub repo will be made active later on.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at mcronin@oreilly.com.

In Chapter 1, we explored the idea of deciding whether to cancel a meeting in a data-driven way. We decided on a probabilistic decision criterion: to cancel the meeting with a client if the probability of the flight arriving within 15 minutes of the scheduled arrival time was less than 70%. To model the arrival delay given a variety of attributes about the flight, we need historical data that covers a large number of flights. Historical data that includes this information from 1987 onward is available from the US Bureau of Transportation Statistics (BTS). One of the reasons that the government captures this data is to monitor the fraction of flights by a carrier that are on-time (defined as flights that arrive less than 15 minutes late), so as to be able to hold airlines accountable.1 Because the key use case is to compute on-time performance, the dataset that captures flight delays is called Airline On-time Performance Data. That’s the dataset we will use in this book.

TIP

All of the code snippets in this chapter are available in the GitHub repository at https://github.com/GoogleCloudPlatform/data-science-on-gcp/ in the folder 02_ingest. See the last section of Chapter 1 for instructions on how to clone the repository and the README.md file in the 02_ingest directory for instructions on how to do the steps described in this chapter.

Airline On-Time Performance Data

For the past 35 years, all major US air carriers have been required to file statistics about each of their domestic flights with the BTS. The data they are required to file includes the scheduled departure and arrival times as well as the actual departure and arrival times. From the scheduled and actual arrival times, the arrival delay associated with each flight can be calculated. Therefore, this dataset can give us the true value or “label” for building a model to predict arrival delay.

The actual departure and arrival times are defined rather precisely, based on when the parking brake of the aircraft is released and when it is later reactivated at the destination. The rules even go as far as to define what happens if the pilot forgets to engage the parking brake—in that case, the time that the passenger door is closed or opened is used instead. Because of the precise nature of the rules, and the fact that they are enforced, we can treat arrival and departure times from all carriers uniformly. Had this not been the case, we would have to dig deeper into the quirks of how each carrier defines “departure” and “arrival,” and do the appropriate translations.2 Good data science begins with such standardized, repeatable, trustable data collection rules; you should use the BTS’s very well-defined data collection rules as a model when creating standards for your own data collection, whether it is log files, web impressions, or sensor data that you are collecting. The airlines report this particular data monthly, and it is collated by the BTS across all carriers and published as a free dataset on the web.

In addition to the scheduled and actual departure and arrival times, the data includes information such as the origin and destination airports, flight numbers, and nonstop distance between the two airports. It is unclear from the documentation whether this distance is the distance taken by the flight in question or whether it is simply a precomputed distance—if the flight needs to go around a thunderstorm, is the distance in the dataset the actual distance traveled by the flight or the great-circle3 distance between the airports? This is something that we will need to examine—it should be easy to ascertain whether the distance between a pair of airports remains the same or changes. In addition, a flight is broken into three parts (Figure 2-1)—taxi-out duration, air time, and taxi-in duration—and all three time intervals are reported.

Figure 2-1. A flight is broken into three parts: taxi-out duration, air time, and taxi-in duration

Knowability

Before we get started with ingesting data, we need to decide what it is that we have to ingest into our model. There are two potential traps—causality and training-serving skew (I’ll define them shortly). We should take care to avoid these problems during the ingest phase, in order to save us a lot of heartburn later.

The key question boils down to this: what data will we be able to provide to the model at the time that we need to make predictions?

Some of the fields in the dataset could form the inputs to our model to help us predict the arrival delay as a function of these variables. Some, but not all. Why? It should be clear that we cannot use taxi-in duration or actual flight distance because at the time the aircraft is taking off, which is when we want to make our decision on whether to cancel the meeting, we will not know either of these things. The in-air flight time between two airports is not known a priori given that pilots have the ability to speed up or slow down. Thus, even though we have these fields in our historical dataset, we should not use them in our prediction model. This is called a causality constraint.

The causality constraint is one instance of a more general principle. Before using any field as input to a model, we should consider whether the data will be known at the time we want to make the decision. It is not always a matter of logic as with the taxi-in duration. Sometimes, practical considerations such as security (is the decision maker allowed to know this data?), the latency between the time the data is collected and the time it is available to the model, and cost of obtaining the information also play a part in making some data unusable. At the same time, it is possible that approximations might be available for fields that we cannot use because of causality—even though, for example, we cannot use the actual flight distance, we should be able to use the great-circle distance between the airports in our model.

Similarly, we might be able to use the data itself to create approximations for fields that are obviated by the causality constraint. Even though we cannot use the actual taxi-in duration, we can use the mean taxi-in duration of this flight at this airport on previous days, or the mean taxi-in duration of all flights at this airport over the past hour to approximate what the taxi-in duration might be. Over the historical data, this could be a simple batch operation after grouping the data by airport and hour. When predicting in real time, though, this will need to be a moving average on streaming data. Indeed, approximations to unknowable data will be an important part of our models.

Training–Serving Skew

A training–serving skew is the condition in which you use a variable that’s computed differently in your training dataset than in the production model. For example, suppose that you train a model with the distance between cities in miles, but when you predict, the distance that you receive as input is actually in kilometers. That is obviously a bad thing and will result in a bad result from the model because the model will be providing predictions based on the distances being 1.6 times their actual value. Although it is obvious in clear-cut cases such as unit mismatches, the same principle (that the training dataset has to reflect what is done to inputs at prediction time) applies to more subtle scenarios as well.

For example, it is important to realize that even though we have the actual taxi-in duration in the data, we cannot use that taxi-in duration in our modeling. Instead, we must approximate the taxi-in duration using time aggregates and use those time aggregates in our training; otherwise, it will result in a training–serving skew. If our model uses taxi-in duration as an input, and that input in real-time prediction is going to be computed as an average of taxi-in durations over the previous hour, we will need to ensure that we also compute the average in the same way during training. We cannot use the recorded taxi-in as it exists in the historical dataset. If we did that, our model will be treating our time averages (which will tend to have the extrema averaged out) as the actual value of taxi-in duration (which in the historical data contained extreme values). If the model, in our training, learns that such extreme values of taxi-in duration are significant, the training–serving skew caused by computing the taxi-in duration in different ways could be as bad as treating kilometers as miles.

As our models become increasingly sophisticated—and more and more of a black box—it will become extremely difficult to troubleshoot errors that are caused by a training–serving skew. This is especially true if the code bases for computing inputs for training and during prediction are different and begin to diverge over time. We will always attempt to design our systems in such a way that the possibilities of a training–serving skew are minimized. In particular, we will gravitate toward solutions in which we can use the same code in training (building a model) as in prediction.

The dataset includes codes for the airports (such as ATL for Atlanta) from which and to which the flight is scheduled to depart and land. Planes might land at an airport other than the one they are scheduled to land at if there are in-flight emergencies or if weather conditions cause a deviation. In addition, the flight might be canceled. It is important for us to ascertain how these circumstances are reflected in the dataset—although they are relatively rare occurrences, our analysis could be adversely affected if we don’t deal with them in a reasonable way. The way we deal with these out-of-the-ordinary situations also must be consistent between training and prediction.

The dataset also includes airline codes (such as AA for American Airlines), but it should be noted that airline codes can change over time (for example, United Airlines and Continental Airlines merged and the combined entity began reporting as United Airlines in 2012). If we use airline codes in our prediction, we will need to cope with these changes in a consistent way, too.

Downloading Data

As of August 2021, there were nearly 200 million records in the on-time performance dataset, with records starting in 1987. The last available data was June 2021, indicating that there is more than a month’s delay in updating the dataset – this is going to be important when we automate the process of getting new data.

In this book, our model will use input fields drawn mostly from this dataset, but where feasible and necessary, we will include other datasets such as airport locations and weather. We can download the on-time performance data from the BTS website as comma-separated value (CSV) files. The web interface requires you to filter by geography and period, as illustrated in Figure 2-2. The data itself is offered in two ways: one, with all the data in a zipped file and the other, of just the fields that we select in the form.

Figure 2-2. The BTS web interface to download the flights on-time arrival dataset

This is not the most helpful way to provide data for download. For one thing, the data can be downloaded only one month at a time. For another, going through a web form is pretty error-prone. Imagine that you want to download all of the data for 2015. In that scenario, you’d painstakingly select the fields you want for January 2015, submit the form, and then have to repeat the process for February 2015. If you forgot to select a field in February, that field would be missing, and you wouldn’t know until you began analyzing the data!

Obviously, we can script the download to make it less tiresome and ensure consistency.4 However, it is better to download all the raw data, not just a few selected fields. Why? Won’t the files be larger if we ask for all the fields? Won’t larger files take longer to download?

Hub and Spoke Architecture

Yes, the files will be larger if we download all the fields using the static link. But there is a significant drawback to doing preselection. In order to support the interactive capability of selecting fields, the BTS does server-side processing – it extracts the fields we want, creates a custom zip file and makes the zip file available for download. This would make our code reliant on the BTS servers having the necessary uptime and reliability.5 Avoiding the server-side processing should help reduce this dependency.6

An even more salient reason is that best practice in data engineering now is to build ELT (Extract-Load-Transform) pipelines, rather than ETL pipelines. What this means is that we will extract the data from BTS and immediately load the data into a data warehouse rather than rely on the BTS server to do transformation for us before loading it into Google Cloud. This point is important. The recommended modern data architecture is to minimize the preprocessing of data – instead, land all available data as-is into the Enterprise Data Warehouse (EDW) and then carry out whatever transformations are necessary for different use cases (see Figure 2-3). This is called a hub-and-spoke architecture, with the EDW functioning as the hub.

Figure 2-3. The recommended data architecture, whenever you can make it work, is the hub-and-spoke architecture.

Dataset Fields

Even though I’m going to download all the fields, it’s worthwhile reading through the column descriptions provided by BTS to learn more about the dataset and get a preliminary idea about what fields are relevant to our problem and whether there are any caveats. For example, Table 2-1 shows three ways in which the airline is recorded. Which of these should we use?

Table 2-1. The airline operating the flight is recorded in three separate columns.

Column Name

Description (copied from BTS website)

FlightDate

Flight Date (yyyymmdd).

Reporting_Airline

Unique Carrier Code. When the same code has been used by multiple carriers, a numeric suffix is used for earlier users; for example, PA, PA(1), PA(2). Use this field for analysis across a range of years.

DOT_ID_Reporting_Airline

An identification number assigned by the US Department of Transportation (DOT) to identify a unique airline (carrier). A unique airline (carrier) is defined as one holding and reporting under the same DOT certificate regardless of its Code, Name, or holding company/corporation.

IATA_CODE_Reporting_Airline

Assigned by International Air Transport Association (IATA) and commonly used to identify a carrier. Because the same code might have been assigned to different carriers over time, the code is not always unique. For analysis, use the Unique Carrier Code.

It’s clear that we could use either the Reporting_Airline or the DOT_ID_Reporting_Airline since they are both unique. Ideally, we’d use whichever one of these corresponds to the common nomenclature (for example, UA or United Airlines). Fortunately, the BTS provides an “analysis” link for the columns (see Figure 2-4), so we don’t have to wait until we explore the data to make this decision. It turns out that the Reporting_Airline is what we want – the IATA code consists of the number 19977 for United Airlines whereas the Reporting_Airline is UA as we would like.

Figure 2-4. The BTS provides an Analysis link for some of the columns. These provide a handy way to learn what values a field can take.

The first thing to do in any real-world problem where we are fortunate enough to be provided documentation is to read it! After reading through the descriptions of the 100-plus fields in the dataset, there are a few fields that appear relevant to the problem of training, predicting, or evaluating flight arrival delay. Table 2-2 presents the fields I shortlisted.

Table 2-2. Selected fields from the airline on-time performance dataset downloaded from the BTS (there is a separate table for each month)

Column Name

Description (copied from BTS website)

FlightDate

Flight Date (yyyymmdd).

Reporting_Airline

Unique Carrier Code. When the same code has been used by multiple carriers, a numeric suffix is used for earlier users; for example, PA, PA(1), PA(2). Use this field for analysis across a range of years.

Origin

Origin Airport.

Dest

Destination Airport.

CRSDepTime

Computerized Reservation System (CRS) Departure Time (local time: hhmm).

DepTime

Actual Departure Time (local time: hhmm).

DepDelay

Difference in minutes between scheduled and actual departure time. Early departures show negative numbers.

TaxiOut

taxi-out duration, in minutes.

WheelsOff

Wheels-Off time (local time: hhmm).

WheelsOn

Wheels-On Time (local time: hhmm).

TaxiIn

taxi-in duration, in minutes.

CRSArrTime

CRS Arrival Time (local time: hhmm).

ArrTime

Actual Arrival Time (local time: hhmm).

ArrDelay

Difference in minutes between scheduled and actual arrival time. Early arrivals show negative numbers.

Cancelled

Cancelled flight indicator (1 = Yes).

CancellationCode

Specifies the reason for cancellation.

Diverted

Diverted flight indicator (1 = Yes).

Distance

Distance between airports (miles).

Separation of Compute and Storage

There are essentially three options when it comes to processing large datasets (see Table 2-3), and all three are possible on GCP. Which one you use depends on the problem – in this book, we’ll use the third option because it is the most flexible. However, this option requires a bit of preplanning on our part – we will have to store our data in Google Cloud Storage and in Google BigQuery. Let’s see why we choose this.

Table 2-3. How to choose between scaling up, scaling out with data sharding, or scaling out with data in situ

Option

Performance and cost

Required platform capabilities

How to implement on Google Cloud Platform

Example use case

Scaling up

Expensive on both compute and storage; fast, but limited to capability of most powerful machine

Very powerful machines; ability to rent machines by the minute; attachable persistent SSDs

Compute Engine with persistent SSDs

Vertex AI Notebooks

Job that requires rereading of data (e.g. training an ML model)

Scaling out with sharding

High storage costs; inexpensive compute; add machines to achieve desired speed, but limited to ability to preshard the data on a cluster of desired size

Data local to the compute nodes; attachable persistent SSDs

Cloud Dataproc (with Spark) and HDFS

Light compute on a splittable dataset (e.g. creating a search index from thousands of documents). Many data analytics use cases used to be in this segment.

Scaling out with data in situ

Inexpensive storage, compute; add machines to achieve desired speed

Extremely fast networking, cluster-wide filesystem

Cloud Dataproc + Spark on Cloud Storage, BigQuery, Cloud Dataflow, Vertex AI Training, etc.

Any use case where we can configure datasets so that I/O keeps up with computation. Data analytics use cases are increasingly falling into this segment.

Even if you are used to downloading data to your laptop for data analysis and development, you should realize that this is a suboptimal solution. Would it be great to directly ingest the BTS files into our data analysis programs without having to go through a step of downloading them? Having a single source of truth has many advantages, ranging from security (providing and denying access) to error correction (no need to worry about stale copies of the data lying around). Of course, the reason we don’t do this is that we’d have to read the BTS data over the internet, and the public internet typically has speeds of 3 to 10 MBps. If you are carrying out analysis on your laptop, accessing data via the internet every time you need it will become a serious bottleneck.

Figure 2-5. Comparison of data access speeds if data is accessed over the public internet versus from a disk drive

Downloading the data has the benefit that subsequent reads happen on the local drive and this is both inexpensive and fast (see Figure 2-5). For small datasets and short, quick computation, it’s perfectly acceptable to download data to your laptop and do the work there. This doesn’t scale, though. What if our data analysis is very complex or the data is so large that a single laptop is no longer enough? We have two options: scale up or scale out.

Scaling Up

One option to deal with larger datasets or more difficult computation jobs is to use a larger, more powerful machine with many CPUs/GPUs, lots of RAM, and many terabytes of drive space. This is called scaling up, and it is a perfectly valid solution. However, such a computer is likely to be quite expensive. Because we are unlikely to be using it 24 hours a day, we might choose to rent an appropriately large computer from a public cloud provider. In addition, the public cloud offers persistent drives that can be shared between multiple instances and whose data is replicated to guard against data loss. In short, then, if you want to do your analysis on one large machine but keep your data permanently in the cloud, a good solution would be to marry a powerful, high-memory Compute Engine instance with a persistent drive, download the data from the external datacenter (BTS’s computer in our case) onto the persistent drive, and start up compute instances on demand, as depicted in Figure 2-6 (cloud prices in Figure 2-6 are estimated monthly charges; actual costs may be higher or lower than the estimate).

Figure 2-6. One solution to cost-effective and fast data analysis is to store data on a persistent disk that is attached to an ephemeral, high-memory Compute Engine instance

When you are done with the analysis, you can delete the Compute Engine instance.7 Provision the smallest persistent drive that adequately holds your data — temporary storage (or caches) during analysis can be made to an attached SSD that is deleted along with the instance, and persistent drives can always be resized if your initial size proves too small. This gives you all the benefits of doing local analysis but with the ability to use a much more powerful machine at a lower cost. I will note here that this recommendation assumes several things: the ability to rent powerful machines by the minute, to attach resizeable persistent drives to compute instances, and to achieve good-enough performance by using solid-state persistent drives. These are true of Google Cloud and other public cloud providers, but are unlikely to be true on-premises.

Scaling up is a common approach whenever you have a job that needs to read the data multiple times. This is quite common when training machine learning models, and so scaling up is a common approach in machine learning, especially machine learning on images and video. Indeed, Google Cloud offers special Compute Engine instances, called Deep Learning VM that come preinstalled with the GPUs and libraries that are needed for machine learning. Jupyter Notebook instances are also frequently scaled up as necessary to fit the job. You’d create a Deep Learning VM, and attach to it a persistent disk containing the training data.

Scaling Out with Sharded Data

The solution of using a high-memory Compute Engine instance along with persistent drives and caches might be reasonable for jobs that can be done on a single machine, but it doesn’t work for jobs that are bigger than that. Configuring a job into smaller parts so that processing can be carried out on multiple machines is called scaling out . One way to scale out a data processing job is to shard 8 the data and store the pieces on the drives attached to multiple compute instances or persistent drives that will be attached to multiple instances. Then, each compute instance can carry out analysis on a small chunk of data at high speeds—these operations are called the map operations. The results of the analysis on the small chunks can be combined, after some suitable collation, on a different set of compute nodes—these combination operations are called the reduce operations. Together, this model is known as MapReduce . This approach also requires an initial download of the data from the external datacenter to the cloud. In addition, we also need to split the data onto preassigned drives or nodes.

Whenever we need to carry out analysis, we will need to spin up the entire cluster of nodes, reattach the persistent drives and carry out the computation. Fortunately, we don’t need to build the infrastructure to do the sharding or cluster creation ourselves. We could store the data on the Hadoop Distributed File System (HDFS), which will do the sharding for us, spin up a Cloud Dataproc cluster (which has Hadoop, Presto, Spark, etc. preinstalled on a cluster of Compute Engine VMs), and run our analysis job on that cluster. Figure 2-7 presents an overview of this approach.

Figure 2-7. For larger datasets, one potential solution is to store data on the HDFS and use an ephemeral Dataproc cluster to carry out the analysis

A MapReduce framework like the Hadoop ecosystem requires data to be presharded. Because the presharded data must be stored on drives that are attached to compute instances, the scheme can be highly wasteful unless all the data happens to get used all the time by those compute instances. In essence, whenever you need to run a job, you ship the code to whichever nodes happen to be storing the data. What you should be doing is to trying to find a machine that has free capacity. Shipping the analysis code to run on storage nodes regardless of their computational load leads to poor efficiency because it is likely that there are long periods during which a node might have nothing to do, and other periods when it is subject to resource contention.

In summary, we have two options to work with large datasets: keep the data as is and scale up by using a large-enough computer, or scale out by sharding the data and shipping code to the nodes that store the data. Both of these options have some drawbacks. Scaling up is subject to the limitations of whatever the most powerful machine available to you can do. Scaling out is subject to inefficiencies of resource allocation. Is there a way to keep data in situ and scale out?

Scaling out with Data in Situ

Recall that much of the economics of our case for downloading the data onto nodes on which we can do the compute relied on the slowness of an internet connection as compared to drive speeds—it is because the public internet operates at only 3 to 10 MBps, whereas drives offer two orders of magnitude faster access, that we moved the data to a large Compute Engine instance (scaling up) or sharded it onto persistent drives attached to Compute Engine instances (scaling out).

What if, though, you are operating in an environment in which networking speeds are higher, and files are available to all compute instances at those high speeds? For example, what if you had a job that uses 100,000 servers and those servers could communicate with one another at 1 GBps? This is seriously fast—it is twice the speed of SSD, 10 times the speed of a local hard drive, and 100 times faster than the public internet. What if, in addition, you have a cluster-level filesystem (not node-by-node 9 ) whose metadata is sharded across the datacenter and replicated on-write for durability? Because the total bisection bandwidth of the Jupiter networks in Google’s datacenters is 125,000 GBps 10 and because Google’s next-generation Colossus filesystem operates at the cluster level, this is the scenario that operates if your data is available in Google Cloud Storage and your jobs are running on Compute Engine instances in the same datacenter as the file. At that point, it becomes worthwhile to treat the entire datacenter as a single computer. The speed of the network and the design of the storage make both compute and data fungible resources that can be allocated to whichever part of the datacenter is most free. Scheduling a set of jobs over a single large data center provides much higher utilization than scheduling the same set of jobs over many smaller clusters. This resource allocation can be automatic—there is no need to preshard the data, and if we use an appropriate computation framework (such as BigQuery, Cloud Dataflow, or Vertex AI), we don’t even need to instantiate a Compute Engine instance ourselves. Figure 2-8 presents this framework, where compute and storage are separate.

Therefore, choose between scaling up, scaling out with data sharding, or scaling out with data in situ depending on the problem that you are solving (see Table 2-3).

Figure 2-8. On the Google Cloud Platform, the speed of the networking within a datacenter allows us to store the data persistently and cheaply on Cloud Storage and access it as needed from a variety of ephemeral managed services. This is called separation of compute and storage.

GOOGLE CLOUD DATACENTERS ARE DIFFERENT

Google datacenters, unlike most other datacenters, are optimized for total bisection bandwidth. They are optimized to maximize the network bandwidth between nodes on the backend (“East-West communications” in networking parlance). Most other datacenters are optimized to minimize the network time with an outside client sending, for example, a web request (“North-South communications”).

Why would anybody design a datacenter for East-West communications? Aren’t most applications web applications? You would design a datacenter for East-West networking only if the amount of network calls you do on the backend in response to a user request is several times the traffic of the request itself. That is true of Google (very simple user interface, very complex business logic). Fortunately, this design also comes in extremely useful for data science because it is not necessary to preshard the data.

The Google File System (or GFS, on which the Hadoop Distributed File System, or HDFS, is based) was built for batch operations, whereas Colossus was designed for real-time updates. Although GFS/HDFS suffice for batch processing operations that happen over a few days, Colossus is required to update Google’s search index in real time—this is why Google’s search can now reflect current events. There are several other innovations that were necessary to get to this data processing architecture in which data does not need to be presharded. For example, when performing large fan-out operations, you must be tolerant of latency and design around it. This involves slicing up requests to reduce head-of-line blocking,11 creating hundreds of partitions per machine to make it easy to move partitions elsewhere, making replicas of heavily used data chunks, using backup requests, and canceling other requests as soon as one is completed, among other strategies. To build a cluster-wide filesystem with high throughput speeds to any compute instance within the datacenter, it is necessary to minimize the number of network hops within the datacenter by changing the network definitions through software. Within the Google Cloud Platform, any two machines in the same zone are only one network hop away.

The innovation in networking, compute, and storage at Google and elsewhere is by no means over. Even though the Jupiter network provides bisection bandwidths of 125,000 GBps (the last time Google published this number publicly was in the mid 2010s and it’s probably higher now), engineers estimate that 600,000 GBps is what’s required to match the performance of local disks. Moreover, jobs are not being sliced finely enough—because I/O devices have response times on the order of microseconds, decisions should be scheduled even more finely than the current milliseconds. Next-generation flash storage is still largely untapped within the datacenter. Colossus addresses the issue of building a cluster-level filesystem, but there are applications that need global consistency, not just consistency within a single-region cluster. The challenge of building a globally distributed database is being addressed by Cloud Spanner. The ongoing innovations in computational infrastructure promise exciting times ahead.

All of this is in the way of noting (again!) that your mileage will vary if you do your data processing on other infrastructure—there is a reason why the title of this book includes the words “on Google Cloud Platform.” The hardware optimizations if you implement your data pipelines on-premises or in a different cloud provider will typically target different things.12 The APIs might look the same, and in many cases, you can run the same software as I do, but the performance characteristics will be different. Google TensorFlow, Apache Beam, and others are open source and portable to on-premises infrastructure and across different cloud providers, but the execution frameworks that make Vertex AI and Cloud Dataflow so powerful may not translate well to infrastructure that is not built the same way as Google Cloud Platform.

Another way to see this is that multi-cloud software works faster on Google Cloud than on other cloud platforms. BigQuery Omni, although available on AWS and Azure, does not get the performance of BigQuery on GCP. This performance difference is not limited to multi-cloud software developed by Google. Databricks notes that their GCP implementation is faster on cold startup and certain Spark workloads than on other clouds. Actian Avalanche notes that their GCP implementation is 20% faster than on other cloud platforms.

Ingesting Data

To carry out our data analysis on the on-time performance dataset, we will need to download the monthly data from the BTS website and then upload it to Google Cloud Storage. Doing this manually will be tedious and error-prone, so let’s script this operation.

Reverse Engineering a Web Form

How would you script filling out the BTS web form shown in Figure 2-2? First, verify that the website’s terms of use do not bar you from automated downloads! Then, use the Chrome browser’s developer tools to find what web calls the form makes. Once you know that, you can repeat those web calls in a script.

The BTS web form is a simple HTML form with no dynamic behavior. This type of form collects all the user selections into a single GET or POST request. If we can create that same request from a script, we will be able to obtain the data without going through the web form.

We can find out the exact HTTP command sent by the browser after we make our selections on the BTS website. You can do this while on the BTS download website in the Chrome web browser – in the upper-right menu bar of the browser, navigate to the Developer Tools menu, as shown in Figure 2-9.

Figure 2-9. Navigating to the Developer Tools menu in Chrome

Now, on the BTS website, select the Prezipped File option, select 2015 and January in the dropdowns, and click the Download button. The Developer tools menu shows us that the browser is now making a POST request for the file: https://transtats.bts.gov/PREZIP/On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2015_1.zip

It is pretty obvious what the pattern here is. If we issue a HTTP POST for a file with the pattern:

${BASEURL}_${YEAR}_${MONTH}.zip

we should get the data corresponding to a single month. Let’s try it from the command-line of Cloud Shell:

BTS=https://transtats.bts.gov/PREZIP

BASEURL="${BTS}/On_Time_Reporting_Carrier_On_Time_Performance_1987_present"

YEAR=2015

MONTH=3

curl -o temp.zip ${BASEURL}_${YEAR}_${MONTH}.zip

We see the data for March 2015 starting to get downloaded. Once the file is downloaded, we can unzip it:

unzip temp.zip

We then notice that the zip file contains a Comma Separated Values (CSV) containing the flights data for the month of January 2015.

Dataset Download

In the data exploration phase, I’ll do most of my processing interactively with Linux command-line tools. I will assume that this is what you are using as well. Adapt the commands as necessary if you are working locally in some other environment (e.g., where I ask you to do a sudo apt-get install, you might use the appropriate install command for your Linux-like environment). When we have settled on the processing to be done, we’ll look at how to make this more automated.

Instead of calling the downloaded file temp.zip, let’s call it 201501.zip and place it into a temporary directory. To pad the month 1 to be 01, we can use the printf command in bash:13

MONTH2=$(printf "%02d" $MONTH)

To create a temporary directory, we can use the Linux command mktemp:

TMPDIR=$(mktemp -d)

Then, to download the file to the temporary directory, we can do:

ZIPFILE=${TMPDIR}/${YEAR}_${MONTH2}.zip

curl -o $ZIPFILE ${BASEURL}_${YEAR}_${MONTH}.zip

Now, we can unzip the file, extract the CSV file to the current directory (./) and blow out the remaining contents of the zip file:

unzip -d $TMPDIR $ZIPFILE

mv $TMPDIR/*.csv ./${YEAR}${MONTH2}.csv

rm -rf $TMPDIR

I put the above commands into a file called download.sh and then in the script ingest.sh, I call it from within a for loop:

for MONTH in `seq 1 12`; do

bash download.sh $YEAR $MONTH

done

On running this, we get a set of CSV files, one for each month in 2015 (see Figure 2-10).

The complete download script is on GitHub at https://github.com/GoogleCloudPlatform/data-science-on-gcp/tree/master/02_ingest —if you want to follow along with me, perform these steps:

1. Go to https://console.cloud.google.com

2. On the top strip, activate Cloud Shell using the button shown in Figure 2-10.

Figure 2-10. The CloudShell button on the Google Cloud Platform web console

3. In Cloud Shell, type the following:

git clone \

https://github.com/GoogleCloudPlatform/data-science-on-gcp/

4. This downloads the GitHub code to your CloudShell home directory.

5. Navigate into the flights folder:

cd data-science-on-gcp

6. Make a new directory to hold the data, and then change into that directory:

mkdir data

cd data

7. Run the code to download the files:

for MONTH in `seq 1 12`; do

bash download.sh 2015 $MONTH

done

8. When the script completes, look at the downloaded ZIP files, shown in Figure 2-11:

ls -lrt

Figure 2-11. Details of downloaded files as seen in Cloud Shell

This looks quite reasonable—all the files have different sizes and the sizes are robust enough that one would assume they are not just error messages.

Exploration and Cleanup

At this point, I have 12 CSV files. Let’s look at the first two lines of one of them to ensure the data matches what we think it ought to be:

head -2 201503.csv

The result is shown in Figure 2-12.

Figure 2-12. The first two lines of the CSV file corresponding to March 2015.

There is a header in each CSV file, and the second line looks like data. Some of the fields are enclosed by quotes (perhaps in case the strings themselves have commas), and there are some fields that are missing (there is nothing between successive commas towards the end of the line). There seems to be a pesky extra comma at the end as well.

How many fields are there? Because the second line doesn’t have any commas between the quotes, we can check using:

head -2 201503.csv  | tail -1 | sed 's/,/ /g' | wc -w

The number of words is 81, so there are 81 columns (remember there’s a comma at the end of the line). Here’s how the command works. It first gets the first two lines of the data file (with head -2), and the last line of that (with tail -1) so that we are looking at the second line of 201503.csv. Then, we replace all the commas by spaces and count the number of words with wc -w.

How much data is there? A quick shell command (wc for wordcount, with an -l [lowercase letter L] to display only the line count) informs us that there are between 43,000 and 52,000 flights per month:

$ wc -l *.csv

469969 201501.csv

429192 201502.csv

504313 201503.csv

485152 201504.csv

496994 201505.csv

503898 201506.csv

520719 201507.csv

510537 201508.csv

464947 201509.csv

486166 201510.csv

467973 201511.csv

479231 201512.csv

5819091 total

This adds up to nearly six million flights in 2015! The slowness of this command should tell us that any kind of analysis that involves reading all the data is going to be quite cumbersome. You can repeat this for other years (2016-2019) but let’s wait until we have the whole process complete for one year before we add more years.

You might have realized by now that knowing a little Unix scripting can come in very handy at this initial stage of data analysis.

Uploading Data to Google Cloud Storage

For durability of this raw dataset, let’s upload it to Google Cloud Storage. To do that, you first need to create a bucket, essentially a namespace for Binary Large Objects (blobs) stored in Cloud Storage that you typically want to treat similarly from a permissions perspective. You can create a bucket from the Google Cloud Platform Console. For reasons that we will talk about shortly, make the bucket a single-region bucket.

Bucket names must be globally unique (i.e., unique not just within your project or organization, but across Google Cloud Platform). This means that bucket names are globally knowable even if the contents of the bucket are not accessible. This can be problematic. For example, if you created a bucket named acme_gizmo, a competitor might later try to create a bucket also named acme_gizmo, but fail because the name already exists. This failure can alert your competitor to the possibility that Acme Corp. is developing a new Gizmo. It might seem like it would take Sherlock Holmes–like powers of deduction to arrive at this conclusion, but it’s simply best that you avoid revealing sensitive information in bucket names. A common pattern to create unique bucket names is to use suffixes on the project ID. Project IDs are globally unique,14 and thus a bucket name such as <projectid>-dsongcp will also tend to be unique. In my case, my project ID is cloud-training-demos and my bucket name is cloud-training-demos-ml.

Cloud Storage will also serve as the staging ground to many of the GCP tools and enable collaborative sharing of the data with our colleagues. In my case, to upload the files to Cloud Storage, I type the following:

gsutil -m cp *.csv gs://cloud-training-demos-ml/flights/raw/

This uploads the files to Cloud Storage to my bucket cloud-training-demos-ml in a multithreaded manner (-m) and makes me the owner. If you are working locally, another way to upload the files would be to use the Cloud Platform Console.

It is better to keep these as separate files instead of concatenating them into a single large file because Cloud Storage is a blob store, and not a regular filesystem. In particular, it is not possible to append to a file on Cloud Storage; you can only replace it. Therefore, although concatenating all 12 files into a single file containing the entire year of data will work for this batch dataset, it won’t work as well if we want to later add to the dataset one month at a time, as new data become available. Secondly, because Cloud Storage is blob storage, storing the files separately will permit us to more easily process parts of the entire archive (for example, only the summer months) without having to build in slicing into our data processing pipeline. Thirdly, it is generally a good idea to keep ingested data in as raw a form as possible.

It is preferable that this bucket to which we upload the data is a single-region bucket. There are three reasons: one is that we will create Compute Engine instances in the same region as the bucket and access it only from this one region. A multi-region bucket would be overkill because we don’t need global availability. Second, a single-region bucket is less expensive than a multi-region one. Third, single-region buckets are optimized for high throughput whereas multi-region buckets are optimized for edge latency (think data analytics vs. web traffic). All three of the above factors point to using single region buckets for data analytics and machine learning.

Note that both single-region and multi-region buckets in Google Cloud Platform offer strong consistency, so this does not seem like a consideration to choose one over the other. However, the speed differences inherent in being able to offer strong consistency on global buckets points to using single-region buckets if you can. What exactly is strong versus eventual consistency, and why does it matter? Suppose that a worker in a distributed application updates a piece of data and another worker views that piece of data immediately afterward. Does the second worker always see the updated value? Then, what you have is strong consistency . If, on the other hand, there could be a potential lag between an update and availability (i.e., if different viewers can see potentially different values of the data at the same instant in time), what you have is eventual consistency . Eventually, all viewers of the data will see the updated value, but that lag will be different for different viewers. Strong consistency is an implicit assumption that is made in a number of programming paradigms. However, to achieve strong consistency, we have to make compromises on scalability and performance (this is called Brewer’s theorem ). For example, we might need to lock readers out of the data while it is being updated so that simultaneous readers always see a consistent and correct value.

NOTE

Brewer’s theorem, also called the CAP theorem, states that no computer system can simultaneously guarantee consistency, availability, and partition resilience. Consistency is the guarantee that every reader sees the latest written information. Availability is the guarantee that a response is sent to every request (regardless of whether it is the most current information or not). Partition resilience is the guarantee that the system continues to operate even if the network connecting readers, writers, and storage drops an arbitrary number of messages. Since network failures are a fact of life in distributed systems, the CAP theorem essentially says that you need to choose between consistency and availability. Cloud Spanner doesn’t change this: it essentially makes choices during partitioning—Cloud Spanner is always consistent and achieves five-9s (but not perfect) availability despite operating over a wide area. For more details, see http://bit.ly/2Abs7D8 and neither do multi-regional buckets.

If you need the performance of a regional bucket, but need to be tolerant to failure (for example, you want to be able to carry out your workload even if a region goes down), there are two options: eventual consistency and dual-region buckets. As an example of eventual consistency consider how internet Domain Name System (DNS) servers cache values and have their values replicated across many DNS servers all over the internet. If a DNS value is updated, it takes some time for this modified value to become replicated at every DNS server. Eventually, this does happen, though. Having a centralized DNS server that is locked whenever any DNS value is modified would have led to an extremely brittle system. Because the DNS system is based on eventual consistency, it is highly available and extremely scalable, enabling name lookups for/to millions of internet-capable devices. The other option is to have a dual-region bucket in a multi-region location, so that the metadata remains the same. If, for whatever reason one region is not available for analytics, computation fails over to the other region in a multi-region location (US, EU, Asia). Dual-region buckets are more expensive than either single-region buckets or multi-region buckets, but offer both high-performance and reliability.

This being public data, I will ensure that my colleagues can use this data without having to wait on me:

gsutil acl ch -R -g google.com:R \

gs://cloud-training-demos-ml/flights/raw/

That changes the access control list (acl) recursively (-R), applying to the group google.com read permission (:R) on everything starting from the Cloud Storage URL supplied. Had there been sensitive information in the dataset, I would have to be more careful, of course. We’ll discuss fine-grained security, by providing views with different columns to different roles in my organization, when we talk about putting the data in BigQuery. We’ll also discuss information leakage when information about the flights that people take can leak from the predictions we make on their behalf and how to guard against this when we talk about machine learning predictions.

Loading Data into Google BigQuery

On Google Cloud, the best place for structured and semi-structured data is BigQuery, a serverless data warehouse and SQL engine.

Advantages of a Serverless Columnar Database

Most relational database systems, whether commercial or open source, are row oriented in that the data is stored row by row. This makes it easy to append new rows of data to the database and allows for features such as row-level locking when updating the value of a row. The drawback is that queries that involve table scans (i.e., any aggregation that requires reading every row) can be expensive. Indexing counteracts this expense by creating a lookup table to map rows to column values, so that SELECTs that involve indexed columns do not have to load unnecessary rows from storage into memory. If you can rely on indexes for fast lookup of your data, a traditional Relational Database Management System (RDBMS) works well. For example, if your queries tend to come from software applications, you will know the queries that will come in and can create the appropriate indexes beforehand. This is not an option for use cases like business intelligence for which human users are writing ad hoc queries; therefore, a different architecture is needed.

BigQuery, on the other hand, is a columnar database—data is stored column by column and each column’s data is stored in a highly efficient compressed format that enables fast querying. Because of the way data is stored, many common queries can be carried out such that the query processing time is linear on the size of the relevant data. For applications such as data warehousing and business intelligence for which the predominant operations are read-only SELECT queries requiring full table scans, columnar databases are a better fit. BigQuery, for example, can scan terabytes of data in a matter of seconds. The trade-off is that INSERT, UPDATE, and DELETE, although possible in BigQuery, are significantly more expensive to process than SELECT statements. BigQuery is tuned toward analytics use cases.

BigQuery is serverless, so you don’t actually spin up a BigQuery server in your project. Instead, you simply submit a SQL query, and it is executed on the cloud. Queries that you submit to BigQuery are executed on a large number of compute nodes (called slots) in parallel. These slots do not need to be statically allocated beforehand—instead, they are “always on” available on demand, and scale to the size of your job. Because data is in situ and not sharded (not broken into small chunks that are attached to individual compute instances), the total power of the datacenter can be brought to bear on the problem. Because these resources are elastic and used only for the duration of the query, BigQuery is more powerful and less expensive than a statically preallocated cluster because preallocated clusters will typically be provisioned for the average use case—BigQuery can bring more resources to bear on the above-average computational jobs and utilize fewer resources for below-average ones.

In addition, because you don’t need to reserve any compute resources for your data when you are not querying your data, it is extremely cost effective to just keep your data in BigQuery (you’ll pay for storage, but storage is inexpensive). Whenever you do need to query the data, the data is immediately available—you can query it without the need to start project-specific compute resources. This on-demand, autoscaling of compute resources is incredibly liberating.

BIGQUERY PRICING

If an on-demand cost structure (you pay per query) makes you concerned about costs that can fluctuate month over month, you can specify a billing cap beyond which users will not be able to go. For even more cost predictability, it is possible to pay a fixed monthly price for BigQuery—flat-rate pricing allows you to get a predictable cost regardless of the number of queries run or data processed by those queries—the fixed monthly price essentially buys you access to a specific number of slots.15 In short, BigQuery has two pricing models for analysis: an on-demand pricing model in which your cost depends on the quantity of data processed, and a flat-rate model in which you pay a fixed amount per month for an unlimited number of queries that will run on a specific set of compute resources. You can augment either with flex slots, that are paid for by the minute. In all these cases, storage is a separate cost and depends on data size.

In summary, BigQuery is a columnar database, making it particularly effective for read-only queries that process all of the data. Because it is serverless, can autoscale to thousands of compute nodes, and doesn’t require clusters to be preallocated, it is also very powerful and quite inexpensive.

Staging on Cloud Storage

Although it is possible to ingest files from on-premises hardware directly into BigQuery using the bq command-line tool that comes with the Google Cloud SDK ( gcloud ), you should use that capability only for small datasets. To retrieve data from outside Google Cloud Platform to BigQuery, it is preferable to first load it into Cloud Storage and use Cloud Storage as the staging ground for BigQuery, as demonstrated in Figure 2-13.

Figure 2-13. Use Cloud Storage as a staging ground to ingest data into BigQuery

For larger files, it is better to ingest the files into Cloud Storage using gsutil first because gsutil takes advantage of multithreaded, resumable uploads and is better suited to the public internet. Fortunately, we already have the flights CSV files in Cloud Storage!

When should you save your data in Cloud Storage, and when should you store it in BigQuery? First, if the data is not tabular-like (that is: images, videos, and other arbitrary file types), then Google Cloud Storage (GCS) is the right choice. For tabular-like data, the answer boils down to what you want to do with the data and the kinds of analyses you want to perform. If you’ll mostly be running custom code that expects to read plain files, or your analysis involves reading the entire dataset, use Cloud Storage. On the other hand, if your desired access pattern is to run interactive SQL queries on the data, store your data in BigQuery. In the pre-cloud world, if you would use flat files, use Cloud Storage. If you’d put the data in a database, put it in BigQuery.

Access Control

The first step to ingest data into BigQuery is to create a BigQuery dataset – a dataset is a container for tables. You can have multiple datasets within a project. Go to the web console at https://console.cloud.google.com/bigquery and choose the Create Dataset option. Then, create a dataset called dsongcp.

You can also do this from the command-line:

bq mk dsongcp

Datasets in BigQuery are mostly just an organizational convenience—tables are where data resides and it is the columns of the table that dictate the queries we write. Besides providing a way to organize tables, though, datasets also serve as the access control point. You can provide view or edit access only at the project or dataset level, not at the table level. Cloud Identity Access Management (Cloud IAM) on Google Cloud Platform provides a mechanism to control who can carry out what actions on which resource (Figure 2-14).

The “who” can be specified in terms of an individual user (identified by his Google account such as a gmail.com address or company email address if the company is a GSuite customer), a Google Group (i.e., all current members of the group), or a GSuite domain (anyone with a Google account in that domain). Google groups and GSuite domains provide a convenient mechanism for aggregating a number of users and providing similar access to all of them.

Figure 2-14. Cloud IAM provides a mechanism to control access to resources

In addition, different logical parts of an application can be assigned separate identities (linked to email addresses) called service accounts. Service accounts are a very powerful concept because they allow different parts of a codebase to have permissions that are independent of the access level of the person running that application. For example, you might want an application to be able to query a table but not delete it even if the developer who created the application and the person running the application have that authority.

You should use service accounts with care for scenarios in which audit records are mandatory. Providing access at the Google Groups level provides more of an audit trail: because Google groups don’t have login credentials (only individual users do), the user who made a request or action is always recorded, even if their access is provided at the level of a Google Group or GSuite domain. However, service accounts are themselves login credentials, and so the audit trail turns cold if you provide access to service accounts—you will no longer know which user initiated the application request unless that application in turn logs this information. This is something to keep in mind when granting access to service accounts. Try to avoid providing service account access to resources that require auditability. If you do provide service account access, you should ensure that the application to which you have provided access itself provides the necessary audit trail by keeping track of the user on behalf of whom it is executing the request. The same considerations apply to service accounts that are part of Google Groups or domains. Because audit trails go cold with service accounts,16 you should restrict Google Groups and GSuite domains to only human users and service accounts that belong to applications that provide any necessary legal auditability guarantees.

Creating single-user projects is another way to ensure that service accounts map cleanly to users, but this can lead to significant administrative overhead associated with shared resources and departing personnel. Essentially, you would create a project that is billed to the same company billing account, but each individual user would have her own project in which she works. You can use the gcloud command to script the creation of such single-user projects in which the user in question is an editor (not an owner).17

In addition to specific users, groups, domains, and service accounts, there are two wildcard options available. Access can be provided to allAuthenticatedUsers, in which case anyone authenticated with either a Google account or a service account is provided access. Because allAuthenticatedUsers includes service accounts, it should not be used for resources for which a clear audit trail is required. The other wildcard option is to provide access to allUsers, in which case anyone on the internet has access—a common use case for this is to provide highly available static web resources by storing them on Cloud Storage. Be careful about doing this indiscriminately—egress of data from Google Cloud Platform is not free, so you will pay for the bandwidth consumed by the download of your cloud-hosted datasets.

The “what” actions depend on the resource access which is being controlled. The resources themselves fall into a policy hierarchy.

Policies can be specified at an organization level (i.e., to all projects in the organization), at the project level (i.e., to all resources in the project), or at the resource level (i.e., to a Compute Engine instance or a BigQuery dataset). As Figure 2-15 shows, policies specified at higher levels are inherited at lower levels, and the policy in effect is the union of all the permissions granted—there is no way to restrict some access to a dataset to a user who has that access inherited from the project level. Moving a project from one organization to another automatically updates that project’s Cloud IAM policy and ends up affecting all the resources owned by that project.

Figure 2-15. Policies specified at higher levels are inherited at lower levels

What type of actions can be carried out depends on the resource in question. Before Cloud IAM was introduced on the Google Cloud Platform, there were only three roles: owner, editor, and viewer/reader for all resources. Cloud IAM brought with it much finer-grained roles, but the original three roles were grandfathered in as primitive roles. Table 2-4 lists some of the roles that are possible for BigQuery datasets.

Table 2-4. Table 2-4. Some of the available roles in BigQuery

Role

Capabilities

Inherits from

Project Viewer

Execute a query

List datasets

Project Editor

Create a new dataset

Project Viewer

Project Owner

List/delete datasets

View jobs run by other project users

Project Editor

bigquery.user

Execute a query

List datasets

bigquery.dataViewer

Read, query, copy, export tables in the dataset

bigquery.dataEditor

Append, load data into tables in the dataset

Project Editor

bigquery.dataViewer

bigquery.dataOwner

Update, delete on tables in dataset

Project Owner

bigquery.dataEditor

bigquery.admin

All

Ingesting CSV Files

We can load the data directly into BigQuery’s native storage using the command line utility bq that comes with the gcloud SDK :

bq load --autodetect --source_format=CSV \

--skip_leading_rows=1 dsongcp.flights_auto \

gs://${BUCKET}/flights/raw/201501.csv

Here, we are asking BigQuery to autodetect the schema from the CSV file and loading the January data into a table named flights_auto (if you are following along with me, make sure to change the bucket to reflect the bucket that your files are in).

If you now go to the BigQuery web console ( https://console.cloud.google.com/bigquery ) and examine the dataset dsongcp, you will see that there is a table named flights_auto in it. You can examine the autodetected schema and preview the contents of the table.

We can try querying the data to find the average departure and arrival delays at the busiest airports:

SELECT

ORIGIN,

AVG(DEPDELAY) AS dep_delay,

AVG(ARRDELAY) AS arr_delay,

COUNT(ARRDELAY) AS num_flights

FROM

dsongcp.flights_auto

GROUP BY

ORIGIN

ORDER BY num_flights DESC

LIMIT 10

The result (see Table 2-5) starts with Atlanta (ATL), Dallas (DFW), and Chicago (ORD) which is what we would expect.

Table 2-5. Table 2-4. Average arrival and departure delays in January 2015 at the busiest 10 airports

Row

ORIGIN

dep_delay

arr_delay

num_flights

1

ATL

7.265885087329549

1.0802479706819135

29197

2

DFW

11.761812240572308

9.37162730937924

22571

3

ORD

19.96205128205128

17.016131923283645

22316

4

LAX

7.476340878516738

5.542057719380547

17048

5

DEN

15.506798076352176

11.842324888226543

16775

6

IAH

9.07378596782721

5.353498597528596

13191

7

PHX

8.066722908198505

6.197786998616902

13014

8

SFO

10.328127477406069

9.038424821002382

12570

9

LAS

8.566096692995435

5.0543525523958595

11499

10

MCO

9.887440638577354

5.820512820512793

9867

Auto detection is hit-and-miss, though. This is because the way it works is that BigQuery samples about a hundred rows of data in order to determine what the data type needs to be. If the arrival delay was an integer for all 100 rows that it saw, but turns out to be a string (NA) somewhere else in the file, the loading will fail. Autodetection may also fail if many of the fields are empty.

Partitioning

Because of this, autodetection is okay during initial exploration, but we should quickly pivot to actually specifying the schema. At that time, it may be worthwhile to also consider whether this table should be partitioned by date – if most of our queries will be, not on the full table, but only a few days, then partitioning will lead to cost savings. If that were the case, we would create the table first, specifying that it should be partitioned by date:

bq mk --time_partitioning_type=DAY dsongcp.flights_auto

When loading the data, we’d need to load each partition separately (partitions are named flights_auto$20150101 , for example). We can also partition by a column in the data (FlightsDate, for example).

Currently, we don’t know much about the fields, so we can ask BigQuery to treat all the columns except the FlightDate as a string:

SCHEMA=Year:STRING,...,FlightDate:DATE,Reporting_Airline:STRING,...

Putting all these together, the loading becomes (see bqload.sh in the course repository):

for MONTH in `seq -w 1 12`; do

CSVFILE=gs://${BUCKET}/flights/raw/${YEAR}${MONTH}.csv

bq --project_id $PROJECT \

load --time_partitioning_field=FlightDate \

--time_partitioning_type=MONTH \

--source_format=CSV --ignore_unknown_values \

--skip_leading_rows=1 --schema=$SCHEMA \

${PROJECT}:dsongcp.flights_raw\$${YEAR}${MONTH} $CSVFILE

done

At this point, we have the CSV files in Cloud Storage and the raw data in BigQuery. We have successfully ingested the 2015 flights data into GCP! If you want, you can repeat this for years 2016-201918 by changing the for loop in ingest.sh to:

for YEAR in `seq 2016 2019`; do

However, it is not necessary to do so – all the code in this book will work fine with just 2015 data. In Chapter 3, we will start to look at the 2015 data and do useful things with it.

But before we move on, let’s digress a little and consider automation.

Scheduling Monthly Downloads

Now that we have some historical flight data in our Cloud Storage bucket, it is natural to wonder how to keep the bucket current. After all, airlines didn’t stop flying in 2021, and the BTS continues to refresh its website on a monthly basis. It would be good if we could schedule monthly downloads to keep ourselves synchronized with the BTS.

There are two scenarios to consider here. The BTS could let us know when it has new data, and we could then proceed to ingest the data. The other option is that we periodically monitor the BTS website and ingest new data as it becomes available. The BTS doesn’t offer a mechanism by which we can be notified about data updates, so we will need to resort to polling. We can, of course, be smart about how we do the polling. For example, if the BTS tends to update its website around the 5th of every month, we could poll at that time.

Where should this ingest program be executed? Realizing that this is a program that will be invoked only once a month (more often if retries are needed if an ingest fails), we realize that this is not a long-running job, but is instead something that should be scheduled to run periodically. The traditional way to do this is to schedule a cron 19 job in Unix/Linux. To schedule a cron job, you add a line to a crontab 20 file and then register it with a Unix daemon that takes care of the scheduling. For example, adding this line

1 2 10 * * /etc/bin/ingest_flights.py

to crontab will cause the Python program /etc/bin/ingest_flights.py (that would carry out the same steps to ingest the flights data that we did on the command line in the previous section) to be run by the system at 02:01 on the 10th of every month.

Although cron jobs are a straightforward solution, there are several problems that all come down to resilience and repeatability:

- The cron job is scheduled on a particular server. If that server happens to be rebooted around 2 AM on April 10, the ingest might never take place that month.

- The environment that cron executes in is very restricted. Our task will need to download data from BTS, uncompress it, clean it up, and upload it to the cloud. These impose a variety of requirements in terms of memory, space, and permissions and it can be difficult to configure cron appropriately. In practice, system administrators configure cron jobs on particular machines, and find it difficult to port them to other machines that do not have the same system paths.

- If the ingest job fails (if, for example, the network is down), there is no way to retry it. Retries and other such failure-recovery efforts will have to be explicitly coded in our Python program.

- Remote monitoring and one-time, ad hoc executions are not part of the cron interface. If you need to monitor, troubleshoot, and restart the ingest from a mobile device, good luck.

This litany of drawbacks is not unique to cron. They are implicit in any solution that is tied to specific servers. So, how would you do it on the cloud? What you should not do is to create a Compute Engine VM and schedule a cron job on it—that will be subject to some of the same problems!

For resilience and reliability, we need a serverless way to schedule ingest jobs. Obviously, the ingest job will need to be run on some machine somewhere. However, we shouldn’t need to manage that machine at all. This is a job that needs perhaps two minutes of compute resources a month. We should be looking for a way to write the ingest code and let the cloud infrastructure take care of provisioning resources, making retries, and providing for remote monitoring and ad hoc execution.

On Google Cloud Platform, Cloud Scheduler provides a way to schedule periodic jobs in a serverless manner. These jobs can involve hitting a HTTP endpoint (which is what we will do), but can also send a message via Cloud Pub/Sub or trigger a Google Kubernetes Engine or Cloud Dataflow job. Figure 2-16 presents our architecture for the monthly ingest job.

Figure 2-16. The architecture of the monthly ingest job

First, we will write a standalone ingest_flights.py application that is capable of downloading the data for a specific year/month and uploading the data to Cloud Storage. We will invoke the ingest code from making sure to explicitly capture our dependencies in a Docker file. Cloud Run will run our container.21

The way scheduling works in Cloud Scheduler is that we must specify a URL that will be invoked or a Cloud Pub/Sub topic that must be monitored. Whereas in the previous Linux cron example we specified a script on the server that was running the cron daemon, the Cloud Scheduler endpoint will be a URL that will be visited according to the schedule that we specify (this can be any URL; it doesn’t need to be a service that we write). Because our ingest code is a standalone Python program, we will wrap that ingest code into a Python Flask application (main.py) so that we can invoke it by using a URL (Flask is a web application framework).

Ingesting in Python

While exploring the data, we carried out the ingest on the command line in Bash. We saved our commands as we went along in the form of Bash scripts. We created our ingest program by simply making a Bash script 22 that invokes those intermediate steps:

#!/bin/bash

for MONTH in `seq 1 12`; do

bash download.sh $YEAR $MONTH

done

# upload the raw CSV files to our GCS bucket

bash upload.sh

# load the CSV files into BigQuery as string columns

bash bqload.sh

This is the sort of decision that leads to spaghetti-like code that is difficult to unravel and to maintain. There are many assumptions made by this set of Bash scripts in terms of what to download, where the temporary storage resides, and where to upload it. Changing any of these will involve changing multiple scripts. Using Bash to quickly get a handle on the data is a good idea, as is the idea of saving these scripts so as to continue the exploration. But when it comes to making the ingest more systematic and routine, you do not want to use a shell scripting language; a more formal programming language is better.

In this book, we will use Python wherever we can because of its ability to span a wide range of computing tasks, from systems programming to statistics and machine learning. Python is currently the best choice if you need to pick a single language in which to do most of your work. Java is typesafe and performant. Its object-orientation and packaging architecture are suitable for large, multideveloper programs, but it makes the code too verbose. Moreover, the lack of a Read-Evaluate-Process-Loop (REPL) interpreter makes Java unwieldy for quick experimentation. C++ is numerically very efficient but standard libraries for non-numerical computing are often nonexistent. Given the choice of a data package available in Python, Scala, R, and Java, significant majorities choose Python and Scala over R and Java.23 Scala combines the benefits of Python (easy scriptability, conciseness) with the benefits of Java (type-safety, speed), but the tooling for Scala (such as for statistics and visualization) is not as pervasive as it is for Python. Today, therefore, the best choice of programming language is Python. For certain use cases for which speed is important and Python is not performant enough, it might be necessary to use Java.

The ingest program in Python24 goes through the same four steps as before when we did it manually on the command line:

1. Download data from the BTS website to a local file.

2. Unzip the downloaded ZIP file and extract the CSV file it contains.

3. Upload the CSV file to Google Cloud Storage.

4. Load the CSV data into a BigQuery partitioned table

Whereas our download Bash script got all 12 months of a hardcoded year (2015), our download subroutine in Python will take as input the year and month:

def download(YEAR, MONTH, destdir):

'''

Downloads on-time performance data and returns local filename

YEAR e.g.'2015'

MONTH e.g. '01' for January

'''

url = os.path.join("https://transtats.bts.gov/PREZIP",

"_{}_{}.zip".format(YEAR, int(MONTH)))

filename = os.path.join(destdir, "{}{}.zip".format(YEAR, MONTH))

with open(filename, "wb") as fp:

response = urlopen(url)

fp.write(response.read())

return filename

Another thing to note is that our Bash script simply downloaded the ZIP file from BTS to the current working directory of the user. However, since our Python script is meant to be executed on demand by the scheduler service, we cannot make assumptions about the directory in which the script will be run. In particular, we don’t know whether that directory will be writeable and have enough space. Hence, we ask the caller of the function to provide an appropriate destination directory in which to store the downloaded ZIP file.

Here’s how to unzip the file and extract the CSV contents:

def zip_to_csv(filename, destdir):

zip_ref = zipfile.ZipFile(filename, 'r')

cwd = os.getcwd()

os.chdir(destdir)

zip_ref.extractall()

os.chdir(cwd)

csvfile = os.path.join(destdir, zip_ref.namelist()[0])

zip_ref.close()

Unzipping explodes the size of the file. We can optimize things slightly. Rather than upload the text file, we can gzip it since BigQuery knows how to load gzipped CSV files:

gzipped = csvfile + ".gz"

with open(csvfile, 'rb') as ifp:

with gzip.open(gzipped, 'wb') as ofp:

shutil.copyfileobj(ifp, ofp)

return gzipped

Annoyingly, the zipfile module in Python doesn’t provide a way to extract contents to a specific directory—it insists on extracting the contents in the current working directory. So, we make sure to change to the destination directory before doing the extraction. Then, we change back.

Here’s the code to upload the CSV file for a given month to Cloud Storage:

def upload(csvfile, bucketname, blobname):

client = storage.Client()

bucket = client.get_bucket(bucketname)

blob = Blob(blobname, bucket)

blob.upload_from_filename(csvfile)

gcslocation = 'gs://{}/{}'.format(bucketname, blobname)

print ('Uploaded {} ...'.format(gcslocation))

return gcslocation

The code asks for the bucketname (the single-region bucket that was created during our exploration) and a blobname (e.g., flights/201501.csv ) and carries out the upload using the Cloud Storage Python library. Although it can be tempting to simply use the subprocess module in Python to invoke gsutil operations, it is better not to do so. If you go the subprocess route, you will then need to ensure that the Cloud SDK (that gsutil comes with) is installed on whichever machine this is going to run on. This won’t be a problem in Cloud Run, but might pose problems if you switch the way you provide URL access later (to, say, Google App Engine or Cloud Functions). It is preferable to use pure Python modules when possible and add those modules to requirements.txt , as follows:

Flask

google-cloud-storage

google-cloud-bigquery

gunicorn==20.1.0

The Flask library will help us handle HTTP requests (covered shortly) and Google Cloud Storage is needed so as to invoke the get_bucket() and upload_from_filename() operations. While using the latest version of libraries is okay, it poses the problem that an upgrade to those dependencies might break our code. For production code, it is better to pin the library versions to the ones with which the code has been tested:

Flask==2.0.1

google-cloud-storage==1.42.0

google-cloud-bigquery==2.25.1

gunicorn==20.1.0

If you do pin libraries, though, you will have to have a process in place to periodically test and upgrade to the latest stable version of your dependencies. Otherwise, your code might go stale or, worse, using library versions with known vulnerabilities.

We can now write an ingest() method that calls the four major steps, plus the verification, in order:

def ingest(year, month, bucket):

'''

ingest flights data from BTS website to Google Cloud Storage

return cloud-storage-blob-name on success.

raises DataUnavailable if this data is not on BTS website

'''

tempdir = tempfile.mkdtemp(prefix='ingest_flights')

try:

zipfile = download(year, month, tempdir)

bts_csv = zip_to_csv(zipfile, tempdir)

gcsloc = 'flights/raw/{}{}.csv.gz'.format(year, month)

gcsloc = upload(bts_csv, bucket, gcsloc)

return bqload(gcsloc, year, month)

finally:

print ('Cleaning up by removing {}'.format(tempdir))

shutil.rmtree(tempdir)

The destination directory that we use to stage the downloaded data before uploading to Cloud Storage is obtained using the tempfile package in Python. This ensures that if, for whatever reason, there are two instances of this program running at the same time, they will not cause contention issues.

We can try out the code by writing a main() that is executed if this program25 is run on the command line:

if __name__ == '__main__':

import argparse

parser = argparse.ArgumentParser(

description='ingest flights data from BTS website to GCS')

parser.add_argument('--bucket', help='GCS bucket to upload data to',

required=True)

parser.add_argument('--year', help='Example: 2015.', required=True)

parser.add_argument('--month', help='01 for Jan.', required=True)

try:

args = parser.parse_args()

gcsfile = ingest(args.year, args.month, args.bucket)

print ('Success ... ingested to {}'.format(gcsfile))

except DataUnavailable as e:

print ('Try again later: {}'.format(e.message))

Specifying a valid month ends with a new (or replaced) file on Cloud Storage:

$ ./ingest_flights.py  --bucket cloud-training-demos-ml \

--year 2015 --month 01

...

Success ... ingested to gs://cloud-training-demos-ml/flights/201501.csv

Trying to download a month that is not yet available results in an error message:

$ ./ingest_flights.py --bucket cloud-training-demos-ml \

--year 2029 --month 01

...

HTTP Error 403: Forbidden

On Cloud Scheduler, this will result in the call failing and being retried subject to a maximum number of retries. Retries will also happen if the BTS web server cannot be reached.

At this point, we have the equivalent of our exploratory Bash scripts, but with some additional resilience, repeatability, and fault tolerance built in. Our Python program expects us to provide a year, month, and bucket. However, if we are doing monthly ingests, we already know which year and month we need to ingest. No, not the current month—recall that there is a time lag between the flight events and the data being reported by the carriers to the BTS. Instead, it is simply the month after whatever files we already have on Cloud Storage! So, we can automate this, too:

def next_month(bucketname):

'''

Finds which months are on GCS, and returns next year,month to download

'''

client = storage.Client()

bucket = client.get_bucket(bucketname)

blobs = list(bucket.list_blobs(prefix='flights/raw/'))

files = [blob.name for blob in blobs if 'csv' in blob.name] # csv files only

lastfile = os.path.basename(files[-1]) # e.g. 201503.csv

year = lastfile[:4]

month = lastfile[4:6]

dt = datetime.datetime(int(year), int(month), 15) # 15th of month

dt = dt + datetime.timedelta(30) # will always go to next month

return '{}'.format(dt.year), '{:02d}'.format(dt.month)

To get the next month given that there is a file, say 201503.csv , on Cloud Storage, we add 30 days to the Ides of March—this gets around the fact that there can be anywhere from 28 days to 31 days in a month, and that timedelta requires a precise number of days to add to a date.

By changing the year and month to be optional parameters, we can try out the ingest program’s ability to find the next month and ingest it to Cloud Storage. We simply add:

if args.year is None or args.month is None:

year, month = next_month(args.bucket)

else:

year = args.year

month = args.month

gcsfile = ingest(year, month, args.bucket)

Now that we have an ingest program that is capable of updating our Cloud Storage bucket one month at a time, we can move on to building the scaffolding to have it be executed in a serverless way.

Cloud Run

Cloud Run is a serverless framework that provides an autoscaling, resilient runtime for containerized code. The container (see Figure 2-16) will consist of code that listens for requests or events. Cloud Run abstracts away all the infrastructure management that would otherwise be needed.

Now that we have a Python function that will do the ingest, we will wrap it inside a web application. To write the web application, we will use Flask which is a lightweight Python web application framework and as a web server, we will use gunicorn. Flask provides the ability to invoke Python code in response to a HTTP request while gunicorn will listen to HTTP requests and send them to the Flask app. Our container will consist of the gunicorn server, Flask application, and its dependencies. This is expressed in the form of a Dockerfile:

FROM python:3.6-slim

# Copy local code to the container image.

ENV APP_HOME /app

WORKDIR $APP_HOME

COPY . ./

# Install production dependencies.

RUN pip install --no-cache-dir -r requirements.txt

# Run the web service on container startup.

# Timeout is set to 0 to disable the timeouts of

# the workers to allow Cloud Run to handle instance scaling.

CMD exec gunicorn --bind :$PORT --workers 1 \

--threads 8 --timeout 0 main:app

In our main.py, we have a function that gets invoked in response to the URL trigger:

import logging

from flask import escape

from ingest_flights import *

app = Flask(__name__)

@app.route("/", methods=['POST'])

def ingest_flights(request):

try:

json = request.get_json()

year = escape(json['year']) if 'year' in json else None

month = escape(json['month']) if 'month' in json else None

bucket = escape(json['bucket']) # required

if year is None or month is None or len(year) == 0 or len(month) == 0:

year, month = next_month(bucket)

tableref, numrows = ingest(year, month, bucket)

ok = 'Success ... ingested {} rows to {}'.format(numrows, tableref)

return ok

except Exception as e:

logging.exception('Try again later')

Essentially, the main.py has a single function that receives a Flask request object, from which we can extract the JSON payload of the HTTP POST by which the Cloud Run will be triggered. We get the next month by looking to see what months are already in the bucket and then ingest the necessary data using the existing code in the module ingest_flights . We can deploy the our code base as a container to Cloud Run using:

NAME=ingest-flights-monthly

REGION=us-central1

gcloud run deploy $NAME --region $REGION --source=$(pwd) \

--platform=managed --timeout 12m

But there are a couple of serious security and governance problems if we do this.

Securing Cloud Run

What are the security problems?

1. Anyone can invoke the URL and cause our dataset to get updated. We have to disallow unauthenticated users.

2. Allowing this code to run with our user account’s permissions will pollute any audit logs since we are not actually running the ingest interactively. We need to create a separate account so that the Cloud Run service can run with that identity.

3. Allowing this code to run with our user account’s permissions is also quite dangerous because our user account will typically have very broad permissions. We’d like to restrict the tasks that this automated service can do: we want it to be able to write only to specific Cloud Storage buckets and BigQuery tables.

The way to address the first point is to disallow unauthenticated users. The way to accomplish the second requirement is to specify that the Cloud Run service will have to run as a service account. A service account is an account whose identity is meant to be taken on by automated services. Like any identity, it can be configured to have specific and limited permissions. Therefore, before we can deploy the Cloud Run service, we will need to create a service account. Service accounts have email addresses of the form svc-monthly-ingest@cloud-training-demos.iam.gserviceaccount.com.

You can create a service account by going to the web console in the IAM area, but as usual, I prefer to script things:26

SVC_ACCT=svc-monthly-ingest

PROJECT_ID=$(gcloud config get-value project)

BUCKET=${PROJECT_ID}-cf-staging

REGION=us-central1

SVC_PRINCIPAL=serviceAccount:${SVC_ACCT}@${PROJECT_ID}.iam.gserviceaccount.com

gcloud iam service-accounts create $SVC_ACCT \

--display-name "flights monthly ingest"

Then, we make the service account the admin of the staging GCS bucket so that it can read, write, list, delete, etc. on this bucket (and only this bucket):

gsutil mb -l $REGION gs://$BUCKET

gsutil uniformbucketlevelaccess set on gs://$BUCKET

gsutil iam ch ${SVC_PRINCIPAL}:roles/storage.admin gs://$BUCKET

We will also allow the service account to create and delete partitions on tables in just the BigQuery dataset dsongcp (and no other datasets):

bq --project_id=${PROJECT_ID} query --nouse_legacy_sql \

"GRANT \`roles/bigquery.dataOwner\` ON SCHEMA dsongcp TO '$SVC_PRINCIPAL' "

gcloud projects add-iam-policy-binding ${PROJECT_ID} \

--member ${SVC_PRINCIPAL} \

--role roles/bigquery.jobUser

Are these permissions enough? One way is to try to ingest a month of data when running as this service account. To do so, we will have to impersonate the service account: 27

1. Visit the Service Accounts section of the GCP Console

2. Select the newly created service account svc-monthly-ingest and click Manage Keys

3. Add key (Create a new JSON key) and download it to a file named tempkey.json. Transfer this key file to your Cloud Shell instance.

4. Run

gcloud auth activate-service-account \

--key-file tempkey.json

5. Try ingesting one month

./ingest_flights.py --bucket $BUCKET \

--year 2015 --month 03 --debug

Once you have ensured that the service account has all the necessary permissions, go back to running commands as yourself using gcloud auth login.

Deploying and Invoking Cloud Run

Now that we have the code for the Flask application and a service account with the right permissions, we can deploy the code to Cloud Run to run as this service account: 28

NAME=ingest-flights-monthly

SVC_ACCT=svc-monthly-ingest

PROJECT_ID=$(gcloud config get-value project)

REGION=us-central1

SVC_EMAIL=${SVC_ACCT}@${PROJECT_ID}.iam.gserviceaccount.com

gcloud run deploy $NAME --region $REGION --source=$(pwd) \

--platform=managed --service-account ${SVC_EMAIL}  \

--no-allow-unauthenticated --timeout 12m

Recall that we started the discussion on securing the Cloud Run instance by saying that we would disallow unauthenticated users and have the Cloud Run service run as a service account. Note how we are turning on both these options when we deploy to Cloud Run.

Once the application has been deployed to Cloud Run, we can try accessing the URL of the service with our authentication details in the header of the web request and a JSON message as its POST: 29

# Feb 2015

echo {\"year\":\"2015\"\,\"month\":\"02\"\,\"bucket\":\"${BUCKET}\"\}\

> /tmp/message

curl -k -X POST $URL \

-H "Authorization: Bearer $(gcloud auth print-identity-token)" \

-H "Content-Type:application/json" --data-binary @/tmp/message

But what is the URL? Cloud Run generates the URL when we deploy the container, and we can obtain it using:

gcloud run services describe ingest-flights-monthly \

--format 'value(status.url)')

Changing the message to provide only the bucket (no year or month) will make the service get the next month:

echo {\"bucket\":\"${BUCKET}\"\} > /tmp/message

curl -k -X POST $URL \

-H "Authorization: Bearer $(gcloud auth print-identity-token)" \

-H "Content-Type:application/json" --data-binary @/tmp/message

Scheduling Cloud Run

Our intent is to automatically invoke CloudRun once a month. We can do that using Cloud Scheduler, which is also serverless and doesn’t require us to manage any infrastructure. We simply specify the schedule and the URL to hit. This URL is what came from the output of the Cloud Run deployment command in the previous section.

echo {\"bucket\":\"${BUCKET}\"\} > /tmp/message

cat /tmp/message

gcloud scheduler jobs create http monthlyupdate \

--description "Ingest flights using Cloud Run" \

--schedule="8 of month 10:00" \

--time-zone "America/New_York" \

--uri=$SVC_URL --http-method POST \

--oidc-service-account-email $SVC_EMAIL \

--oidc-token-audience=$SVC_URL \

--max-backoff=7d \

--max-retry-attempts=5 \

--max-retry-duration=2d \

--min-backoff=12h \

--headers="Content-Type=application/json" \

--message-body-from-file=/tmp/message

The preceding parameters would make the first retry happen after 12 hours. Subsequent retries are increasingly farther apart, up to a maximum of 2 days between attempts. We fail the task permanently if it fails five times within a defined time period and the task is more than 7 days old (both limits must be passed for the task to fail).

To try out the Cloud Scheduler, we could wait for the 8th of the month to roll around. Or we could go to the GCP web console and click on “Run Now”. Unfortunately, it won’t work because Cloud Scheduler wants to run as the service account while you are logged in as yourself. So, give yourself the ability to impersonate the service account by going to the Service Accounts part of the web console. Once you’ve done that, you will be able to get “Run Now” to work.

The monthly update mechanism works if you have the previous month’s data on Cloud Storage. If you start out with only 2015 data, updating it monthly means that you will inevitably be many months behind. So, you will need to run it ad hoc until your data is up to date and then let the Cron service take care of things after that. Alternatively, you can take advantage of the fact that the ingest task is cheap and non-intrusive when there is no new data. So, you can change the schedule to be every day instead of every month. A better solution is to change the ingest task so that if it is successful in ingesting a new month of data, it immediately tries to ingest the next month. This way, your program will crawl month-by-month to the latest available month and then keep itself always up-to-date.

At this point, it is worth reflecting a bit on what we have accomplished. We are able to ingest data and keep it up-to-date by doing just these steps:

1. Write some Python code.

2. Deploy that Python code to the Google Cloud Platform.

We did not need to manage any infrastructure in order to do this. We didn’t install any OS, manage accounts on those machines, keep them up to date with security patches, maintain failover systems, and so on—a serverless solution that consists simply of deploying code to the cloud is incredibly liberating. Not only is our ingest convenient, it is also very inexpensive--everything scales down to zero when it is not being used. All this falls comfortably within the free tier or might cost less than 5¢ a month.

Summary

The US BTS collects, and makes publicly available, a dataset of flight information. It includes nearly a hundred fields, including scheduled and actual departure and arrival times, origin and destination airports, and flight numbers of every domestic flight scheduled by every major carrier. We will use this dataset to estimate the likelihood of an arrival delay of more than 15 minutes of the specific flight whose outcome needs to be known in order for us to decide whether to cancel the meeting.

There are three possible data processing architectures on the cloud for large datasets: scaling up, scaling out with sharded data, and scaling out with data in situ. Scaling up is very efficient, but is limited by the size of the largest machine you can get a hold of. Scaling out is very popular but requires that you preshard your data by splitting it among compute nodes, which leads to maintaining expensive clusters unless you can hit sustained high utilization. Keeping data in situ is possible only if your datacenter supports petabits per second of bisectional bandwidth so that any file can be moved to any compute node in the datacenter on demand. Because Google Cloud Platform has this capability, we will upload our data to Google Cloud Storage, a blob storage that is not presharded and to BigQuery which will allow us to carry out interactive exploration on large datasets.

To automate the ingest of the files, we reverse engineered the BTS’s web form and obtained the format of the POST request that we need to make. With that request in hand, we were able to write a Bash script to pull down 12 months of data, uncompress the ZIP file, and load the data into BigQuery. It is quite straightforward to change this script to loop through multiple years.

We discussed the difference between strong consistency and eventual consistency and how to make the appropriate trade-off imposed by Brewer’s CAP theorem. In this case, we wanted strong consistency and did not need global availability. Hence, we chose to use a single-region bucket. We then uploaded the downloaded, unzipped, and cleaned CSV files to Google Cloud Storage.

To schedule monthly downloads of the BTS dataset, we made our download and cleanup Python program and made it callable from Cloud Run so that it was completely serverless. We used Cloud Scheduler to periodically request the Cloud Run application to download BTS data, unzip it, and upload it to both Cloud Storage and BigQuery.

Code Break

This is the point at which you should put this book off to the side and attempt to repeat all the things I’ve talked about. All the code snippets in this book have corresponding code in the GitHub repository.

I strongly encourage you to play around with the code in 02_ingest with the goal of understanding why it is organized that way and being able to write similar code yourself. At minimum, though, you should do the following:

1. Open Cloud Shell and git clone the book’s code repository as explained in Chapter 1.

2. Go to the 02_ingest folder of the repository.

3. Go to the Storage section of the GCP web console and create a new bucket.

4. Change the BUCKET variable in upload.sh to reflect the bucket that you created.

5. Run ./ingest.sh. This will populate your bucket and BigQuery dataset with data from 2015. Optionally, you can change the year loop in this file to download all the data corresponding to 2015-2019.

6. Because software changes, an up-to-date list of the preceding steps is available in the course repository in 02_ingest/README.md. This is true for all the following chapters.

1   See, for example, https://www.congress.gov/congressional-report/107th-congress/senate-report/13/1. The bill referenced in the report was not enacted into law, but it illustrates Congress’s monitoring function based on the statistics collected by the Department of Transportation.

2   For example, weather radar data from before 2000 had timestamps assigned by a radar engineer. Essentially, the engineers would look at their wristwatches and enter a time into the radar products generator. Naturally, this was subject to all kinds of human errors—dates could be many hours off. The underlying problem was fixed by the introduction of network clocks to ensure consistent times between all the radars on the US weather radar network. When using historical weather data, though, time correction is an important preprocessing step.

3   The shortest path between two points on the globe is an arc that passes through the two points and whose focus point is the center of the globe.

4   Indeed, scripting the field selection and download is what I did in the first edition of the book. If interested, see the select-and-download code in GitHub in the branch edition1_tf2 – the key thing is that the web request sends the selected fields inside the POST request and handles the resulting client-side redirect to obtain the zip file that is created on demand.

5   Over the last 5 years, I have observed that the BTS server that does this zip file creation is frequently down. Yes, they would ideally use a public Cloud to host their website, but you try telling the US government what to do.

6   Another thing I am doing is to host the zip files on Google Cloud, and have my code hit the Google Cloud server. The code by default will not hit the BTS server anymore. The original BTS URL is still present in the code, just commented out, so change it back if you want to try it out.

7   You could also just stop (and not delete) the Google Cloud Platform Compute Engine instance. Stopping the instance stops the bill associated with the compute machine, but you will continue to pay for storage. In particular, you will continue to pay for the SSD associated with the Compute Engine instance. The key advantage of a stopped instance is that you get to resume exactly where you left off, but this might not be important if you always start from a clean (known) state each time.

8   To shard a large database is to partition it into smaller, more easily managed parts. Whereas normalization of a database table places the columns of a database into different tables, sharding splits the rows of the database and uses different database server instances to handle each part. For more information, go to https://en.wikipedia.org/wiki/Shard_(database_architecture).

9   In other words, rather than a filesystem that is local to a single machine, it is common to the entire cluster of machines that form the datacenter.

10   The blog on Google’s networking infrastructure is worth a read. One petabit is 1 million gigabits, so the 1 Pbps quoted in the article works out to 125,000 GBps. Networking has only gotten better since 2015, of course.

11   A condition in which network packets need to be delivered in order; thus, a slow packet holds up delivery of later packets.

12   Microsoft Azure seems to involve a centralized host layer, for example, while AWS S3 seems to prioritize network latency. You’d design your software for such infrastructure differently.

13   See the script 02_ingest/download.sh in the course repository.

14   You can get your unique project ID from the Cloud Platform Console dashboard; it could be different from the common name that you assigned to your project. By default, Google Cloud Platform tries to give you a project ID that is the same as your project name, but if that name is already taken, you will get an autogenerated, unique project ID. Because of this default, you should be similarly careful about giving projects sensitive names.

15   See . As of this writing, switching to the flat-rate pricing model requires you to contact your Google Cloud Platform account representative.

16   A service account is tied to a project, but project membership evolves over time. So, even the subset of users who could have invoked the action might not be known unless you have strict governance over who is allowed to be an owner/editor/viewer of a project.

17   See for a gcloud script that will create single-user projects. The users will be editors on the project, but project ownership will reside with the person who has billing administrator rights.

18   Let’s ignore 2020 and 2021 because those were the years of the COVID pandemic.

19   A shortened form of a misspelling of chronos, Greek for time, cron is the name of the Unix daemon process that executes scheduled jobs at specific times.

20   Shortened form of cron table.

21   Docker containers are lightweight wrappers around a piece of software (here, the Flask endpoint main.py) that contain everything needed to run that software—code (e.g., ingest_flights.py), runtime (Python dependencies, etc.), configuration files and system libraries (here, a specific Linux distribution). Unlike a virtual machine, different containers running on the same machine can share layers of operating system dependencies.

22   02_ingest/ingest.sh

23   In 2016, Databricks found that 65% of survey respondents used Spark in Scala versus 62% in Python, 44% in SQL, 29% in Java and 20% in R. See the infographie and survey linked from https://databricks.com/blog/2016/09/27/spark-survey-2016-released.html..

24   See 02_ingest/monthlyupdate/ingest_flights.py

25   The full program is available as ingest_flights.py on GitHub at http://bit.ly/2BPhya4 —try it out.

26   See 02_ingest/monthlyupdate/01_setup_svc_acct.sh

27   See the instructions in README.md in 02_ingest.

28   See 02_ingest/monthlyupdate/02_deploy_cr.sh

29   See 02_ingest/monthlyupdate/03_call_cr.sh

Chapter 3. Creating Compelling Dashboards

A NOTE FOR EARLY RELEASE READERS

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 3rd chapter of the final book. Please note that the GitHub repo will be made active later on.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at mcronin@oreilly.com.

In Chapter 2, we ingested On-time Performance Data from the US Bureau of Transportation Statistics (BTS) so as to be able to model the arrival delay given various attributes of an airline flight—the purpose of the analysis is to cancel a meeting if the probability of the flight arriving within 15 minutes of the scheduled arrival time is less than 70%.

Before we delve into building statistical and machine learning models, it is important to explore the dataset and gain an intuitive understanding of the data—this is called exploratory data analysis , and it’s covered in more detail in Chapter 5. You should always carry out exploratory data analysis for any dataset that will be used as the basis for decision making. In this chapter, though, I talk about a different aspect of depicting data—of depicting data to end users and decision makers so that they can understand the recommendation that you are making. The audience of these visual representations, called dashboards , that we talk about in this chapter are not other data scientists, but are instead the end users. Keep the audience in mind as we go through this chapter, especially if you come from a data science background—the purpose of a dashboard is to explain an existing model, not to develop it. A dashboard is an end-user report that is interactive, tailored to end users, and continually refreshed with new data. See Table 3-1.

Table 3-1. A dashboard is different from exploratory data analysis

For Decision Makers

For Data scientists

Usage pattern

Dashboards

Exploratory data analysis

Kinds of depictions

Current status, pie charts, trendlines

Model fits with error bars, kernel density estimates

What does it explain?

Model recommendations and confidence

Input data, feature importance, model performance, etc.

Data represented

Subset of dataset, tailored to user’s context

Aggregate of historical data

Typical tools

Data Studio, Tableau, Qlik, Looker, etc.

Jupyter, python, R Studio, S-plus, matplotlib, seaborn, Matlab, etc.

Mode of interaction

GUI-driven

Code-driven

Update

Real time

Not real time

Covered in

Chapter 3, Chapter 4

Chapter 5

Example

From AAA fuel-gauge report

From AAA safety and educational foundation

Very often, this step of creating end-user visual depictions goes by the anodyne name of “visualization,” as in visualizing data. However, I have purposefully chosen not to call it by that name, because there is more to this than throwing together a few bar graphs and charts. Dashboards are highly visual, interactive reports that have been designed to depict data and explain models.

TIP

All of the code snippets in this chapter are available in the GitHub repository at https://github.com/GoogleCloudPlatform/data-science-on-gcp/ in the folder 03_sqlstudio. See the README.md file in that directory for instructions on how to do the steps described in this chapter.

Explain Your Model with Dashboards

The purpose of this step in the modeling process is not simply to depict the data, but to improve your users’ understanding of how the model behaves. Whenever you are designing the display of a dataset, evaluate the design in terms of three aspects:

1. Does it accurately and honestly depict the data? This is important when the raw data itself can be a basis for decision making.

2. How well does it help envision not just the raw data, but the information content embodied in the data? This is crucial for the cases when you are relying on human pattern recognition and interaction to help reveal insights about the environment in which the data was collected.

3. Is it constructed in such a way that it explains the model being used to provide recommendations?

You want to build displays that are always accurate and honest. At the same time, the displays need to be interactive so as to provide viewers with the ability to play with the data and gain insights. Insights that users have gained should be part of the display of that information going forward in such a way that those insights can be used to explain the data.

The last point, that of explanatory power, is very important. The idea is to disseminate data understanding throughout your company. A statistical or machine learning model that you build for your users will be considered a black box, and while you might get feedback on when it works well and when it doesn’t, you will rarely get pointed suggestions on how to actually improve that model in the field. In many cases, your users will use your model at a much more fine-grained level than you ever will, because they will use your model to make a single decision, whereas in both training and evaluation, you would have been looking at model performance as a whole.

Although this holistic overview is useful for statistical rigor, you need people taking a close look at individual cases, too. Because users are making decisions one at a time, they are analyzing the data one scenario at a time. If you provide your users not just with your recommendation, but with an explanation of why you are recommending it, they will begin to develop insights into your model. However, your users will only be able to develop such insights into the problem and your recommendations if you give them ways to observe the data that went into your model. Give enough users ways to view and interact with your data, and you will have unleashed a never-ending wave of innovation.

Your users have other activities that require their attention. Why would they spend their time looking at your data? One of the ways to entice them to do that is by making the depiction of the information compelling. In my experience, 1 the most compelling displays are displays of real-time information in context. You can show people the average airline delay at JFK on January 12, 2012, and no one will care. But show a traveler in Chicago the average airline delay at ORD, right now and you will have their interest—the difference is that the data is in context (O’Hare Airport, or ORD, for a traveler in Chicago) and that it is real time.

In this chapter, therefore, we will look at building dashboards that combine accurate depictions with explanatory power and interactivity in a compelling package. This seems to be a strange time to be talking about building dashboards—shouldn’t the building of a dashboard wait until after we have built the best possible predictive model?

Why Build a Dashboard First?

Building a dashboard when building a machine learning model is akin to building a form or survey tool to help you build the machine learning model. To build powerful machine learning models, you need to understand the dataset and devise features that help with prediction. By building a dashboard, you get to rope in the eventual users of the predictive model to take a close look at your data. Their fine-grained look at the data (remember that everyone is looking at the data corresponding to their context) will complement your more overarching look at it. As they look at the data and keep sending suggestions and insights about the data to your team, you will be able to incorporate them into the machine learning model that you actually build.

In addition, when presented with a dataset, you should be careful that the data is the way you imagine it to be. There is no substitute for exposing and exploring the data to ensure that. Doing such exploratory data analysis with an immediately attainable milestone—building a dashboard from your dataset—is a fine way to do something real with the data and develop awareness of the subtleties of your data. Just as you often understand a concept best when you explain it to someone, building an explanatory display for your data is one of the best ways to develop your understanding of a dataset. The fact that you have to visualize the data in order to do basic preprocessing such as outlier detection makes it clear that building visual representations is work you will be doing anyway. If you are going to be doing it, you might as well do it well, with an eye toward its eventual use in production.

Eventual use in production is the third reason why you should develop the dashboard first instead of leaving it as an afterthought. Building explanatory power should be constantly on your mind as you develop the machine learning model. Giving users just the machine learning model will often go down poorly—they have no insight into why the system is recommending whatever it does. Adding explanations to the recommendations is more likely to succeed. For example, if you accompany your model prediction with five of the most salient features presented in an explanatory way, it will help make the model output more believable and trustworthy. 2

Even for cases for which the system performs poorly, you will receive feedback along the lines of “the prediction was wrong, but it is because Feature #3 was fishy. I think maybe you should also look at Factor Y.” In other words, shipping your machine learning model along with an explanation of its behavior gets you more satisfied users, and users whose criticism will be a lot more constructive. It can be tempting to ship the machine learning model as soon as it is ready, but if there is a dashboard already available (because you were building it in parallel), it is easier to counsel that product designers consider the machine learning model and its explanatory dashboard as the complete product.

Where should these dashboards be implemented? Find out the environment that gets the largest audience of experts and eventual users and build your dashboard to target that environment.

Your users might already have a visualization interface with which they are familiar. Especially when it comes to real-time data, your users might spend their entire work-day facing a visualization program that is targeted toward power users—this is true of weather forecasts, air traffic controllers, and options traders. If that is the case, look for ways to embed your visualizations into that interface. In other cases, your users might prefer that your visualizations be available from the convenience of their web browser. If this is the case, look for a visualization tool that lets you share the report as an interactive, commentable document (not just a static web page). In many cases, you might have to build multiple dashboards for different sets of users (don’t shoehorn everything into the same dashboard).

Accuracy, Honesty, and Good Design

Because the explanatory power of a good dashboard is why we are building visualizations, it is important to ensure that our explanations are not misleading. In this regard, it is best not to do anything too surprising. Although modern-day visualization programs are chock-full of types of graphs and palettes, it is best to pair any graphic with the idiom for which it is appropriate. For example, some types of graphics are better suited to relational data than others, and some graphics are better suited to categorical data than to numerical data.

Broadly, there are four fundamental types of graphics: relational (illustrating the relationship between pairs of variables), time–series (illustrating the change of a variable over time), geographical maps (illustrating the variation of a variable by location), and narratives (to support an argument). Narrative graphics are the ones in magazine spreads, which win major design awards. The other three are more worker-like.

You have likely seen enough graphical representations to realize intuitively that the graph is somehow wrong when you violate an accuracy, honesty, or aesthetic principle,3 but this section of the book lists a few of the canonical ones. For example, it is advisable to choose line graphs or scatter plots for relational graphics, and to ensure that autoscaling of the axes doesn’t portray a misleading story about your data. A good design principle is that your time–series graphs should be more horizontal than vertical, and that it is the data lines and not the graph’s accoutrements (grid lines, labels, etc.) that ought to dominate your graphics. Maximizing the ratio of data to space and ink is a principle that will stand you in good stead when it comes to geographical data—ensure that the domain is clipped to the region of interest, and go easy on place names and other text labels.

Just as you probably learned to write well by reading good writers, one of the best ways to develop a feel for accurate and compelling graphics is to increase your exposure to good exemplars. The Economist newspaper4 has a Graphic Detail blog that is worth following—they publish a chart, map, or infographic every weekday, and these run the gamut of the fundamental graphics types. Figure 3-1 shows a graphic from the blog.5

The graphic depicts the increase in the number of coauthors on scientific papers over the past two decades. The graphic itself illustrates several principles of good design. It is a time–series, and as you’d expect of this type of graphic, the time is on the horizontal axis and the time-varying quantity (number of authors per article or the number of articles per author) is on the vertical axis. The vertical axis values start out at zero, so that the height of the graphs is an accurate indicator of magnitude. Note how minimal the chart junk is—the axes labels and gridlines are very subtle and the title doesn’t draw attention to itself. The data lines, on the other hand, are what pop out. Note also the effective use of repetition—instead of all the different disciplines (Economics, Engineering, etc.) being on the same graph, each discipline is displayed on its own panel. This serves to reduce clutter and makes the graphs easy to interpret. Each panel has two graphs, one for authors per article and the other for articles per author. The colors remain consistent across the panels for easy comparison, and the placement of the panels also encourages such comparisons. We see, for example, that the increase in number of authors per article is not accompanied by an increase in articles per author in any of the disciplines, except for Physics & Astronomy. Perhaps the physicists and astronomers are gaming the system?

Figure 3-1. Graphic from the Economist showing increase in number of authors of papers in various academic disciplines over time

The graphic does, however, subtly mislead viewers who are in a hurry. Take a moment and try to critique the graphic—figure out how a viewer might have been misled. It has to do with the arrangement of the panels. It appears that the creator of the graphic has arranged the panels to provide a pleasing upward trend between the panels, but this upward trend is misleading because there is no relationship between the number of authors per article in Economics in 2016 and the same quantity in Engineering in 1996. This misdirection is concerning because the graph is supposed to support the narrative of an increasing number of authors, but the increase is not from one author to six authors over two decades—the actual increase is much less dramatic (for example, from four to six in medicine). However, a viewer who only glances at the data might wrongly believe that the increase in number of authors is depicted by the whole graph, and is therefore much more than it really is.

Loading Data into Cloud SQL

To create dashboards to allow interactive analysis of the data, we will need to store the data in a manner that permits fast random access and aggregations. Because our flight data is tabular, SQL is a natural choice, and if we are going to be using SQL, we should consider whether a relational database meets our needs. Relational databases are a mature technology and remain the tool of choice for many business problems. Relational database technology is well known and comes with a rich ecosystem of interoperable tools. The problem of standardized access to relational databases from high-level programming languages is pretty much solved.

PostgreSQL is a very popular, open-source relational database that is used in production at many enterprises. In addition to its high performance, PostgreSQL is easy to program against—it supports ANSI-SQL, GIS functionality, client libraries in a variety of programming languages, and standard connector technologies such as Open Database Connectivity (ODBC) and Java Database Connectivity (JDBC).

Create a Google Cloud SQL Instance

Google Cloud SQL offers a managed database service that supports PostgreSQL, MySQL and SQL Server. Cloud SQL manages backups, patches, updates, and even replication while providing for global availability, automatic failover, and high uptime. For best performance, choose a machine whose RAM is large enough to hold your largest table in memory—as of this writing, available machine types range from a single CPU with less than 4 GB of memory all the way to a 96 CPU machine with 624 GB of memory. Balance this desire for speed with the monthly cost of a machine, of course.

Let’s configure a Cloud SQL instance, create a database table in it, and load the table with the data we ingested into Cloud Storage. You can do all these things on the command line using gcloud, but let’s begin by using the SQL section of Cloud Platform Console and select Create Instance. Choose PostgreSQL and then fill out the form as follows:

1. Call the instance “flights”

2. Generate a strong password by clicking on the GENERATE button

3. Choose the default PosgreSQL version

4. Choose the region where your bucket of CSV data exists.

5. Choose a single zone instance since we are just trying it out. We won’t take this to production.

6. Choose a Standard machine type with 2 vCPU.

7. Click Create Instance, accepting all the other defaults

INTERACTING WITH GOOGLE CLOUD PLATFORM

Instead of filling out the dialog box by hand, we could have used the command-line tool gcloud from CloudShell (or any other machine that has gcloud installed); here’s how to do that:

gcloud sql instances create flights \

--database-version=POSTGRES_13 --cpu=2 --memory=8GiB \

--zone=us-central1-f --root-password=somestrongpassword

In the rest of the book, I will show you just the gcloud commands, but you don’t need to memorize them. You can use --help at any time on gcloud to get a list of options. For example,

gcloud sql instances create --help

will give you all the options available to create Cloud SQL instances (the database-version, its zone, etc.), whereas

gcloud sql instances --help

will give you all the ways in which you can work with Cloud SQL instances (create, delete, restart, export, etc.).

In general, everything you can do on the command line is doable using the Cloud Platform Console, and vice versa. In fact, both the Cloud Platform Console and the gcloud command invoke REST API actions. You can invoke the same REST APIs from your programs (the APIs are documented on the Google Cloud Platform website). Here is the REST API call to create an instance from Bash:

ACCESS_TOKEN="$(gcloud auth application-default print-access-token)"

curl --header "Authorization: Bearer ${ACCESS_TOKEN}" \

--header 'Content-Type: application/json' \

--data '{"name":"flights", "settings":

{"database-version":"POSTGRES_13", …}' \

> https://www.googleapis.com/sql/v1beta4/projects/[PROJECT-ID]/instances \

-X POST

Alternatively, you can use the gcloud client library (available for a variety of programming languages) to issue the REST API calls. We saw this in Chapter 2 when we used the google.cloud.storage Python package to interact with Cloud Storage.

In summary, there are four ways to interact with Google Cloud Platform:

- The Cloud Platform Console

- gcloud command from the command line in CloudShell or a machine that has the gcloud SDK installed

- Directly invoke the REST API

- Google Cloud Client library (available as of this writing for Go, Java, Node.js, Python, Ruby, PHP, and C#)

In this book, I use primarily option 2 (from the shell) and option 4 (from Python programs).

Create Table of Data

In order to import data into a Postgres table, we first have to create an empty database and a table with the correct schema.

In the GCP web console, navigate to the databases section of Cloud SQL and create a new database called bts. This will be where we load our data.

Next, we have to creating a file with the following syntax, to create a column for every field in the CSV file:

drop table if exists flights;

CREATE TABLE flights (

"Year" TEXT,

"Quarter" TEXT,

"Month" TEXT,

"DayofMonth" TEXT,

"DayOfWeek" TEXT,

"FlightDate" DATE,

"Reporting_Airline" TEXT,

"DOT_ID_Reporting_Airline" TEXT,

"IATA_CODE_Reporting_Airline" TEXT,

…

For your convenience, the file I created is already in the Git repository, so just go to Cloud Shell, change into the 03_sqlstudio directory

Stage the file into Google Cloud Storage (changing the bucket to one that you own):

gsutil cp create_table.sql \

gs://cloud-training-demos-ml/flights/ch3/create_table.sql

In the web console, navigate to the flights instance of Cloud SQL and select IMPORT. In the form, specify the location of create_table.sql and specify that you want to create a table in the database bts (see Figure 3-2).

Figure 3-2. Creating an empty table.

A few seconds later, the empty table will be created.

We can now load the CSV files into this table. Start by loading the January data by browsing to 201501.csv in your bucket and specifying CSV as the format, bts as the database, and flights as the table (see Figure 3-3).

Figure 3-3. Populating the table with data from January.

WARNING

Note that the user interface doesn’t provide a way to skip the first line, so the header will also get loaded as a row in the table. Fortunately, our schema calls all the fields as text, so this doesn’t pose a problem – after loading the data, we can delete the row corresponding to the header. If we have a more realistic schema, we will have to remove the header line before loading the file.

Interacting with the database

We can connect to the Cloud SQL instance from Cloud Shell using: 6

gcloud sql connect flights --user=postgres

In the prompt that comes up, we connect to the bts database:

\c bts;

Then, we can run a query to obtain the 5 busiest airports:

SELECT "Origin", COUNT(*) AS num_flights

FROM flights GROUP BY "Origin"

ORDER BY num_flights DESC

LIMIT 5;

While this is performant because the dataset is relatively small (only January!), as I added more months, the database started to slow down.

Relational databases are particularly well suited to smallish datasets on which we wish to do ad hoc queries. Even for larger datasets, we can tune the performance of a relational database by indexing the columns of interest. Further, because relational databases typically support transactions and guarantee strong consistency, they are an excellent choice for data that will be updated often.

However, a relational database is a poor choice if your data is primarily read-only, if your dataset sizes go into the terabyte range, or if your data streams in at high rates. This describes our flight delay use case. So, let’s switch from a relational database to an analytics data warehouse – BigQuery. The analytics data warehouse will allow us to use SQL and is much more capable of dealing with large datasets and ad hoc queries (i.e. doesn’t need the columns to be indexed).

If you are following along with me, delete the Cloud SQL instance. We won’t need it any further in this book.

Querying Using BigQuery

In Chapter 2, we loaded the CSV data into BigQuery into a table named flights_raw in the dataset dsongcp . Let’s explore that dataset a bit – this is not a full exploratory analysis, which I will do in Chapter 5.

My goal here is to do “just enough” analysis on the data and then quickly pivot to building my first model. Once I have the model, I will be able to build a dashboard to explain that model. The idea is to get a first iteration out in front of users as quickly as possible. Going from ingested data to minimum viable outputs (model, dashboard, etc) quickly is what agile development in data science looks like.

Teams that wait until they build a fully capable model before incorporating it into decision tools often build the wrong model (i.e. they solve the wrong problem because of misunderstanding how the decision will be used) or choose unviable technology (that is hard to get into production). Avoid these traps by testing your work with real users as quickly as possible!

Schema Exploration

Navigate to the BigQuery section of the Google Cloud web console, select the flights_raw table. On the right side of the window, select Schema (see Figure 3-4). Which fields do you think are relevant to predicting flight arrival delays?

Figure 3-4. The schema of the flights_raw table that we loaded into BigQuery in Chapter 2.

Just looking at the schema is not enough. For example, do we really need the Year, Month, DayOfMonth, and so on? Isn’t the FlightDate enough? It’s best to not have duplicative data – the more columns we have, the more work we have to do to keep analysis consistent.

Similarly, which of the various Airline columns do we need? For the Airline columns, we did read the description on the BTS website in Chapter 2, and will probably follow their recommendation that Reporting_Airline be the one that we use. Still, it’s worth verifying why that is.

To make decisions like this, we can use two features – the Preview tab and the Table Explorer tab (see Figure 3-4).

Using Preview

Looking at the preview (see Figure 3-5), the Year, Month, etc. columns do seem to be redundant. (If you are following along with me, you may see different rows, because the Preview just picks whatever is most handy.)

Figure 3-5. Preview of the flights_raw table that we loaded into BigQuery in Chapter 2.

Let’s check whether we can resurrect the FlightDate from the other columns and extract the date pieces from the FlightDate. We can do that with SQL:

SELECT

FORMAT("%s-%02d-%02d",

Year,

CAST(Month AS INT64),

CAST(DayofMonth AS INT64)) AS resurrect,

FlightDate,

CAST(EXTRACT(YEAR FROM FlightDate) AS INT64) AS ex_year,

CAST(EXTRACT(MONTH FROM FlightDate) AS INT64) AS ex_month,

CAST(EXTRACT(DAY FROM FlightDate) AS INT64) AS ex_day,

FROM dsongcp.flights_raw

LIMIT 5

The result appears to bear this out:

Row

resurrect

FlightDate

ex_year

ex_month

ex_day

1

2015-02-19

2015-02-19

2015

2

19

2

2015-02-20

2015-02-20

2015

2

20

3

2015-02-22

2015-02-22

2015

2

22

4

2015-02-23

2015-02-23

2015

2

23

5

2015-02-25

2015-02-25

2015

2

25

But we have to be sure. Let’s print out rows where the extracted data from FlightDate are not identical:

WITH data AS (

SELECT

FORMAT("%s-%02d-%02d",

Year,

CAST(Month AS INT64),

CAST(DayofMonth AS INT64)) AS resurrect,

FlightDate,

CAST(EXTRACT(YEAR FROM FlightDate) AS INT64) AS ex_year,

CAST(EXTRACT(MONTH FROM FlightDate) AS INT64) AS ex_month,

CAST(EXTRACT(DAY FROM FlightDate) AS INT64) AS ex_day,

FROM dsongcp.flights_raw

)

SELECT * FROM data

WHERE resurrect != CAST(FlightDate AS STRING)

This query returns an empty resultset, and so we are sure that we can safely keep only the FlightDate column.

Using Table Explorer

How about the Airline code? Switch to the Table Explorer tab and select the three airline columns as shown in Figure 3-6.

Figure 3-6. Selecting fields for Table Explorer.

BigQuery analyzes the full dataset and shows the unique values in the table, as shown in Figure 3-6.

Figure 3-7. Distinct values for the three Airline fields.

It is clear from the Table Explorer that we want to use either the Reporting_Airline or the IATA_CODE_Reporting_Airline. As before checking to see if there are rows where these are different indicates that Reporting_Airline is sufficient.

Creating BigQuery View

Based on such analysis on the remaining fields, I came up with the following sets of operations I want to do to the raw data to make it more usable. For example, the Departure Delay should be a floating point number and not a string. The cancellation code should be a boolean and not “1.00”:

CREATE OR REPLACE VIEW dsongcp.flights AS

SELECT

FlightDate AS FL_DATE,

Reporting_Airline AS UNIQUE_CARRIER,

OriginAirportSeqID AS ORIGIN_AIRPORT_SEQ_ID,

Origin AS ORIGIN,

DestAirportSeqID AS DEST_AIRPORT_SEQ_ID,

Dest AS DEST,

CRSDepTime AS CRS_DEP_TIME,

DepTime AS DEP_TIME,

CAST(DepDelay AS FLOAT64) AS DEP_DELAY,

CAST(TaxiOut AS FLOAT64) AS TAXI_OUT,

WheelsOff AS WHEELS_OFF,

WheelsOn AS WHEELS_ON,

CAST(TaxiIn AS FLOAT64) AS TAXI_IN,

CRSArrTime AS CRS_ARR_TIME,

ArrTime AS ARR_TIME,

CAST(ArrDelay AS FLOAT64) AS ARR_DELAY,

IF(Cancelled = '1.00', True, False) AS CANCELLED,

IF(Diverted = '1.00', True, False) AS DIVERTED,

DISTANCE

FROM dsongcp.flights_raw;

In order to avoid repeating these casts in all queries from here on out, I am creating a view that consists of the SELECT statement (see the first line in the listing above). A view is a virtual table – we can query the view just as if it were a table:

SELECT

ORIGIN,

COUNT(*) AS num_flights

FROM dsongcp.flights

GROUP BY ORIGIN

ORDER BY num_flights DESC

LIMIT 5

Any queries that happen on the view are rewritten by the database engine to happen on the original table – conceptually, a view works as if the SQL corresponding to the view was to be inserted into every query that uses the view.

What if the view includes a WHERE clause so that the number of rows is much less? In such cases, it would be far more efficient to export the results into a table and query that table instead:

CREATE OR REPLACE TABLE dsongcp.flights AS

SELECT

But what if you export the results into a table and then the original table has a new month of data added to it? We’d have to rerun the table creation statement to make the extracted table up-to-date. In the case of a view, we wouldn’t have to do anything special – all new queries would automatically be querying the entire raw table, and so include the new month of data.

Can we have our cake and eat it too? Can we get the “live” nature of a view, but the query efficiency of a table? Yes. It’s called a materialized view :

CREATE MATERIALIZED VIEW dsongcp.flights AS

SELECT

The view is materialized into a table, and kept up to date by BigQuery. While views are free, materialized views carry an extra cost because of the extra storage and compute overhead they involve.

In this book, I’ll use a regular view during development, since it’s easy to come back and add new columns, etc.. Later on, once we go to production, it’s quite simple to change it over to a materialized view – none of the client code will need to change.

Building Our First Model

Intuitively, we feel that if the flight is delayed by 15 minutes, it will also tend to arrive 15 minutes late. So, our model could be that we cancel the meeting if the departure delay of the flight is 15 minutes or more. Of course, there is nothing here about the probability (recall that we wanted to cancel if the probability of an arrival delay of 15 minutes was greater than 30%). Still, it will be a quick start and give us something that we can ship now and iterate upon.

Contingency Table

Suppose that we need to know how often we will be making the right decision if our decision rule is the following:

If DEP_DELAY ≥ 15, cancel the meeting; otherwise, go ahead.

There are four possibilities in the contingency table or the confusion matrix , which you can see in Table 3-2.

Table 3-2. Table 3-2. Confusion matrix for the decision to cancel the meeting

Arrival delay < 15 minutes

Arrival delay ≥ 15 minutes

We did not cancel meeting

Correct (true negative)

False negative

We cancel meeting

False positive

Correct (true positive)

If we don’t cancel the meeting and it turns out that the flight arrived more than 15 minutes late (let’s call that a “positive”), it is clear that we made a wrong decision. It is arbitrary whether we refer to it as a false negative (treating the late arrival as a positive event) or as a false positive (treating the late arrival as a negative event). The common approach is to term the rarer event, or the event we are looking for, as the positive. Here, delayed flights are (hopefully) rarer, and we are deciding whether to cancel the meeting. How do we find out how often the decision rule of thresholding the departure delay at 15 minutes will tend to be correct? We can evaluate the first box in the confusion matrix using BigQuery:

SELECT

COUNT(*) AS true_negatives

FROM dsongcp.flights

WHERE dep_delay < 15 AND arr_delay < 15

There are 4430885 such flights.

To compute all four values in a single statement, move the WHERE clause into the SELECT itself:

SELECT

COUNTIF(dep_delay < 15 AND arr_delay < 15) AS true_negatives,

COUNTIF(dep_delay < 15 AND arr_delay >= 15) AS false_negatives,

COUNTIF(dep_delay >= 15 AND arr_delay < 15) AS false_positives,

COUNTIF(dep_delay >= 15 AND arr_delay >= 15) AS true_positives,

COUNT(*) AS total

FROM dsongcp.flights

WHERE arr_delay IS NOT NULL AND dep_delay IS NOT NULL

Each of the COUNTIF statements counts maintains a count of the number of rows that match the given criterion, and COUNT(*) counts all rows. This way, we get to scan the table just once, and still manage to collect the four numbers that form the confusion matrix:

Row

true_negatives

false_negatives

false_positives

true_positives

total

1

4430885

232701

219684

830738

5714008

Recall that these numbers assume that we are making a decision by thresholding the departure delay at 15 minutes. But is that the best threshold?

Threshold Optimization

Ideally, we want to try out different values of the threshold and pick the one that provides the best results. To do so, we can declare a variable called THRESH and use it in the query. This way, there is just one number to change when we want to try out a different threshold.:

DECLARE THRESH INT64;

SET THRESH = 15;SELECT

COUNTIF(dep_delay < THRESH AND arr_delay < 15) AS true_negatives,

COUNTIF(dep_delay < THRESH AND arr_delay >= 15) AS false_negatives,

COUNTIF(dep_delay >= THRESH AND arr_delay < 15) AS false_positives,

COUNTIF(dep_delay >= THRESH AND arr_delay >= 15) AS true_positives,

COUNT(*) AS total

FROM dsongcp.flights

WHERE arr_delay IS NOT NULL AND dep_delay IS NOT NULL

Still, I’d rather not run the query several times, once for each threshold. It’s not about the drudgery of it – I could avoid the manual work by using a for loop in a script. What I’m objecting to is scanning the table four times. The better way to do this in SQL is to declare an array of possible thresholds and then group by them:

SELECT

THRESH,

COUNTIF(dep_delay < THRESH AND arr_delay < 15) AS true_negatives,

COUNTIF(dep_delay < THRESH AND arr_delay >= 15) AS false_negatives,

COUNTIF(dep_delay >= THRESH AND arr_delay < 15) AS false_positives,

COUNTIF(dep_delay >= THRESH AND arr_delay >= 15) AS true_positives,

COUNT(*) AS total

FROM dsongcp.flights, UNNEST([5, 10, 11, 12, 13, 15, 20]) AS THRESH

WHERE arr_delay IS NOT NULL AND dep_delay IS NOT NULL

GROUP BY THRESH

This way, we get to run a single query, which scans the table just once, and still manage to create contingency tables for all the thresholds we want to try. The result consists of the four contingency table values for each of the seven values of the threshold:

Row

THRESH

true_negatives

false_negatives

false_positives

true_positives

total

1

5

3931979

144669

718590

918770

5714008

2

10

4242286

184944

408283

878495

5714008

3

11

4288279

193912

362290

869527

5714008

4

12

4329146

203068

321423

860371

5714008

5

13

4366641

212498

283928

850941

5714008

6

15

4430885

232701

219684

830738

5714008

7

20

4542475

291791

108094

771648

5714008

TIP

Learn SQL. You’ll thank me later.

This is all well-and-good, but recall that our goal (see Chapter 1) is to cancel the client meeting if the probability of arriving 15 minutes late is 30% or more. How close do we get with each of these thresholds?

To know this, we need to compute the fraction of times a decision is wrong. We can do this by calling the above result the contingency table, and then computing the necessary ratios:

WITH contingency_table AS ( SELECT

THRESH,

COUNTIF(dep_delay < THRESH AND arr_delay < 15) AS true_negatives,

COUNTIF(dep_delay < THRESH AND arr_delay >= 15) AS false_negatives,

COUNTIF(dep_delay >= THRESH AND arr_delay < 15) AS false_positives,

COUNTIF(dep_delay >= THRESH AND arr_delay >= 15) AS true_positives,

COUNT(*) AS total

FROM dsongcp.flights, UNNEST([5, 10, 11, 12, 13, 15, 20]) AS THRESH

WHERE arr_delay IS NOT NULL AND dep_delay IS NOT NULL

GROUP BY THRESH

)

SELECT

ROUND((true_positives + true_negatives)/total, 2) AS accuracy,

ROUND(false_positives/(true_positives+false_positives), 2) AS fpr,

ROUND(false_negatives/(false_negatives+true_negatives), 2) AS fnr,

*

FROM contingency_table

The result now includes the accuracy, false positive rate, and false negative rate:

Row

accuracy

fpr

fnr

THRESH

true_negatives

false_negatives

false_positives

true_positives

total

1

0.85

0.44

0.04

5

3931979

144669

718590

918770

5714008

2

0.9

0.32

0.04

10

4242286

184944

408283

878495

5714008

3

0.9

0.29

0.04

11

4288279

193912

362290

869527

5714008

4

0.91

0.27

0.04

12

4329146

203068

321423

860371

5714008

5

0.91

0.25

0.05

13

4366641

212498

283928

850941

5714008

6

0.92

0.21

0.05

15

4430885

232701

219684

830738

5714008

7

0.93

0.12

0.06

20

4542475

291791

108094

771648

5714008

We want to cancel the meeting if the probability of the flight being late is more than 30%. If we are going to make the decision based on the departure delay, we have to choose a departure delay that is such that 30% of flights with that departure delay are late. Because we defined on-time arrival as the positive event, this implies that the false positive rate is 30%.

It is clear from the table above that if we want our decision to have a false positive rate of 30%, the departure delay threshold needs to be 10 or 11 minutes (in the dataset, departure delay is an integer, so an intermediate threshold like 10.3 minutes does not make sense).

If we choose a threshold of 10 minutes, we will make the correct decision 96% of the time when we don’t cancel the meeting and 68% of the time when we cancel the meeting. Overall, we are correct 90% of the time.

Note that 10 minutes is not the threshold that maximizes the overall accuracy. Had we chosen a threshold of 20 minutes, we’d cancel far fewer meetings (108k vs. 408k) and be correct more often overall (93%). However, that would be very conservative. Since it is not our goal to be correct 88% of the times we cancel the meeting – we only want to be correct 70% of the time – 10 minutes is the right threshold.

However, we could also consider that if we can increase the threshold to 20 minutes, we would be correct far more often with very little impact on the false negative rate. Until we looked at the data, we didn’t know what was achievable, and it is possible that the original target was set in a fog of uncertainty. It might be worthwhile asking our stakeholders whether they are really wedded to the 30% false negative rate, and whether we have leeway to change the tradeoffs available to users of our application – a dashboard that shows the impact of a threshold is an excellent way to gauge this. If the stakeholders don’t know, it might be worth doing an A/B test with a focus group, and that’s what we are about to do next.

IS THIS MACHINE LEARNING?

What we did here—trying different thresholds—is at the heart of machine learning. Our model is a simple rule that has a single parameter (the departure delay threshold) and we can tune it to maximize some objective measure (here, the desired precision). We can (and will) use more complex models, and we can definitely make our search through the parameter space a lot more systematic, but this process of devising a model and tuning it is the gist of machine learning. We haven’t evaluated the model (we can’t take the 70% we got on the 12 months of 2015 data and claim that to be our model performance—we need an independent dataset to do that), and that is a key step in machine learning. However, we can plausibly claim that we have built a simple machine learning model to provide guidance on whether to cancel a meeting based on historical flight data.

Building a Dashboard

Even this simple model is enough for us to begin getting feedback from end users. Recall that my gut instinct at the beginning of the previous section was that I needed to use a 15-minute threshold on the departure delay. Analysis of the contingency table, however, indicated that the right threshold to use was 10 minutes. I’m satisfied with this model as a first step, but will our end users be? Let’s go about building a dashboard that explains the model recommendations to end users. Doing so will also help clarify what I mean by explaining a model to end users.

There is a large number of business intelligence and visualization tools available, and many of them connect with data sources like BigQuery and Cloud SQL on Google Cloud Platform. In this chapter, we build dashboards using Data Studio, which is free and comes as part of Google Cloud Platform, but you should be able to do similar things with Tableau, QlikView, Looker, and so on.

LOOKER OR DATA STUDIO?

Google Cloud has two business intelligence tools – Looker and Data Studio. Data Studio is free and much more suitable for self-service use. Looker is much more capable and more suitable for enterprise use.

What do I mean by enterprise use? Here are a few examples of things that Looker can do that Data Studio can’t:

Consistency

It is possible for one team to define a semantic layer consisting of standard nomenclature for columns and ways of computing key performance metrics.

Multi-cloud

Looker can access data in BigQuery, Amazon Redshift, Azure SQL Data Warehouse, and Snowflake simultaneously in the same report.

Embedded analytics

Have you been to a website where you can see charts and graphs of your activity? This is provided by a lot of B2B applications, such as marketplaces allowing sellers to visualize their own data. Looker allows you to embed analytics.

Alerts and updates

Data Studio requires the user to visit the Data Studio web page and refresh the graphics. With Looker, you can push reports on a schedule or whenever an event happens.

That said, in our case, all we want is a self-serve dashboard and Data Studio fits the bill perfectly.

For those of you with a data science background, I’d like to set expectations here—a dashboard is a way for end users to quickly come to terms with the current state of a system and is not a full-blown, completely customizable, statistical visualization package. Think about the difference between what’s rendered in the dashboard of a car versus what would be rendered in an engineering visualization of the aerodynamics of the car in a wind tunnel—that’s the difference between what we will do in Data Studio versus what we will use Vertex AI Notebooks for in later chapters. Here, the emphasis is on providing information effectively to end users—thus, the key aspects are interactivity and collaboration. With Data Studio, you can share reports similarly to Google Workspace documents; that is, you can give different colleagues viewing or editing rights, and colleagues with whom you have shared a visualization can refresh the charts to view the most current data.

Getting Started with Data Studio

To work with Data Studio, navigate to http://datastudio.google.com/. There are two key concepts in Data Studio: reports and data sources. A report is a set of charts and commentary that you create and share with others. The charts in the report are built from data that is retrieved from a data source. The first step, therefore, is to set up a data source. Because our data is in BigQuery, the data source we need to set up is for Data Studio to connect to BigQuery.

On the Data Studio home page, click on the Create button, click the Data source menu item, and choose the BigQuery button, as illustrated in Figure 3-8.7

Figure 3-8. Choose BigQuery from the Data Source menu item in Data Studio

Select your project, the dsongcp dataset and flights as the table. Then, click on the Connect button. Recall that flights is the view that we have set up with the streamlined set of fields.

A list of fields in the table displays, with Data Studio inferring something about the fields based on a sampling of the data in that table. We’ll come back and correct some of these, but for now, just click Create Report, accepting all the prompts.

Creating Charts

On the top ribbon, select the scatter chart icon from the “Add a chart” pulldown (see Figure 3-10) and draw a rectangle somewhere in the main window; Data Studio will draw a chart. The data that is rendered is pulled from some rather arbitrary columns.

Figure 3-9. Initial chart rendered by Data Studio

Ignoring the Date Range Dimension for now, there are three columns being used: the Dimension is the quantity being plotted; Metric X is along the x-axis; and Metric Y is along the y-axis. Change (if necessary) Dimension to UNIQUE_CARRIER, Metric X to DEP_DELAY, Metric Y to ARR_DELAY, and change the aggregation metric for both Metric X and Metric Y to Average. Ostensibly, this should give us the average departure delay and arrival delay of different carriers. Click the Style tab and add in a linear trendline and show the data labels. Figure 3-11 depicts the resulting chart.

Figure 3-10. Chart after changing metric, dimension, and style.

Adding End-User Controls

So far, our chart is static—there is nothing for end users to interact with. They get to see a pretty picture, but do not get to change anything about our graph. To permit the end user to change something about the graph, we should add controls to our graph.

Let’s give our end users the ability to set a date range. On the top icon ribbon, click the “Date range” button, as illustrated in Figure 3-11.

Figure 3-11. The Date Range control on the top icon ribbon

On your chart, place the rectangle where you’d like the control to appear. Change the time window to be Fixed and set the Start Date to Jan 1, 2015 and end date to Dec 31, 2019 8 . This is how the report will initially appear to users.

In the upper-right corner, change the toggle to switch to the View mode. This is the mode in which users interact with your report. Change the data range to Jan 1, 2015 to May 31, 2015 (see Figure 3-12) and you should see the chart immediately update.

Figure 3-12. The chart in View mode.

Pause a bit here and ask yourself what kind of a model the chart in Figure 3-12 would explain. Because there is a line, it strongly hints at a linear model. If we were to recommend meeting cancelations based on this chart, we’d be suggesting, based on the linear trend of arrival delay with departure delay, that departure delays of more than 20 minutes lead to arrival delays of more than 15 minutes. That, of course, was not our model—we did not do linear regression, and certainly not airline by airline. Instead, we picked a departure threshold based on a contingency table over the entire dataset. So, we should not use the graph above in our dashboard—it would be a misleading description of our actual model.

Showing Proportions with a Pie Chart

How would you explain our contingency table–based thresholds to end users in a dashboard? Recall that the choice comes down to the proportion of flights that arrive more than 15 minutes after their scheduled time. That is what our dashboard needs to show.

One of the best ways to show a proportion is to use a pie chart. 9 Switch back to the Edit mode, and from the pull-down menu, select the “donut chart” button (this is a type of pie chart), and then, on your report, draw a square where you’d like the donut chart to appear (it is probably best to delete the earlier scatter plot from it). As we did earlier, we need to edit the dimensions and metrics to fit what it is that we want to display. Perhaps things will be clearer if you see what the end product ought to look like. Figure 3-13 gives you a glimpse.

Figure 3-13. Desired end result is a chart that shows the proportion of flights that are late versus on time

In this chart, we are displaying the proportion of flights that arrived late versus those that arrived on time. The labeled field ON TIME versus LATE is the Dimension. The number of flights is the metric that will be apportioned between the labels. So, how do you get these values from the BigQuery view?

It is clear that there is no column in the database that indicates the total number of flights. However, Data Studio has a special value Record Count that we can use as the metric, after making sure to change the aggregate from the default Sum to Count.

The “islate” value, though, will have to be computed as a formula. Conceptually, we need to add a new calculated field to the data that looks like this:

CASE WHEN

(ARR_DELAY < 15)

THEN

"ON TIME"

ELSE

"LATE"

END

Click on the current Dimension column and click on “Create Field”. Give the field the name is_late , enter the above formula, and change the type to Text (see Figure 3-14).

Figure 3-14. How to set up the is_late definition

The pie chart is now complete and reflects the proportion of flights that are late versus those that are on time. You can switch over to the Style tab if you’d like to change the appearance of the pie chart to be similar to mine.

Because the proportion of flights that end up being delayed is the quantity on which we are trying to make decisions, the pie chart translates quite directly to our use case. However, it doesn’t tell the user what the typical delay would be. To do that, let’s create a bar (column) chart that looks like the one shown in Figure 3-15.

Figure 3-15. Typical delay for each carrier

Here, the labeled quantity (or Dimension) is the Carrier. There are two metrics being displayed: the DEP_DELAY and ARR_DELAY, both of which are aggregated to their averages over the dataset. Figure 3-16 shows the specifications.

Figure 3-16. How to set the bar chart properties to generate the desired chart

Note the Sort column at the end—it is important to have a reliable sort order in dashboards so that users become accustomed to finding the information they want in a known place. Also, the default is to use different axes for the two variables.

Switch over to the Style tab and change this to use a single axis. Finally, Data Studio defaults to 10 bars. In the Style tab, change this to reflect that we expect to have up to 20 unique carriers (Figure 3-17).

Figure 3-17. How to set the bar chart properties to generate the desired chart

Of course, we can now add in a date control as we did earlier to end up with the report in Figure 3-18 (the All Flights in the diagram is just a text label that I added).

Figure 3-18. Resulting dashboard consisting of a pie chart and bar chart

It appears that, on average, about 80% of flights are on time and that the typical arrival delay varies between airlines but lies in a range of 0 to 15 minutes.

Explaining a Contingency Table

Even though the dashboard we just created shows users the decision-making criterion (proportion of flights that will be late) and some characteristics of that decision (the typical arrival delay), it doesn’t actually show our model. Recall that our model involved a threshold on the departure delay. We need to show that. Figure 3-19 shows what we want the dashboard to look like.

In other words, we want to show the same two charts, but for the decision thresholds that we considered—departure delays of 10, 15, and 20 minutes or more.

Figure 3-19. Dashboard consisting of three pairs of pie charts and bar charts along with a date control

To get there, we need to change our data source. No longer can we populate the chart from the entire table. Instead, we should populate it from a query that pulls only those flights whose departure delay is greater than the relevant threshold. In BigQuery, we can create the views we need 10 and use those views as data sources. Here’s how:

CREATE OR REPLACE VIEW dsongcp.delayed_10 AS

SELECT * FROM dsongcp.flights WHERE dep_delay >= 10;

CREATE OR REPLACE VIEW dsongcp.delayed_15 AS

SELECT * FROM dsongcp.flights WHERE dep_delay >= 15;

CREATE OR REPLACE VIEW dsongcp.delayed_20 AS

SELECT * FROM dsongcp.flights WHERE dep_delay >= 20;

Looking at the resulting pie chart for a 10 minute-threshold (Figure 3-19), we see that it comes quite close to our target of 30% on-time arrivals. The bar chart for the 10-minute delay explains why the threshold is important. Hint: it is not about the 10 minutes. It is about what the 10-minute delay is indicative of. Can you decipher what is going on?

Although the typical departure delay of a flight is only about 5 minutes (see the chart corresponding to all flights that we created earlier), flights that are delayed by more than 10 minutes fall into a separate statistical regime. The typical departure delay of an aircraft that departs more than 10 minutes late is around 50 minutes! A likely explanation is that a flight that is delayed by 10 minutes or more typically has a serious issue that will not be resolved quickly. If you are sitting in an airplane and it is more than 10 minutes late in departing, you might as well cancel your meeting—you are going to be sitting at the gate for a while. 11

At this point, we have created a very simple model and created dashboards to explain the model to our end users. Our end users have a visual, intuitive way to see how often our model is correct and how often it is wrong. The model might be quite simple, but the explanation of why the model works is a satisfying one.

There is one teeny, tiny thing missing, though. Context. The dashboard that we have built so far is all about historical data, whereas real dashboards need to be timely. Our dashboard shows aggregates of flights all over the country, but our users will probably care only about the airport from which they are departing and the airport to which they are going. We have a wonderfully informative dashboard, but without such time and location context, few users would care. In Chapter 4, we look at how to build real-time, location-aware dashboards—unfortunately, however, there is a problem with our dataset that prevents us from doing so immediately.

Summary

In this chapter, we discussed the importance of bringing the insights of our end users into our data modeling efforts as early as possible. Bringing their insights is possible only if you make it a point to explain your models in context from the get-go.

We tried using Cloud SQL, a transactional, relational database whose management is simplified by virtue of it running on the cloud and being managed by Google Cloud Platform. However, it stopped scaling once we got to millions of flights. Transactional databases are not built for queries that involve scanning the entire table. For such queries, we want to use an analytics data warehouse. Hence, we switched to using BigQuery.

Within BigQuery, we previewed the table, selected a subset of columns and created a view to make downstream analysis simpler.

The first model that we built was to suggest that our road warriors cancel their immediately scheduled meeting if the departure delay of the flight was more than 10 minutes. This would enable them to make 70% of their meetings with 15 minutes to spare.

We then built a dashboard in Data Studio to explain the contingency table model. Because our choice of threshold was driven by the proportion of flights that arrived late given a particular threshold, we illustrated the proportion using a pie chart for two different thresholds. We also depicted the average arrival delay given some departure delay—this gives users an intuitive understanding of why we recommend a 10-minute threshold.

1   When I worked on developing machine learning algorithms for weather prediction, nearly every one of the suggestions and feature requests that I received emanated when the person in question was looking at the real-time radar feed. There would be a storm, my colleague would watch it go up on radar, observe that the tracking of the storm was patchy, and let me know what aspect of the storm made it difficult to track. Or, someone would wake up, look at the radar image, and discover that birds leaving to forage from their roost had been wrongly tagged a mesocyclone. It was all about the real-time data. No matter how many times I asked, I never ever got anyone to look at how the algorithms performed on historical data. It was also often about Oklahoma (where our office was) because that’s what my colleagues would concentrate on. Forecasters from around the country would derisively refer to algorithms that had been hypertuned to Oklahoma supercells.

2   Explanations can be a double-edged sword because humans are not fully rational beings. Explanations can be the result of apophenia – the tendency of humans to see meaningful patterns even when there are none. This can lead to motivated reasoning – the tendency of humans to create justifications for decisions that are more desirable at an emotional level. The combination of apophenia and motivated reasoning can lead to just-so stories that attempt to justify whatever the state of the world is on the basis of spurious explanations. As data scientists, we should realize that we too are human. We need to be careful to set aside these biases, consider counterfactuals, and be willing to revise our initial judgements. Easier said than done, of course.

3   If you are not familiar with design principles, I recommend The Visual Display of Quantitative Information by Edward Tufte.

4   The Economist is published weekly as an 80-page booklet stapled in the center, and each page is about the size of a letter paper. However, for historical reasons, the company refers to itself as a newspaper rather than a magazine.

5   http://www.economist.com/blogs/graphicdetail/2016/11/daily-chart-18, published on Nov. 25, 2016.

6   If your organization has set up a security policy to allow access only from authorized networks, you might have to use a SQL proxy to connect to the instance. At the time of writing, this is available only in the beta version, so do: gcloud beta sql connect flights --user=postgres

7   Graphical user interfaces are often the fastest-changing parts of any software. So, if the user interface has changed from these screenshots by the time this book gets into your hands, please hunt around a bit. There will be some way to add a new data source.

8   Or whatever the last month that you downloaded is. Just so that you don’t have to wait a long time for the data to be available in your Google Cloud project, the ingest.sh script in Chapter 2, by default, downloads only 2015 data. Change the YEAR loop in that script to download 2015 to 2019.

9   An alternative way to show proportions, especially of a time-varying whole, is a stacked column chart.

10   Data Studio does support a BigQuery query as a data source, but it is preferable to read from a view because views are more reusable.

11   Road warriors know this well. Ten minutes in, and they whip out their phones to try to get on a different flight.

Chapter 4. Streaming Data: Publication and Ingest with Pub/Sub and Dataflow

A NOTE FOR EARLY RELEASE READERS

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 4th chapter of the final book. Please note that the GitHub repo will be made active later on.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at mcronin@oreilly.com.

In Chapter 3, we developed a dashboard to explain a contingency table–based model of suggesting whether to cancel a meeting. However, the dashboard that we built lacked immediacy because it was not tied to users’ context. Because users need to be able to view a dashboard and see the information that is relevant to them at that point, we need to build a real-time dashboard with location cues.

How would we add context to our dashboard? We’d have to show maps of delays in real time. To do that, we’ll need locations of the airports, and we’ll need real-time data. Airport locations can be obtained from the US Bureau of Transportation Statistics (BTS; the same US government agency from which we obtained our historical flight data). Real-time flight data, however, is a commercial product. If we were to build a business out of predicting flight arrivals, we’d purchase that data feed. For the purposes of this book, however, let’s just simulate it.

Simulating the creation of a real-time feed from historical data has the advantage of allowing us to see both sides of a streaming pipeline (production as well as consumption). In the following section, we look at how we could stream the ingest of data into the database if we were to receive it in real time.

TIP

All of the code snippets in this chapter are available in the GitHub repository at https://github.com/GoogleCloudPlatform/data-science-on-gcp/ in the folder 04_streaming. See the README.md file in that directory for instructions on how to do the steps described in this chapter.

Designing the Event Feed

Let’s assume that we wish to create an event feed, not with all 100 fields in the raw BTS dataset, but with only the few fields that we selected in Chapter 3 as being relevant to the flight delay prediction problem (see Figure 4-1).

Figure 4-1. In Chapter 3, we created a view in BigQuery with the fields relevant to the flight delay prediction problem. In this chapter, we will simulate a real-time stream of this information.

To simulate a real-time stream of the flight information shown in Figure 4-1, we can begin by using the historical data in the flights view in BigQuery but will need to transform it further. What kinds of transformations are needed?

Transformations Needed

Note that FL_DATE is a Date while DEP_TIME is a STRING. This is because FL_DATE is of the form 2015-07-03 for July 3, 2015, whereas DEP_DATE is of the form 1406 for 2:06 PM local time. This is unfortunate. I’m not worried about the separation of date and time into two columns—we can remedy that. What’s unfortunate is that there is no time zone offset associated with the departure time. Thus, in this dataset, a departure time of 1406 in different rows can be different times depending on the time zone of the origin airport.

The time zone offsets (there are two, one for the origin airport and another for the destination) are not present in the data. Because the offset depends on the airport location, we need to find a dataset that contains the timezone offset of each airport and then mash this data with that dataset. 1 To simplify downstream analysis, we will then put all the times in the data in a common time zone—Coordinated Universal Time (UTC) is the traditional choice of common time zone for datasets. We cannot however, get rid of the local time—we will need the local time in order to carry out analysis, such as the typical delay associated with morning flights versus evening flights. So, although we will convert the local times to UTC, we will also store the time zone offset (e.g., −3,600 minutes) to retrieve the local time if necessary.

Therefore, we are going to carry out two transformations to the original dataset. First, we will convert all the time fields in the raw dataset to UTC. Secondly, in addition to the fields present in the raw data, we will add three fields to the dataset for the origin airport and the same three fields for the destination airport: the latitude, longitude, and time zone offset. These fields will be named:

DEP_AIRPORT_LAT, DEP_AIRPORT_LON, DEP_AIRPORT_TZOFFSET

ARR_AIRPORT_LAT, ARR_AIRPORT_LON, ARR_AIRPORT_TZOFFSET

The third transformation that we will need to carry out is that for every row in the historical dataset, we will need to publish multiple events. This is because it would be too late if we wait until the aircraft has arrived to send out a single event containing all the row data. If we do this at the time the aircraft departs, our models will be violating causality constraints. Instead, we will need to send out events corresponding to each state the flight is in. Let’s choose to send out five events for each flight: when the flight is first scheduled, when the flight departs the gate, when the flight lifts off, when the flight lands, and when the flight arrives. These five events cannot have all the same data associated with them because the knowability of the columns changes during the flight. For example, when sending out an event at the departure time, we will not know the arrival time. For simplicity, we can notify the same structure, but we will need to ensure that unknowable data is marked by a null and not with the actual data value.

Architecture

Table 4-1 lists when those events can be sent out and the fields that will be included in each event.

Table 4-1. Table 4-1. Fields that will be included in each of the five events that will be published . Compare the order of the fields with those in the schema in Figure 4-1.

Event

Sent at (UTC)

Fields included in event message

Scheduled

CRS_DEP_TIME minus 7 days

FL_DATE,UNIQUE_CARRIER,ORIGIN_AIRPORT_SEQ_ID,ORIGIN,DEST_AIRPORT_SEQ_ID,DEST,CRS_DEP_TIME,[nulls],CRS_ARR_TIME,[nulls],DISTANCE

Departed

DEP_TIME

All fields available in scheduled message, plus:DEP_TIME,DEP_DELAYCANCELLED,CANCELLATION_CODEDEP_AIRPORT_LAT,DEP_AIRPORT_LON,DEP_AIRPORT_TZOFFSET

Wheelsoff

WHEELS_OFF

All fields available in departed message, plus:TAXI_OUT,WHEELS_OFF

Wheelson

WHEELS_ON

All fields available in wheelsoff message, plus:WHEELS_ONDIVERTEDARR_AIRPORT_LAT,ARR_AIRPORT_LON,ARR_AIRPORT_TZOFFSET

Arrived

ARR_TIME

All fields available in wheelson message, plus:ARR_TIME,ARR_DELAY

We will carry out the transformations needed and then store the transformed data in a database so that it is ready for the event simulation code to use. Figure 4-2 shows the steps we are about to carry out in our Extract-Transform-Load (ETL) pipeline, and the subsequent steps to simulate an event stream from these events, and then create a real-time dashboard from the simulated event stream.

Figure 4-2. (a) Steps in our ETL pipeline to transform the raw data into events (b) Simulating the event stream (c) Processing the event stream to populate a real-time dashboard

Getting airport information

In order to do the time correction, we need to obtain the latitude and longitude of each airport. The BTS has a dataset that contains this information, which we can use to do the lookup. For convenience, I’ve downloaded the data and made it publicly available at:

gs://data-science-on-gcp/edition2/raw/airports.csv

Let’s examine the data to determine how to get the latitude and longitude of the airports. In Chapter 2, when I needed to explore the flights data to create the first delays model, I loaded the data into BigQuery.

Do we have to import all the data that is shared with us into our BigQuery dataset in order to do exploration? Of course not. We can query BigQuery datasets in other projects without having to make our own copies of the data. In the FROM clause of the BigQuery query, all that we have to do is to specify the name of the project that the dataset lives in:

SELECT

airline,

AVG(departure_delay) AS avg_dep_delay

FROM `bigquery-samples.airline_ontime_data.flights`

GROUP BY airline

ORDER by avg_dep_delay DESC

What if someone shares a CSV file with us, though? Do we have to load the data into BigQuery in order to see what’s in the file? No.

BigQuery allows us to query data in Cloud Storage through its “federated query” capabilities. This is the ability of BigQuery to query data that is not stored within the data warehouse product, but instead operate on such data sources as Google Sheets (a spreadsheet on Google Drive) or files on Cloud Storage. Thus, we could leave the files as CSV on Cloud Storage, define a table structure on it, and query the CSV files directly. Recall that we suggested using Cloud Storage if your primary analysis pattern involves working with your data at the level of flat files—this is a way of occasionally applying SQL queries to such datasets.

The first step is to get the schema of these files. Let’s look at the first line:

gsutil cat gs://data-science-on-gcp/edition2/raw/airports.csv | head -1

We get:

"AIRPORT_SEQ_ID","AIRPORT_ID","AIRPORT","DISPLAY_AIRPORT_NAME","DISPLAY_AIRPORT_CITY_NAME_FULL","AIRPORT_WAC_SEQ_ID2","AIRPORT_WAC","AIRPORT_COUNTRY_NAME","AIRPORT_COUNTRY_CODE_ISO","AIRPORT_STATE_NAME","AIRPORT_STATE_CODE","AIRPORT_STATE_FIPS","CITY_MARKET_SEQ_ID","CITY_MARKET_ID","DISPLAY_CITY_MARKET_NAME_FULL","CITY_MARKET_WAC_SEQ_ID2","CITY_MARKET_WAC","LAT_DEGREES","LAT_HEMISPHERE","LAT_MINUTES","LAT_SECONDS","LATITUDE","LON_DEGREES","LON_HEMISPHERE","LON_MINUTES","LON_SECONDS","LONGITUDE","UTC_LOCAL_TIME_VARIATION","AIRPORT_START_DATE","AIRPORT_THRU_DATE","AIRPORT_IS_CLOSED","AIRPORT_IS_LATEST"

Use this header to write a BigQuery schema string of the format (specify STRING for any column you are not sure about, since you can always CAST it to the appropriate format when querying the data):

AIRPORT_SEQ_ID:INTEGER,AIRPORT_ID:STRING,AIRPORT:STRING, …

Alternately, if you have a similar dataset lying around, start from its schema and edit it:

bq show --format=prettyjson dsongcp.sometable > starter.json

Once we have the schema of the GCS files, we can make a table definition for the federated source: 2

bq mk --external_table_definition=./airport_schema.json@CSV=gs://data-science-on-gcp/edition2/raw/airports.csv dsongcp.airports_gcs

If you visit the BigQuery web console now, you should see a new table listed in the dsongcp dataset (reload the page if necessary). This is a federated data source in that its storage remains the CSV file on Cloud Storage. Yet you can query it just like any other BigQuery table:

SELECT

AIRPORT_SEQ_ID, AIRPORT_ID, AIRPORT, DISPLAY_AIRPORT_NAME,

LAT_DEGREES, LAT_HEMISPHERE, LAT_MINUTES, LAT_SECONDS, LATITUDE

FROM dsongcp.airports_gcs

WHERE DISPLAY_AIRPORT_NAME LIKE '%Seattle%'

In the query above, I am trying to find what airport column in the file I need to use, and what latitude column. The result indicates that AIRPORT and LATITUDE are the columns of interest but that there are several rows corresponding to the airport SEA.

Row

AIRPORT_SEQ_ID

AIRPORT_ID

AIRPORT

DISPLAY_AIRPORT_NAME

LAT_DEGREES

LAT_HEMISPHERE

LAT_MINUTES

LAT_SECONDS

LATITUDE

1

1247701

12477

JFB

Seattle 1st National.Bank Helipad

47

N

36

25

47.60694444

2

1474701

14747

SEA

Seattle International

47

N

26

50

47.44722222

3

1474702

14747

SEA

Seattle/Tacoma International

47

N

26

57

47.44916667

4

1474703

14747

SEA

Seattle/Tacoma International

47

N

27

0

47.45

Fortunately, there is a column that indicates which row is the latest information, so what I need to do is:

SELECT

AIRPORT, LATITUDE, LONGITUDE

FROM dsongcp.airports_gcs

WHERE AIRPORT_IS_LATEST = 1 AND AIRPORT = 'DFW'

Don’t get carried away by federated queries, though. The most appropriate uses of federated sources involve frequently changing, relatively small datasets that need to be joined with large datasets in BigQuery native tables. Because the columnar storage in BigQuery is so fundamental to its performance, we will load most data into BigQuery’s native format. Fortunately, the cost of storage in BigQuery is similar to the cost of storage in Cloud Storage, and the discount offered for long-term storage is similar—as long as the table data is not changed (querying the data is okay), the long-term discount starts to apply . So, if storage costs are a concern, we could ingest data into BigQuery and get rid of the files stored in Cloud Storage.

Sharing data

Now that we have the airports.csv in Cloud Storage and the airports’ dataset in BigQuery, it is quite likely that our colleagues will want to use this data too. Let’s share it with them – one of the benefits of bringing your data to the cloud (and more specifically into a data warehouse) is to allow the mashing of datasets across organizational boundaries. So, unless you have a clear reason not to do so, like security precautions, try to make your data widely accessible.

Costs of querying are borne by the person submitting the query to the BigQuery engine, so you don’t need to worry that you are incurring additional costs for your division by doing this. It is possible to make a GCS bucket “requester-pays” to get the same sort of billing separation for data in Cloud Storage.

Sharing a Cloud Storage dataset

To share some data in Cloud Storage, use gsutil:

gsutil -m acl ch -r -u abc@xyz.com:R gs://$BUCKET/data

In the command above, the -m indicates multithreaded mode, the -r provides access recursively starting with the top-level directory specified, and the -u indicates that this is a user being granted read (:R) access.

We could provide read access to the entire organization or a Google Group using -g:

gsutil -m acl ch -r -g your-domain.com:R gs://$BUCKET/data

Sharing a BigQuery dataset

BigQuery sharing can happen at the granularity of a column, a table, or a dataset. None of our BigQuery tables hold personally identifiable or confidential information. Therefore, there is no compelling access-control reason to control the sharing of flight information at a column-level or table-level. So, we can share the dsongcp dataset that was created in Chapter 2, and we can make everyone in the organization working on this project a bigquery.user so that they can carry out queries on this dataset. You can do this by from the BigQuery web console from the dataset menu.

In some cases, you might find that your dataset or table contains certain columns that have personally identifying or confidential information. You might need to restrict access to those columns3 while leaving the remainder of the table accessible to a wider audience. Whenever you need to provide access to a subset of a table in BigQuery (whether it is specific columns or specific rows), you can use views. Put the table itself in a dataset that is accessible to a very small set of users. Then, create a view on this table that will pull out the relevant columns and rows and save this view in a separate dataset that has wider accessibility. Your users will query only this view, and because the personally identifying or confidential information is not even present in the view, the chances of inadvertent leakage are lowered.

Another way to restrict access at the level of a BigQuery table is to use Cloud IAM. To control access at the level of a column, you’d use policy tags and Data Catalog.

Dataplex and Analytics Hub

Once you get into the habit of sharing data widely, governance can become problematic. It is better if you can administer data across Cloud Storage in a consistent manner and track lineage, etc. That’s what Dataplex is for.

It can be rather cumbersome to share tables and datasets one at a time with one user or one group at a time. To implement sharing at scale, and get statistics on how people are using the data you have shared, use Analytics Hub.

Time Correction

Correcting times reported in local time to UTC is not a simple endeavor. There are several steps:

1. Local time depends on, well, the location. The flight data that we have records only the name of the airport (ALB for Albany). We, therefore, need to obtain the latitude and longitude given an airport code. The BTS has a dataset that contains this information, which we can use to do the lookup.

2. Given a latitude/longitude pair, we need to look up the time zone from a map of global time zones. For example, given the latitude and longitude of the airport in Albany, we would need to get back America/New_York. There are several web services that do this, but the Python package timezonefinder is a more efficient option because it works completely offline. The drawback is that this package does not handle oceanic areas and some historical time zone changes,4 but that’s a trade-off that we can make for now.

3. The time zone offset at a location changes during the year due to daylight savings corrections. In New York, for example, it is six hours in summer and five hours in winter. Given the time zone (America/New_York), therefore, we also need the local departure date and time (say Jan 13, 2015 2:08 PM) in order to find the corresponding time zone offset. The Python package pytz provides this capability by using the underlying operating system.

The problem of ambiguous times still remains—every instant between 01:00 and 02:00 local time occurs twice on the day that the clock switches from daylight savings time (summer time) to standard time (winter time). So, if our dataset has a flight arriving at 01:30, we need to make a choice of what time that represents. In a real-world situation, you would look at the typical duration of the flight and choose the one that is more likely. For the purposes of this book, I’ll always assume the winter time (i.e., is_dst is False) on the dubious grounds that it is the standard time zone for that location.

The complexity of these steps should, I hope, convince you to follow best practices when storing time. Always try to store two things in your data: (1) the timestamp in UTC so that you can merge data from across the world if necessary and (2) the currently active time zone offset so that you can carry out analysis that requires the local time.5

Apache Beam/Cloud Dataflow

The canonical way to build data pipelines on Google Cloud Platform is to use Cloud Dataflow. Cloud Dataflow is an externalization of technologies called Flume and Millwheel that have been in widespread use at Google for several years. It employs a programming model that handles both batch and streaming data in a uniform manner, thus providing the ability to use the same code base both for batch and continuous stream processing. The code itself is written in Apache Beam , either in Java or Python, 6 and it is portable in the sense that it can be executed on multiple execution environments including Apache Flink and Apache Spark . On GCP, Cloud Dataflow provides a fully managed (serverless) service that is capable of executing Beam pipelines. Resources are allocated on-demand and they autoscale so as to achieve both minimal latency and high resource utilization.

Beam programming involves building a pipeline (a series of data transformations) that is submitted to a runner. The runner will build a graph and then stream data through it. Each input dataset comes from a source and each output dataset is sent to a sink. Figure 4-3 illustrates the Beam pipeline that we are about to build.

Compare the steps in Figure 4-2 with the block diagram of the ETL pipeline at the beginning of this section in Figure 4-3. Let’s build the data pipeline piece by piece.

Figure 4-3. The Dataflow pipeline that we are about to build

Parsing Airports Data

You can download information about the location of airports from the BTS website. I selected all of the fields, downloaded it to my local hard drive, extracted the CSV file, and compressed it with gzip. The gzipped airports file is available in the GitHub repository for this book.

In order to use Apache Beam from Cloud Shell, we need to install it into our Python environment. Go ahead and install the time zone packages also at this time:

virtualenv ~/beam_env

source ~/beam_env/bin/activate

python3 -m pip install --upgrade \

timezonefinder pytz \

'apache-beam[gcp]'

The Read transform in the Beam pipeline that follows reads in the airports file line by line. 7

with beam.Pipeline('DirectRunner') as pipeline:

airports = (pipeline

| beam.io.ReadFromText('airports.csv.gz')

| beam.Map(lambda line: next(csv.reader([line])))

| beam.Map(lambda fields: (fields[0], (fields[21], fields[26])))

)

For example, suppose that one of the input lines read out of the text file source is the following:

1000401,10004,"04A","Lik Mining Camp","Lik, AK",101,1,"United

States","US","Alaska","AK","02",3000401,30004,"Lik,

AK",101,1,68,"N",5,0,68.08333333,163,"W",10,0,-163.16666667,"",2007-07-01,,0,1,

The first Map takes this line and passes it to a CSV reader that parses it (taking into account fields like "Lik, AK" that have commas in them) and pulls out the fields as a list of strings. These fields are then passed to the next transform. The second Map takes the fields as input and outputs a tuple of the form (the extracted fields are shown in bold in the previous example):

(1000401, (68.08333333,-163.16666667))

The first number is the unique airport code (we use this, rather than the airport’s three-letter code, because airport locations can change over time) and the next two numbers are the latitude/longitude pair for the airport’s location. The variable airports, which is the result of these three transformations, is not a simple in-memory list of these tuples. Instead, it is an immutable collection, termed a PCollection, that you can take out-of-memory and distribute.

We can write the contents of the PCollection to a text file to verify that the pipeline is behaving correctly:

(airports

| beam.Map(lambda airport_data: '{},{}'.format(airport_data[0], ',' \

.join(airport_data[1])) )

| beam.io.WriteToText('extracted_airports')

)

Try this out: the code, in 04_streaming/transform/df01.py , is just a Python program that you can run from the command line. First, install the Apache Beam package if you haven’t yet done so and then run the program df01.py while you are in the directory containing the GitHub repository of this book:

cd 04_streaming/simulate

./install_packages.sh

python3 ./df01.py

This runs the code in df01.py locally. Later, we will change the pipeline line to:

with beam.Pipeline('DataflowRunner') as pipeline:

and get to run the pipeline on the Google Cloud Platform using the Cloud Dataflow service. With that change, simply running the Python program launches the data pipeline on multiple workers in the cloud. As with many distributed systems, the output of Cloud Dataflow is potentially sharded to one or more files. You will get a file whose name begins with “extracted_airports” (mine was extracted_airports-00000-of-00001 ), a few of whose lines might look something like this:

1000101,58.10944444,-152.90666667

1000301,65.54805556,-161.07166667

The columns are AIRPORT_SEQ_ID,LATITUDE,LONGITUDE —the order of the rows you get depends on which of the parallel workers finished first, and so could be different.

Adding Time Zone Information

Let’s now change the code to determine the time zone corresponding to a latitude/longitude pair. In our pipeline, rather than simply emitting the latitude/longitude pair, we emit a list of three items: latitude, longitude, and time zone:

airports = (pipeline

| beam.Read(beam.io.ReadFromText('airports.csv.gz'))

| beam.Map(lambda line: next(csv.reader([line])))

| beam.Map(lambda fields: (fields[0], addtimezone(fields[21], fields[26])))

)

The lambda keyword in Python sets up an anonymous function. In the case of the first use of lambda in the above snippet, that method takes one parameter ( line ) and returns the stuff following the colon. We can determine the time zone by using the timezonefinder package: 8

def addtimezone(lat, lon):

import timezonefinder

tf = timezonefinder.TimezoneFinder()

lat = float(lat)

lon = float(lon)

return (lat, lon, tf.timezone_at(lng=lon, lat=lat))

The location of the import statement in the preceding example might look strange (most Python imports tend to be at the top of the file), but is recommended by Cloud Dataflow9 so that pickling of the main session when we finally do submit it to the cloud doesn’t end up pickling imported packages, also.

For now, though, we are going to run this (df02.py) locally. This will take a while because the time zone computation involves a large number of polygon intersection checks and because we are running locally, not (yet!) distributed in the cloud. So, let’s speed it up by adding a filter to reduce the number of airport locations we have to look up:

| beam.io.ReadFromText('airports.csv.gz')

| beam.Filter(lambda line: "United States" in line and line[-2:] == '1,')

The BTS flight delay data is only for US domestic flights, so we don’t need the timezones of airports outside the United States. The reason for the second check is that airport locations change over time, but we are interested only in the current location of the airport. For example, here are the airport locations for ORD (or Chicago):

1393001,...,"ORD",...,41.97805556,...,-87.90611111,...,1950-01-01,2011-06-30,0,0,

1393002,...,"ORD",...,41.98166667,...,-87.90666667,...,2011-07-01,2013-09-30,0,0,

1393003,...,"ORD",...,41.97944444,...,-87.90750000,...,2013-10-01,2015-09-30,0,0,

1393004,...,"ORD",...,41.97722222,...,-87.90805556,...,2015-10-01,,0,1,

The first row captures the location of Chicago between 1950 and June 30, 2011. The second row is valid from July 1, 2011 to September 30, 2013. The last row, however, is the current location and this is marked by the last column (the AIRPORT_IS_LATEST field) being 1.

That’s not the only line we are interested in, however! Flights before 2015-10-01 will report the id of the last-but-one row. We could add a check for this, but this looks rather dicey for a slight bit of optimization. I’ll remove that last check, so that we have only:

| beam.io.ReadFromText('airports.csv.gz')

| beam.Filter(lambda line: "United States" in line)

Once I do this and run df02.py, the extracted information for the airports looks like this:

1672301,62.03611111,-151.45222222,America/Anchorage

1672401,43.87722222,-73.41305556,America/New_York

1672501,40.75722222,-119.21277778,America/Los_Angeles

The last column in the extracted information has the time zone, which was determined from the latitude and longitude of each airport.

Converting Times to UTC

Now that we have the time zone for each airport, we are ready to tackle converting the times in the flights data to UTC. At the time that we are developing the program, we’d prefer not to process all the months we have in BigQuery. Instead, we will create a small sample of the flights data in BigQuery against which to develop our code: 10

SELECT *

FROM dsongcp.flights

WHERE RAND() < 0.001

This returns about 6000 rows. We can use the BigQuery web UI to save these results as a JSON file. However, I prefer to script things out: 11

bq query --destination_table dsongcp.flights_sample \

--replace --nouse_legacy_sql \

'SELECT * FROM dsongcp.flights WHERE RAND() < 0.001'

bq extract --destination_format=NEWLINE_DELIMITED_JSON \

dsongcp.flights_sample  \

gs://${BUCKET}/flights/ch4/flights_sample.json

gsutil cp gs://${BUCKET}/flights/ch4/flights_sample.json .

This creates a file named flight_sample.json , a row of which looks similar to this:

{"FL_DATE":"2015-04-28","UNIQUE_CARRIER":"EV","ORIGIN_AIRPORT_SEQ_ID":"1013503","ORIGIN":"ABE","DEST_AIRPORT_SEQ_ID":"1039705","DEST":"ATL","CRS_DEP_TIME":"1600","DEP_TIME":"1555","DEP_DELAY":-5,"TAXI_OUT":7,"WHEELS_OFF":"1602","WHEELS_ON":"1747","TAXI_IN":4,"CRS_ARR_TIME":"1809","ARR_TIME":"1751","ARR_DELAY":-18,"CANCELLED":false,"DIVERTED":false,"DISTANCE":"692.00"}

Reading the flights data starts out similar to reading the airports data: 12

flights = (pipeline

| 'flights:read' >> beam.io.ReadFromText('flights_sample.json')

| 'flights:parse' >> beam.Map(lambda line: json.loads(line))

This is the same code as when we read the airports.csv.gz file, except that I am also giving a name (flights:read) to this transform step and using a JSON parser instead of a CSV parser.

The next step, though, is different because it involves two PCollections. We need to join the flights data with the airports data to find the time zone corresponding to each flight. To do that, we make the airports PCollection a “side input.” Side inputs in Beam are like views into the original PCollection, and are either lists or dicts. In this case, we will create a dict that maps airport ID to information about the airports:

flights = (pipeline

|'flights:read' >> beam.io.ReadFromText('flights_sample.json')

| 'flights:parse' >> beam.Map(lambda line: json.loads(line))

|'flights:tzcorr' >> beam.FlatMap(tz_correct,

beam.pvalue.AsDict(airports))

)

The FlatMap() method calls out to a method tz_correct() , which takes a line from flights_sample.json (containing a single flight’s information) and a Python dictionary (containing all the airports’ time zone information):

def tz_correct(fields, airport_timezones):

try:

# convert all times to UTC

dep_airport_id = fields["ORIGIN_AIRPORT_SEQ_ID"]

arr_airport_id = fields["DEST_AIRPORT_SEQ_ID"]

dep_timezone = airport_timezones[dep_airport_id][2]

arr_timezone = airport_timezones[arr_airport_id][2]

for f in ["CRS_DEP_TIME", "DEP_TIME", "WHEELS_OFF"]:

fields[f] = as_utc(fields["FL_DATE"], fields[f], dep_timezone)

for f in ["WHEELS_ON", "CRS_ARR_TIME", "ARR_TIME"]:

fields[f] = as_utc(fields["FL_DATE"], fields[f], arr_timezone)

yield json.dumps(fields)

except KeyError as e:

logging.exception(" Ignoring " + line +

" because airport is not known")

Why FlatMap() instead of Map to call tz_correct()? A Map is a 1-to-1 relation between input and output, whereas a FlatMap() can return 0–N outputs per input. The way it does this is with a Python generator function (i.e., the yield keyword—think of the yield as a return that returns one item at a time until there is no more data to return). Using FlatMap here allows us to ignore any flight information corresponding to unknown airports – even though this doesn’t happen in the historical data we are processing, a little bit of defensive programming doesn’t hurt.

The tz_correct() code gets the departure airport ID from the flight’s data and then looks up the time zone for that airport ID from the airport’s data. After it has the time zone, it calls out to the method as_utc() to convert each of the date–times reported in that airport’s time zone to UTC:

def as_utc(date, hhmm, tzone):

try:

if len(hhmm) > 0 and tzone is not None:

import datetime, pytz

loc_tz = pytz.timezone(tzone)

loc_dt = loc_tz.localize(datetime.datetime.strptime(date,'%Y-%m-%d'),

is_dst=False)

loc_dt += datetime.timedelta(hours=int(hhmm[:2]),

minutes=int(hhmm[2:]))

utc_dt = loc_dt.astimezone(pytz.utc)

return utc_dt.strftime('%Y-%m-%d %H:%M:%S')

else:

return '' # empty string corresponds to canceled flights

except ValueError as e:

print('{} {} {}'.format(date, hhmm, tzone))

raise e

As before, you can run this locally. To do that, run df03.py . A line that originally (in the raw data) looked like

{"FL_DATE":"2015-11-05","UNIQUE_CARRIER":"DL","ORIGIN_AIRPORT_SEQ_ID":"1013503","ORIGIN":"ABE","DEST_AIRPORT_SEQ_ID":"1039705","DEST":"ATL","CRS_DEP_TIME":"0600","DEP_TIME":"0556","DEP_DELAY":-4,"TAXI_OUT":12,"WHEELS_OFF":"0608","WHEELS_ON":"0749","TAXI_IN":10,"CRS_ARR_TIME":"0818","ARR_TIME":"0759","ARR_DELAY":-19,"CANCELLED":false,"DIVERTED":false,"DISTANCE":"692.00"}

now becomes:

{"FL_DATE": "2015-11-05", "UNIQUE_CARRIER": "DL", "ORIGIN_AIRPORT_SEQ_ID": "1013503", "ORIGIN": "ABE", "DEST_AIRPORT_SEQ_ID": "1039705", "DEST": "ATL", "CRS_DEP_TIME": "2015-11-05 11:00:00", "DEP_TIME": "2015-11-05 10:56:00", "DEP_DELAY": -4, "TAXI_OUT": 12, "WHEELS_OFF": "2015-11-05 11:08:00", "WHEELS_ON": "2015-11-05 12:49:00", "TAXI_IN": 10, "CRS_ARR_TIME": "2015-11-05 13:18:00", "ARR_TIME": "2015-11-05 12:59:00", "ARR_DELAY": -19, "CANCELLED": false, "DIVERTED": false, "DISTANCE": "692.00"}

All the times have been converted to UTC. For example, the 0759 time of arrival in Atlanta has been converted to UTC to become 12:59:00.

Correcting Dates

Look carefully at the following line involving a flight from Honolulu (HNL) to Dallas–Fort Worth (DFW). Do you notice anything odd?

{"FL_DATE": "2015-03-06", "UNIQUE_CARRIER": "AA", "ORIGIN_AIRPORT_SEQ_ID": "1217302", "ORIGIN": "HNL", "DEST_AIRPORT_SEQ_ID": "1129803", "DEST": "DFW", "CRS_DEP_TIME": "2015-03-07 05:30:00", "DEP_TIME": "2015-03-07 05:22:00", "DEP_DELAY": -8, "TAXI_OUT": 40, "WHEELS_OFF": "2015-03-07 06:02:00", "WHEELS_ON": "2015-03-06 12:32:00", "TAXI_IN": 7, "CRS_ARR_TIME": "2015-03-06 12:54:00", "ARR_TIME": "2015-03-06 12:39:00", "ARR_DELAY": -15, "CANCELLED": false, "DIVERTED": false, "DISTANCE": "3784.00"}

Examine the departure time in Honolulu and the arrival time in Dallas – the flight is arriving the day before it departed! That’s because the flight date (2015-03-06) is the date of departure in local time. Add in a time difference between airports, and it is quite possible that it is not the date of arrival. We’ll look for these situations and add 24 hours if necessary. This is, of course, quite a hack (have I already mentioned that times ought to be stored in UTC?!):

def add_24h_if_before(arrtime, deptime):

import datetime

if len(arrtime) > 0 and len(deptime) > 0 and arrtime < deptime:

adt = datetime.datetime.strptime(arrtime, '%Y-%m-%d %H:%M:%S')

adt += datetime.timedelta(hours=24)

return adt.strftime('%Y-%m-%d %H:%M:%S')

else:

return arrtime

The 24-hour hack is called just before the yield in tz_correct . 13 Now that we have new data about the airports, it is probably wise to add it to our dataset. Also, as remarked earlier, we want to keep track of the time zone offset from UTC because some types of analysis might require knowledge of the local time. Thus, the new tz_correct code becomes the following:

def tz_correct(line, airport_timezones):

fields = json.loads(line)

try:

# convert all times to UTC

dep_airport_id = fields["ORIGIN_AIRPORT_SEQ_ID"]

arr_airport_id = fields["DEST_AIRPORT_SEQ_ID"]

dep_timezone = airport_timezones[dep_airport_id][2]

arr_timezone = airport_timezones[arr_airport_id][2]

for f in ["CRS_DEP_TIME", "DEP_TIME", "WHEELS_OFF"]:

fields[f], deptz = as_utc(fields["FL_DATE"], fields[f], dep_timezone)

for f in ["WHEELS_ON", "CRS_ARR_TIME", "ARR_TIME"]:

fields[f], arrtz = as_utc(fields["FL_DATE"], fields[f], arr_timezone)

for f in ["WHEELS_OFF", "WHEELS_ON", "CRS_ARR_TIME", "ARR_TIME"]:

fields[f] = add_24h_if_before(fields[f], fields["DEP_TIME"])

fields["DEP_AIRPORT_TZOFFSET"] = deptz

fields["ARR_AIRPORT_TZOFFSET"] = arrtz

yield json.dumps(fields)

except KeyError as e:

logging.exception(" Ignoring " + line + " because airport is not known")

With I run df04.py which has these changes applied to it, the flight from Honolulu to Dallas becomes:

{"FL_DATE": "2015-03-06", "UNIQUE_CARRIER": "AA", "ORIGIN_AIRPORT_SEQ_ID": "1217302", "ORIGIN": "HNL", "DEST_AIRPORT_SEQ_ID": "1129803", "DEST": "DFW", "CRS_DEP_TIME": "2015-03-07 05:30:00", "DEP_TIME": "2015-03-07 05:22:00", "DEP_DELAY": -8, "TAXI_OUT": 40, "WHEELS_OFF": "2015-03-07 06:02:00", "WHEELS_ON": "2015-03-07 12:32:00", "TAXI_IN": 7, "CRS_ARR_TIME": "2015-03-07 12:54:00", "ARR_TIME": "2015-03-07 12:39:00", "ARR_DELAY": -15, "CANCELLED": false, "DIVERTED": false, "DISTANCE": "3784.00", "DEP_AIRPORT_TZOFFSET": -36000.0, "ARR_AIRPORT_TZOFFSET": -21600.0}

As you can see, the dates have now been corrected.

Creating Events

After we have our time-corrected data, we can move on to creating events. We’ll limit ourselves for now to just the departed and arrived messages—we can rerun the pipeline to create the additional events if and when our modeling efforts begin to use other events:

def get_next_event(fields):

if len(fields["DEP_TIME"]) > 0:

event = dict(fields)  # copy

event["EVENT_TYPE"] = "departed"

event["EVENT_TIME"] = fields["DEP_TIME"]

for f in ["TAXI_OUT", "WHEELS_OFF", "WHEELS_ON",

"TAXI_IN", "ARR_TIME", "ARR_DELAY", "DISTANCE"]:

event.pop(f, None)  # not knowable at departure time

yield event

if len(fields["ARR_TIME"]) > 0:

event = dict(fields)

event["EVENT_TYPE"] = "arrived"

event["EVENT_TIME"] = fields["ARR_TIME"]

yield event

Essentially, we pick up the departure time and create a departed event at that time after making sure to remove the fields (such as arrival delay) we cannot know at the departure time. Similarly, we use the arrival time to create an arrived event, as shown in Figure 4-4.

Figure 4-4. Events, when they are published, and some of the fields present in those events.

In the pipeline, the event creation code is called on the flights PCollection after the conversion to UTC has happened:

flights = (pipeline

|'flights:read' >> beam.io.ReadFromText('flights_sample.json')

|'flights:tzcorr' >> beam.FlatMap(tz_correct,

beam.pvalue.AsDict(airports))

)

events = flights | beam.FlatMap(get_next_event)

If we now run the pipeline, 14 we will see two events for each flight:

{"FL_DATE": "2015-04-28", "UNIQUE_CARRIER": "EV", "ORIGIN_AIRPORT_SEQ_ID": "1013503", "ORIGIN": "ABE", "DEST_AIRPORT_SEQ_ID": "1039705", "DEST": "ATL", "CRS_DEP_TIME": "2015-04-28 20:00:00", "DEP_TIME": "2015-04-28 19:55:00", "DEP_DELAY": -5, "CRS_ARR_TIME": "2015-04-28 22:09:00", "CANCELLED": false, "DIVERTED": false, "DEP_AIRPORT_TZOFFSET": -14400.0, "ARR_AIRPORT_TZOFFSET": -14400.0, "EVENT_TYPE": "departed", "EVENT_TIME": "2015-04-28 19:55:00"}

{"FL_DATE": "2015-04-28", "UNIQUE_CARRIER": "EV", "ORIGIN_AIRPORT_SEQ_ID": "1013503", "ORIGIN": "ABE", "DEST_AIRPORT_SEQ_ID": "1039705", "DEST": "ATL", "CRS_DEP_TIME": "2015-04-28 20:00:00", "DEP_TIME": "2015-04-28 19:55:00", "DEP_DELAY": -5, "TAXI_OUT": 7, "WHEELS_OFF": "2015-04-28 20:02:00", "WHEELS_ON": "2015-04-28 21:47:00", "TAXI_IN": 4, "CRS_ARR_TIME": "2015-04-28 22:09:00", "ARR_TIME": "2015-04-28 21:51:00", "ARR_DELAY": -18, "CANCELLED": false, "DIVERTED": false, "DISTANCE": "692.00", "DEP_AIRPORT_TZOFFSET": -14400.0, "ARR_AIRPORT_TZOFFSET": -14400.0, "EVENT_TYPE": "arrived", "EVENT_TIME": "2015-04-28 21:51:00"}

The first event is a departed event and is to be published at the departure time, while the second is an arrived event and is to be published at the arrival time. The departed event has a number of missing fields corresponding to data that is not known at that time.

Once we have this code working, let’s add a third event that will sent when the plane takes off:

if len(fields["WHEELS_OFF"]) > 0:

event = dict(fields)  # copy

event["EVENT_TYPE"] = "wheelsoff"

event["EVENT_TIME"] = fields["WHEELS_OFF"]

for f in ["WHEELS_ON", "TAXI_IN",

"ARR_TIME", "ARR_DELAY", "DISTANCE"]:

event.pop(f, None)  # not knowable at departure time

yield event

At this point, we haven’t created a wheelsdown event yet.

Reading and Writing to the Cloud

So far, we have been reading and writing local files. However, once we start to run our code in production, in a serverless environment, the concept of a “local drive” no longer makes sense. We have to read and write from Cloud Storage. Also, because this is structured data, it is preferable to read and write to BigQuery – recall that we loaded our full dataset into BigQuery in Chapter 2. Now, we’d like to put the transformed (time-corrected) data there as well.

Fortunately, all this involves is changing the source or the sink. The rest of the pipeline stays the same. For example, in the previous section (See 04_streaming/transform/df05.py), we read the airports.csv.gz as:

| 'airports:read' >> beam.io.ReadFromText('airports.csv.gz')

Now, in order to read the equivalent file from Cloud Storage, we change the corresponding code in 04_streaming/transform/df056py to be:

airports_filename = 'gs://{}/flights/airports/airports.csv.gz'.format(

bucket)

…

| 'airports:read' >> beam.io.ReadFromText(airports_filename)

Of course, we’ll have to make sure to upload the file to Cloud Storage and make it readable by whoever is going to run this code. Having the data file be available in our Git repository was not going to scale anyway – Cloud Storage (or BigQuery) is the right place for data.

In df05.py, I used to read a local file that contained the JSON export of a smart part of the dataset and used a JSON parser to obtain a dict:

| 'flights:read' >> beam.io.ReadFromText('flights_sample.json')

| 'flights:parse' >> beam.Map(lambda line: json.loads(line))

In df06.py, the corresponding code becomes simpler because the BigQuery reader returns a dict where the column names of the result set are the keys:

'flights:read' >> beam.io.ReadFromBigQuery(

query='SELECT * FROM dsongcp.flights WHERE rand() < 0.001',

use_standard_sql=True)

Of course when we run it for real, we’ll change the query to remove the sampling (rand() < 0.001) so that we can process the entire dataset.

Similarly, where before we wrote to a local file using:

| 'flights:tostring' >> beam.Map(lambda fields: json.dumps(fields))

| 'flights:out' >> beam.io.textio.WriteToText('all_flights')

we’ll change the code to write to Cloud Storage using:

flights_output = 'gs://{}/flights/tzcorr/all_flights'.format(bucket)

…

| 'flights:tostring' >> beam.Map(lambda fields: json.dumps(fields))

| 'flights:gcsout' >> beam.io.textio.WriteToText(flights_output)

Of course, we can write the same data to a BigQuery table also:

flights_schema = \

'FL_DATE:date,UNIQUE_CARRIER:string,...CANCELLED:boolean'

…

| 'flights:bqout' >> beam.io.WriteToBigQuery(

'dsongcp.flights_tzcorr', schema=flights_schema,

write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,

create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED

)

Note that we need to provide a schema when writing to BigQuery, and specify what to do if the table already exists (we ask for the table to be truncated and contents replaced) and if the table doesn’t already exist (we ask for the table to be created).

We can try running this code, but the pipeline will require a few extra parameters. So where we used to have:

with beam.Pipeline('DirectRunner') as pipeline:

We now need:

argv = [

'--project={0}'.format(project),

'--staging_location=gs://{0}/flights/staging/'.format(bucket),

'--temp_location=gs://{0}/flights/temp/'.format(bucket),

'--runner=DirectRunner'

]

with beam.Pipeline(argv=argv) as pipeline:

The reason is that when we read from BigQuery, we are providing a query:

'flights:read' >> beam.io.ReadFromBigQuery(

query='SELECT * FROM dsongcp.flights WHERE rand() < 0.001',

use_standard_sql=True)

So, we need to provide the project that needs to be billed. In addition, and this is an implementation detail, some temporary data needs to be staged and cached in Cloud Storage and we need to provide the pipeline a place to store this temporary data – we will never be sure which operations will require staging or caching, so it’s a good idea to always specify a scratch location in Cloud Storage for this purpose.

We can run ./df06.py and then check that new tables are created in BigQuery. So far, we have been running the code locally, either on your laptop or in Cloud Shell.

Next, let’s look at how to run this in Cloud Dataflow, which is the GCP managed service for running Apache Beam pipelines.

Running the Pipeline in the Cloud

That last run took a few minutes on the local virtual machine (VM), and we were processing only a thousand lines! Let’s change the code (see df07.py) to process all the rows in the BigQuery view:

'flights:read' >> beam.io.ReadFromBigQuery(

query='SELECT * FROM dsongcp.flights',

use_standard_sql=True)

Now that we have much more data, we need to distribute the work, and to do that, we will change the runner from DirectRunner (which runs locally) to DataflowRunner (which lobs the job off to the cloud and scales it out):

argv = [

'--project={0}'.format(project),

'--job_name=ch04timecorr',

'--save_main_session',

'--staging_location=gs://{0}/flights/staging/'.format(bucket),

'--temp_location=gs://{0}/flights/temp/'.format(bucket),

'--setup_file=./setup.py',

'--max_num_workers=8',

'--region={}'.format(region),

'--runner=DataflowRunner'

]

pipeline = beam.Pipeline(argv=argv)

Notice that there are few extra parameters now:

1. The job name provides the name by which this job will be listed in the GCP console. This is so that we can troubleshoot the job if necessary.

2. We ask the Dataflow submission code to save our main session. This is needed whenever we have global variables in our Python program.

3. The file setup.py should list the Python packages that we needed to install (timezonefinder and pytz) as we went along—Cloud Dataflow will need to install these packages on the Compute Engine instances that it launches behind the scenes:

REQUIRED_PACKAGES = [

'timezonefinder',

'pytz'

]

4. By default, Dataflow autoscales the number of workers based on throughput – the more lines we have in our input data files, the more workers we need. This is called Horizontal Autoscaling . To turn off autoscaling, we can specify -- autoscaling_algorithm=NONE , and to constrain it somewhat, we can specify the maximum number of workers.

5. We specify the region in which the Dataflow pipeline needs to run.

6. The runner is no longer DirectRunner (which runs locally). It is now DataflowRunner.

Running the Python program submits the job to the cloud. Cloud Dataflow autoscales each step of the pipeline based on throughput, and streams the events data into BigQuery (See Figure 4-3). You can monitor the running job on the Cloud Platform Console in the Cloud Dataflow section.

Even as the events data is being written out, we can query it by browsing to the BigQuery console and typing the following:

SELECT

ORIGIN,

DEP_TIME,

DEST,

ARR_TIME,

ARR_DELAY,

EVENT_TIME,

EVENT_TYPE

FROM

dsongcp.flights_simevents

WHERE

(DEP_DELAY > 15 and ORIGIN = 'SEA') or

(ARR_DELAY > 15 and DEST = 'SEA')

ORDER BY EVENT_TIME ASC

LIMIT

5

This returns:

Row

ORIGIN

DEP_TIME

DEST

ARR_TIME

ARR_DELAY

EVENT_TIME

EVENT_TYPE

1

SEA

2015-01-01 08:21:00 UTC

IAD

null

null

2015-01-01 08:21:00 UTC

departed

2

SEA

2015-01-01 08:21:00 UTC

IAD

null

null

2015-01-01 08:38:00 UTC

wheelsoff

3

SEA

2015-01-01 08:21:00 UTC

IAD

2015-01-01 12:48:00 UTC

22.0

2015-01-01 12:48:00 UTC

arrived

4

KOA

2015-01-01 10:11:00 UTC

SEA

2015-01-01 15:45:00 UTC

40.0

2015-01-01 15:45:00 UTC

arrived

5

SEA

2015-01-01 16:43:00 UTC

PSP

null

null

2015-01-01 16:43:00 UTC

departed

As expected, we see three events for the SEA-IAD flight, one at departure, the next at wheelsoff, and the third at arrival. The arrival delay is known only at arrival.

BigQuery is a columnar database, so a query that selects all fields

SELECT

*

FROM

dsongcp.flights_simevents

ORDER BY EVENT_TIME ASC

will be inefficient. However, we do need all of the event data in order to send out event notifications. Therefore, we traded off storage for speed by adding an extra column called EVENT_DATA to our BigQuery table and populated it in our Dataflow pipeline as follows:

def create_event_row(fields):

featdict = dict(fields)  # copy

featdict['EVENT_DATA'] = json.dumps(fields)

return featdict

Then, our query to pull the events could simply be as follows:

SELECT

EVENT_TYPE,

EVENT_TIME,

EVENT_DATA

FROM

dsongcp.flights_simevents

WHERE

EVENT_TIME >= TIMESTAMP('2015-05-01 00:00:00 UTC')

AND EVENT_TIME < TIMESTAMP('2015-05-03 00:00:00 UTC')

ORDER BY

EVENT_TIME ASC

LIMIT

2

The result looks like this:

Row

EVENT_TYPE

EVENT_TIME

EVENT_DATA

1

wheelsoff

2015-05-01 00:00:00 UTC

{"FL_DATE”: “2015-04-30”, “UNIQUE_CARRIER”: “DL”, “ORIGIN_AIRPORT_SEQ_ID”: “1295302”, “ORIGIN”: “LGA”, “DEST_AIRPORT_SEQ_ID”: “1330303”, “DEST”: “MIA”, “CRS_DEP_TIME”: “2015-04-30T23:29:00”, “DEP_TIME”: “2015-04-30T23:35:00”, “DEP_DELAY”: 6.0, “TAXI_OUT”: 25.0, “WHEELS_OFF”: “2015-05-01T00:00:00”, “CRS_ARR_TIME”: “2015-05-01T02:53:00”, “CANCELLED”: false, “DIVERTED”: false, “DEP_AIRPORT_TZOFFSET”: -14400.0, “ARR_AIRPORT_TZOFFSET”: -14400.0, “EVENT_TYPE”: “wheelsoff”, “EVENT_TIME”: “2015-05-01T00:00:00"}

2

departed

2015-05-01 00:00:00 UTC

{"FL_DATE”: “2015-04-30”, “UNIQUE_CARRIER”: “DL”, “ORIGIN_AIRPORT_SEQ_ID”: “1295302”, “ORIGIN”: “LGA”, “DEST_AIRPORT_SEQ_ID”: “1320402”, “DEST”: “MCO”, “CRS_DEP_TIME”: “2015-04-30T23:55:00”, “DEP_TIME”: “2015-05-01T00:00:00”, “DEP_DELAY”: 5.0, “CRS_ARR_TIME”: “2015-05-01T02:45:00”, “CANCELLED”: false, “DIVERTED”: false, “DEP_AIRPORT_TZOFFSET”: -14400.0, “ARR_AIRPORT_TZOFFSET”: -14400.0, “EVENT_TYPE”: “departed”, “EVENT_TIME”: “2015-05-01T00:00:00"}

This table will serve as the source of our events; it is from such a query that we will simulate streaming flight data.

Publishing an Event Stream to Cloud Pub/Sub

Now that we have the source events from the raw flight data, we are ready to simulate the stream. Streaming data in Google Cloud Platform is typically published to Cloud Pub/Sub, a serverless real-time messaging service. Cloud Pub/Sub provides reliable delivery and can scale to more than a million messages per second. Unless you are using Cloud Pub/Sub Lite (which is a single-zone service that is built for low cost operation), Pub/Sub stores copies of messages in multiple zones to provide “at least once” guaranteed delivery to subscribers, and there can be many simultaneous subscribers.

Our simulator will read from the events table in BigQuery (populated in the previous section) and publish messages to Cloud Pub/Sub. Essentially, we will walk through the flight event records, getting the notification time from each, and simulate publishing those events as they happen.

Speed-up Factor

However, we’ll also use a mapping between the event notification time (arrival or departure time based on event) and the current system time. Why? Because it is inefficient to always simulate the flight events at real-time speeds. Instead, we might want to run through a day of flight data in an hour (as long as the code that processes these events can handle the increased data rate). At other times, we may be running our event-processing code in a debugging environment that is slower and so we might want to slow down the simulation. I will refer to this ratio between the actual time and simulation time as the speed-up factor —the speed-up factor will be greater than 1 if we want the simulation to be faster than real time, and less than 1 if we want it to be slower than real time.

Based on the speed-up factor, we’ll have to do a linear transformation of the event time to system time. If the speed-up factor is 1, a 60-minute difference between the start of the simulation in event time and the current record’s timestamp should be encountered 60 minutes after the start of the simulation. If the speed-up factor is 60, a 60-minute difference in event time translates to a 1-minute difference in system time, and so the record should be published a minute later. If the event time clock is ahead of the system clock, we sleep for the necessary amount of time so as to allow the simulation to catch up.

The simulation consists of four steps (see also Figure 4-5):

1. Run the query to get the set of flight event records to publish.

2. Iterate through the query results.

3. Accumulate events to publish as a batch.

4. Publish accumulated events and sleep as necessary.

Even though this is an ETL pipeline, the need to process records in strict sequential order and sleep in between makes this ETL pipeline a poor fit for Cloud Dataflow. Instead, we’ll implement this as a pure Python program. The problem with this choice is that the simulation code is not fault tolerant—if the simulation fails, it will not automatically restart and definitely will not start from the last successfully notified event.

Figure 4-5. The four steps of simulation

The simulation code that we are writing is only for quick experimentation with streaming data. Hence, I will not take the extra effort needed to make it fault-tolerant. If we had to do so, we could make the simulation fault-tolerant by starting from a BigQuery query that is bounded in terms of a time range with the start of that time range automatically inferred from the last-notified record in Cloud Pub/Sub. Then, we could launch the simulation script from a Docker container and use Cloud Run or Google Container Engine (which runs Kubernetes) to automatically restart the simulation if the simulation code fails.

Get Records to Publish

The BigQuery query is parameterized by the start and end time of the simulation and can be invoked through the Google Cloud API for Python (see 04_streaming/transform/simulate.py in the Git repository):

bqclient = bq.Client(args.project)

querystr = """

SELECT

EVENT_TYPE,

EVENT_TIME AS NOTIFY_TIME,

EVENT_DATA

FROM

dsongcp.flights_simevents

WHERE

EVENT_TIME >= TIMESTAMP('{}')

AND EVENT_TIME < TIMESTAMP('{}')

ORDER BY

EVENT_TIME ASC

"""

rows = bqclient.query(querystr.format(args.startTime,

args.endTime))

This, however, is a bad idea. Do you see why?

It’s because we are getting the start time and end time from the command-line of the simulation script and directly passing it into BigQuery. This is called SQL injection , and can lead to security problems 15 . A better approach is to use parameterized queries – the BigQuery query contains the parameters marked as @startTime etc. and the Python query function takes the definitions via the job configuration parameter:

bqclient = bq.Client(args.project)

querystr = """

SELECT

EVENT_TYPE,

EVENT_TIME AS NOTIFY_TIME,

EVENT_DATA

FROM

dsongcp.flights_simevents

WHERE

EVENT_TIME >= @startTime

AND EVENT_TIME < @endTime

ORDER BY

EVENT_TIME ASC

"""

job_config = bq.QueryJobConfig(

query_parameters=[

bq.ScalarQueryParameter("startTime", "TIMESTAMP", args.startTime),

bq.ScalarQueryParameter("endTime", "TIMESTAMP", args.endTime),

]

)

rows = bqclient.query(querystr, job_config=job_config)

The query function returns an object (called rows in the above snippet) that we can iterate through:

for row in rows:

# do something

What do we need to do for each of the rows? We’ll need to iterate through the records, build a batch of events, and publish each batch. Let’s see how each of these steps is done.

Iterating Through Records

As we walk through the query results, we need to publish events to Cloud Pub/Sub. There is a separate topic per event type (i.e., an arrived topic, a departed topic, and a wheelsoff topic), so we create three topics: 16

for event_type in ['wheelsoff', 'arrived', 'departed']:

topics[event_type] = publisher.topic_path(args.project, event_type)

try:

publisher.get_topic(topic=topics[event_type])

logging.info("Already exists: {}".format(topics[event_type]))

except:

logging.info("Creating {}".format(topics[event_type]))

publisher.create_topic(name=topics[event_type])

After creating the topics, we call the notify() method passing along the rows read from BigQuery:

# notify about each row in the dataset

programStartTime = datetime.datetime.utcnow()

simStartTime = datetime.datetime.strptime(args.startTime,

TIME_FORMAT).replace(tzinfo=pytz.UTC)

notify(publisher, topics, rows, simStartTime,

programStartTime, args.speedFactor)

Building a Batch of Events

The notify() method consists of accumulating the rows into batches and publishing them when it is time to sleep:

def notify(publisher, topics, rows, simStartTime, programStart, speedFactor):

# sleep computation

def compute_sleep_secs(notify_time):

time_elapsed = (datetime.datetime.utcnow() -

programStart).seconds

sim_time_elapsed = (notify_time - simStartTime).seconds / speedFactor

to_sleep_secs = sim_time_elapsed - time_elapsed

return to_sleep_secs

tonotify = {}

for key in topics:

tonotify[key] = list()

for row in rows:

event, notify_time, event_data = row

# how much time should we sleep?

if compute_sleep_secs(notify_time) > 1:

# notify the accumulated tonotify

publish(publisher, topics, tonotify, notify_time)

for key in topics:

tonotify[key] = list()

# recompute sleep, since notification takes a while

to_sleep_secs = compute_sleep_secs(notify_time)

if to_sleep_secs > 0:

logging.info('Sleeping {} seconds'.format(to_sleep_secs))

time.sleep(to_sleep_secs)

tonotify[event].append(event_data)

# left-over records; notify again

publish(publisher, topics, tonotify, notify_time)

There are a few points to be made here. First is that we work completely in UTC so that the time difference computations make sense. Second, we always compute whether to sleep by looking at the time difference since the start of the simulation. If we simply keep moving a pointer forward, errors in time will accumulate. Finally, note that we check whether the sleep time is more than a second the first time, so as to give records time to accumulate. If, when you run the program, you do not see any sleep, your speed-up factor is too high for the capability of the machine running the simulation code and the network between that machine and Google Cloud Platform. Slow down the simulation, get a larger machine, or run it behind the Google firewall (such as in Cloud Shell or on a Compute Engine instance).

Publishing a Batch of Events

The notify() method that we saw in the previous code example has accumulated the events in between sleep calls. Even though it appears that we are publishing one event at a time, the publisher actually maintains a separate batch for each topic:

def publish(publisher, topics, allevents):

for key in topics:  # 'departed', 'arrived', etc.

topic = topics[key]

events = allevents[key]

logging.info('Publishing {} {} events'.format(len(events), key))

for event_data in events:

publisher.publish(topic, event_data.encode())

Note that Cloud Pub/Sub does not guarantee the order in which messages will be delivered, especially if the subscriber lets a huge backlog build up. Out-of-order messages will happen, and downstream subscribers will need to deal with them. Cloud Pub/Sub guarantees “at-least once” delivery and will resend the message if the subscriber does not acknowledge a message in time. I will use Cloud Dataflow to ingest from Cloud Pub/Sub, and Cloud Dataflow deals with both these issues (out-of-order and duplication) transparently.

We can try out the simulation by typing the following:

python3 simulate.py --startTime '2015-05-01 00:00:00 UTC' \

--endTime '2015-05-04 00:00:00 UTC' --speedFactor=60

This will simulate three days of flight data (the end time is exclusive) at 60 times realtime speed and stream the events into three topics on Cloud Pub/Sub. Because the simulation starts off from a BigQuery query, it is quite straightforward to limit the simulated events to just a single airport or to airports within a latitude/longitude bounding box.

In this section, we looked at how to produce an event stream and publish those events in real time. Throughout this book, we can use this simulator and these topics for experimenting with how to consume streaming data and carry out real-time analytics.

Real-Time Stream Processing

Now that we have a source of streaming data that includes location information, let’s look at how to build a real-time dashboard. Figure 4-6 presents the reference architecture for many solutions on Google Cloud Platform. 17

Figure 4-6. Reference architecture for data processing on Google Cloud Platform

In the previous section, we set up a real-time stream of events into Cloud Pub/Sub that we can aggregate in Cloud Dataflow and write to BigQuery. Data Studio can connect to BigQuery and provide a real-time, interactive dashboard. Let’s get started.

Streaming in Dataflow

We used Beam/Dataflow in Python in the previous chapter because it was an ETL pipeline without any performance requirements. When we carried out the time correction of the raw flight data, we were working off a complete BigQuery flights table, processing them in Cloud Dataflow and writing the events table into BigQuery. Processing a finite, bounded input dataset is called batch processing.

Here, though, we need to process events in Cloud Pub/Sub that are streaming in. The dataset is unbounded. Processing an unbounded set of data is called stream processing. Fortunately, the code to do stream processing in Apache Beam is identical to the code to do batch processing.

We could simply receive the events from Cloud Pub/Sub very similar to how we read data from a CSV file:18

topic_name = "projects/{}/topics/arrived".format(project)

events = (pipeline

| 'read' >> beam.io.ReadFromPubSub(topic=topic_name)

| 'parse' >> beam.Map(lambda s: json.loads(s))

)

The only change we have to do is to turn on the streaming flag in the Dataflow options:

argv = [

…

'--streaming',

]

We can stream the read-in events to BigQuery using code similar to what we used in batch processing:

schema = 'FL_DATE:date,...,EVENT_TYPE:string,EVENT_TIME:timestamp'

(events

| 'bqout' >> beam.io.WriteToBigQuery(

'dsongcp.streaming_events', schema=schema,

create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED

)

)

In the preceding code, we subscribe to a topic in Cloud Pub/Sub and begin reading from it. As each message streams in, we convert it to a TableRow in BigQuery and then write it out. Indeed, if this is all we need, we can simply use the Google-provided Dataflow template that goes from Pub/Sub to BigQuery .

But let’s say that we want to read both the arrived events and the departed events and write them to the same BigQuery table. We can do that quite simply in Beam:

events = {}

for event_name in ['arrived', 'departed']:

topic_name = "projects/{}/topics/{}".format(project, event_name)

events[event_name] = (pipeline

| 'read:{}'.format(event_name) >>

beam.io.ReadFromPubSub(topic=topic_name)

| 'parse:{}'.format(event_name) >> beam.Map(

lambda s: json.loads(s))

)

all_events = (events['arrived'], events['departed']) | beam.Flatten()

Flattening the two sets of events concatenates them into a single collection. We then write out all_events to BigQuery.

To try this code out, we need to run the simulator we wrote in the previous section so that the simulator can publish events to the Pub/Sub topics. To start the simulation, start the Python simulator that we developed in the previous section:

python simulate.py --startTime ’2015-05-01 00:00:00 UTC’

--endTime ’2015-05-04 00:00:00 UTC’ --speedFactor 30

The simulator will send events from May 1, 2015, to May 3, 2015, at 30 times realtime speed, so that an hour of data is sent to Cloud Pub/Sub in two minutes. You can do this from CloudShell or from your local laptop. (If necessary, run install_packages.sh to install the necessary Python packages and gcloud auth application-default login to give the application the necessary credentials to execute queries.)

In another terminal, start avg01.py to read the stream of events and write them out to BigQuery. We can then query the dataset in BigQuery even as the events are streaming in. The BigQuery UI may not even show this streaming table yet, but it can be queried:

SELECT * FROM dsongcp.streaming_events

ORDER BY EVENT_TIME DESC

LIMIT 5

Windowing a pipeline

Although we could do just a straight data transfer, I’d like to do more. When I populate a real-time dashboard of flight delays, I’d like the information to be aggregated over a reasonable interval—for example, I want a moving average of flight delays and the total number of flights over the past 60 minutes at every airport. So, rather than simply take the input received from Cloud Pub/Sub and stream it out to BigQuery, I’d like to carry out time-windowed analytics on the data as I’m receiving it and write those analytics19 to BigQuery. Cloud Dataflow can help us do this.

While we may be averaging over 60 minutes, how often should we compute this 60-minute average? It might be advantageous, for example, to use a sliding window and compute this 60-minute average every five minutes.

Streaming aggregation

The key difference between batch aggregation and streaming aggregation is the unbounded nature of the data in stream processing. What does an operation like “max” mean when the data is unbounded? After all, whatever our maximum at this point in time, a large value could come along in the stream at a later point.

A key concept when aggregating streaming data is that of a window that becomes the scope for all aggregations. Here, we apply a time-based sliding window on the pipeline. From now on, all grouping, aggregation, and so on is within that time window and there is a separate maximum, average, etc. in each time window:

stats = (all_events

| 'byairport' >> beam.Map(by_airport)

| 'window' >> beam.WindowInto(

beam.window.SlidingWindows(60 * 60, 5 * 60))

| 'group' >> beam.GroupByKey()

| 'stats' >> beam.Map(lambda x: compute_stats(x[0], x[1]))

)

Let’s walk through the above code snippet carefully.

The first thing we do is to take all the events and apply the by_airport transformation to the events. What this does is to pull out the origin airport for departed events and destination airport for arrival events:

def by_airport(event):

if event['EVENT_TYPE'] == 'departed':

return event['ORIGIN'], event

else:

return event['DEST'], event

Next, we apply a sliding window to the event stream. The window is of 60 minutes duration, applied every 5 minutes.

Then, we apply a GroupByKey.

What’s the key?

In the by_airport function above, we made the airport the key and the entire event object the value. So, the GroupByKey groups events by airport.

But the GroupByKey is not just by airport. Because we have already applied a sliding window, there is a separate group created for each window. So, each group now consists of 60 minutes of flight events that arrived or departed at a specific airport.

It is on these groups that we call the compute_stats function in the last Map of the snippet. The compute_stats takes the airport and list of events at that airport and computes stats:

def compute_stats(airport, events):

arrived = [event['ARR_DELAY'] for event in events

if event['EVENT_TYPE'] == 'arrived']

avg_arr_delay = float(np.mean(arrived))

if len(arrived) > 0 else None

departed = [event['DEP_DELAY'] for event in events

if event['EVENT_TYPE'] == 'departed']

avg_dep_delay = float(np.mean(departed))

if len(departed) > 0 else None

num_flights = len(events)

start_time = min([event['EVENT_TIME'] for event in events])

latest_time = max([event['EVENT_TIME'] for event in events])

return {

'AIRPORT': airport,

'AVG_ARR_DELAY': avg_arr_delay,

'AVG_DEP_DELAY': avg_dep_delay,

'NUM_FLIGHTS': num_flights,

'START_TIME': start_time,

'END_TIME': latest_time

}

In the code above, we pull out the arrived events and compute the average arrival delay. Similarly, we compute the average departure delay on the departed events. We also compute the number of flights in the time window at this airport and return all these statistics.

The statistics are then written out to BigQuery using code that should look familiar by now:

stats_schema = ','.join(

['AIRPORT:string,AVG_ARR_DELAY:float,AVG_DEP_DELAY:float',

'NUM_FLIGHTS:int64,START_TIME:timestamp,END_TIME:timestamp'])

(stats

| 'bqout' >> beam.io.WriteToBigQuery(

'dsongcp.streaming_delays', schema=stats_schema,

create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED

)

)

As with the previous section, we can start the simulator, and then start avg02.py. When I did this, though, the resulting aggregations were getting produced every 5 minutes, but in that 5 minutes, the 30x simulation had notified about events in 150 minutes!

The stream processing engine was applying the sliding windows based on the time on a wall clock. We, however, want it to apply the window based on the timestamp within the images.

How do we do that?

Using Event Timestamps

What we have to do is to add an attribute at the time we publish the events (in simulate.py):

publisher.publish(topic, event_data.encode(), EventTimeStamp=timestamp)

Then, in our Beam pipeline, when read from Pub/Sub, we should tell the pipeline to disregard the publish time in Pub/Sub and use this attribute of the message as the timestamp instead:

| 'read:{}'.format(event_name) >> beam.io.ReadFromPubSub(

topic=topic_name, timestamp_attribute='EventTimeStamp')

With this change, when I run the query:

SELECT * FROM dsongcp.streaming_delays

WHERE AIRPORT = 'ATL'

ORDER BY END_TIME DESC

I get rows approximately 5 minutes apart as expected:

Row

AIRPORT

AVG_ARR_DELAY

AVG_DEP_DELAY

NUM_FLIGHTS

START_TIME

END_TIME

1

ATL

35.72222222222222

13.666666666666666

48

2015-05-01 02:24:00 UTC

2015-05-01 03:21:00 UTC

2

ATL

35.25

8.717948717948717

59

2015-05-01 02:15:00 UTC

2015-05-01 03:12:00 UTC

3

ATL

38.666666666666664

9.882352941176471

52

2015-05-01 02:19:00 UTC

2015-05-01 03:12:00 UTC

4

ATL

38.473684210526315

5.916666666666667

55

2015-05-01 02:15:00 UTC

2015-05-01 03:08:00 UTC

5

ATL

35.111111111111114

5.53125

50

2015-05-01 02:15:00 UTC

2015-05-01 03:03:00 UTC

The reported times are not exactly 5 minutes apart because the reported times correspond to the earliest/latest flight in Atlanta within the time window. Note also that the length of the time window is approximately an hour.

It is likely, however, that Cloud Shell or your local laptop will struggle to keep up with the event stream. We need to be executing this pipeline in Dataflow in a serverless way.

Executing the Stream Processing

To run the Beam pipeline in Cloud Dataflow, all I have to do is to change the runner (see avg03.py in the course repository):

argv = [

'--project={0}'.format(project),

'--job_name=ch04avgdelay',

'--streaming',

…

'--runner=DirectRunner'

]

Before we start this pipeline, though, it is a good idea to delete the rows already written to the BigQuery table by avg02.py in the previous section. The easiest way to do this is to run the following SQL DML command to truncate the table:

TRUNCATE TABLE dsongcp.streaming_delays

Running avg03.py will launch off a Dataflow job. If you now browse to the Cloud Platform console, to the Cloud Dataflow section, you will see that a new streaming job has started and that the pipeline looks like that shown in Figure 4-7.

Figure 4-7. The streaming pipeline to compute delay statistics in real-time at each airport

The pipeline processes flight events as they stream into Pub/Sub, aggregates them into time windows, and streams the resulting statistics into BigQuery.

Analyzing Streaming Data in BigQuery

Two minutes 20 after the launch of your program, the first set of data will make it into BigQuery. You can query for the statistics for a specific airport from the BigQuery console using the same query as before:

SELECT * FROM dsongcp.streaming_delays

WHERE AIRPORT = 'ATL'

ORDER BY END_TIME DESC

The cool thing is that we can do this querying even as the data is streaming! How would we get the latest data for all airports? We could get all the data for each airport, order it by time and take the latest:

SELECT

AIRPORT,

ARRAY_AGG(

STRUCT(AVG_ARR_DELAY, AVG_DEP_DELAY, NUM_FLIGHTS, END_TIME)

ORDER BY END_TIME DESC LIMIT 1) AS a

FROM dsongcp.streaming_delays d

GROUP BY AIRPORT

The results look something like this:

Row

AIRPORT

a.AVG_ARR_DELAY

a.AVG_DEP_DELAY

a.NUM_FLIGHTS

a.END_TIME

1

BUR

-6.8

-5.666666666666667

8

2015-05-01 03:26:00 UTC

2

HNL

17.11111111111111

-3.7777777777777777

18

2015-05-01 03:46:00 UTC

3

CVG

-7.75

null

4

2015-05-01 03:48:00 UTC

4

PHL

5.636363636363637

16.5

13

2015-05-01 03:48:00 UTC

5

IND

40.6

null

5

2015-05-01 03:45:00 UTC

Queries like these on streaming data will be useful when we begin to build our dashboard. For example, the first query will allow us to build a time–series chart of delays at a specific airport. The second query will allow us to build a map of average delays across the country.

Real-Time Dashboard

Now that we have streaming data in BigQuery and a way to analyze it as it is streaming in, we can create a dashboard that shows departure and arrival delays in context. Two maps can help explain our contingency table–based model to end users: current arrival delays across the country, and current departure delays across the country.

To pull the data to populate these charts, we need to add a BigQuery data source in Data Studio. Although Data Studio supports specifying the query directly in the user interface, it is much better to create a view in BigQuery and use that view as a data source in Data Studio. BigQuery views have a few advantages over queries that you type into Data Studio: they tend to be reusable across reports and visualization tools, there is only one place to change if an error is detected, and BigQuery views map better to access privileges (Cloud Identity Access Management roles) based on the columns they need to access.

Here is the query that I used to create the view:

CREATE OR REPLACE VIEW dsongcp.airport_delays AS

WITH delays AS (

SELECT d.*, a.LATITUDE, a.LONGITUDE

FROM dsongcp.streaming_delays d

JOIN dsongcp.airports a USING(AIRPORT)

WHERE a.AIRPORT_IS_LATEST = 1

)

SELECT

AIRPORT,

CONCAT(LATITUDE, ',', LONGITUDE) AS LOCATION,

ARRAY_AGG(

STRUCT(AVG_ARR_DELAY, AVG_DEP_DELAY, NUM_FLIGHTS, END_TIME)

ORDER BY END_TIME DESC LIMIT 1) AS a

FROM delays

GROUP BY AIRPORT, LONGITUDE, LATITUDE

This is slightly different from the second query in the previous section in that it also adds the location of the airport by joining against the airports table.

Having saved the view in BigQuery, we can create a data source for the view in Data Studio, just as we did in the previous chapter:

1. Visit https://datastudio.google.com

2. Create a BigQuery data source, point it to the airport_delays view, and connect to it.

3. Change the location field from Text to a Geo | Latitude, Longitude, then create report.

4. Add a “Geo Chart” to the report.

5. Specify the location field as the geo dimension (see Figure 4-8)

6. Specify average departure delay as the dimension, and United States as the zoom level.

7. Change the style so that the color bar goes from green to red through white.

8. Repeat for the arrival delay

Figure 4-8. Dashboard of latest flight data from across the United States

It is worth reflecting on what we did in this section. We processed streaming data in Cloud Dataflow, creating 60-minute moving averages that we streamed into BigQuery. We then created a view in BigQuery that would show the latest data for each airport, even as it was streaming in. We connected that to a dashboard in Data Studio. Every time the dashboard is refreshed, it pulls new data from the view, which in turn dynamically reflects the latest data in BigQuery.

Summary

In this chapter, we discussed how to build a real-time analysis pipeline to carry out streaming analytics and populate real-time dashboards. In this book, we are using a dataset that is not available in real time. Therefore, we simulated the creation of a real-time feed so that I could demonstrate how to build a streaming ingest pipeline. Building the simulation also gives us a handy test tool—no longer do we need to wait for an interesting event to happen. We can simply play back a recorded event!

In the process of building out the simulation, we realized that time handling in the original dataset was problematic. Therefore, we improved the handling of time in the original data and created a new dataset with UTC timestamps and local offsets. This is the dataset that we will use going forward.

We also looked at the reference architecture for handling streaming data in Google Cloud Platform. First, receive your data in Cloud Pub/Sub so that the messages can be received asynchronously. Process the Cloud Pub/Sub messages in Cloud Dataflow, computing aggregations on the data as needed, and stream either the raw data or aggregate data (or both) to BigQuery. We worked with all three Google Cloud Platform products (Cloud Pub/Sub, Cloud Dataflow, and BigQuery) using the Google Cloud Platform client libraries in Python. However, in none of these cases did we ever need to create a virtual machine ourselves—these are all serverless and autoscaled offerings. We thus were able to concentrate on writing code, letting the platform manage the rest.

1   Note that this is a common situation. It is only as you start to explore a dataset that you discover you need ancillary datasets. Had I known beforehand, I would have ingested both datasets. But you are following my workflow, and as of this point, I knew that I needed a dataset of timezone offsets but hadn’t yet searched for it!

2   See 04_streaming/design/mktbl.sh

3   Or make a copy or view of the table with anonymized column values—we cover safeguarding personally identifiable information in Chapter 7.

4   For example, the time zone of Sevastopol changed in 2014 from Eastern European Time (UTC+2) to Moscow Time (UTC+4) after the annexation of Crimea by the Russian Federation.

5   For example, is there a spike associated with traffic between 5 PM and 6 PM local time?

6   The Java API is much more mature and performant, but Python is easier and more concise. In this book, we will use both.

7   This code is in 04_streaming/transform/df01.py of the GitHub repository of this book.

8   This code is in 04_streaming/transform/df02.py of the GitHub repository of this book.

9   See the answer to the question “How do I handle NameErrors?” at https://cloud.google.com/dataflow/faq.

10   Normally, the recommended way to sample a BigQuery table is to do SELECT * FROM dsongcp.flights WHERE TABLESAMPLE SYSTEM (0.001) because table sampling isn’t cached and so, we will get different results each time. However, at the time of writing, table sampling works only on tables and flights is a View. Besides, in our current use case, we don’t care whether or not we get different samples each time we run the query. That’s why I’m using rand().

11   See the file 04_streaming/transform/bqsample.sh

12   This code is in 04_streaming/transform/df03.py of the GitHub repository of this book.

13   This code is in 04_streaming/transform/df04.py of the GitHub repository of this book.

14   This code is in 04_streaming/transform/df05.py of the GitHub repository of this book.

15   It opens a door to someone passing in queries that could, for example, delete a table

16   See 04_streaming/simulate/simulate.py

17   For an example, go to https://cloud.google.com/solutions/mobile/mobile-gaming-analysis-telemetry

18   See 04_streaming/realtime/avg01.py

19   If you wanted to write the raw data that is received to BigQuery, you could do that, too, of course—that is what is shown in the previous code snippet. In this section, I assume that we need only the aggregate statistics over the past hour.

20   Recall that we are computing aggregates over 60 minutes every 5 minutes. Cloud Dataflow treats the first “full” window as happening 65 minutes into the simulation. Because we are simulating at 30 times speed, this is two minutes on your clock.

About the Author

Valliappa (Lak) Lakshmanan is the director of analytics and AI solutions at Google Cloud, where he leads a team building cross-industry solutions to business problems. His mission is to democratize machine learning so that it can be done by anyone anywhere. Lak is the author or coauthor of Practical Machine Learning for Computer Vision, Machine Learning Design Patterns, Data Governance The Definitive Guide, Google BigQuery The Definitive Guide, and Data Science on the Google Cloud Platform.

Table of Contents

1. Making Better Decisions Based on Data

Many Similar Decisions

The Role of Data Scientists

Scrappy Environment

Full Stack Cloud Data Scientists

Collaboration

Target audience for the book

Best Practices

Simple to Complex Solutions

Cloud Computing

Serverless

A Probabilistic Decision

Probabilistic Approach

Probability Density Function

Cumulative Distribution Function

Data and Tools

Getting Started with the Code

Summary

2. Ingesting Data into the Cloud

Airline On-Time Performance Data

Knowability

Training–Serving Skew

Downloading Data

Hub and Spoke Architecture

Dataset Fields

Separation of Compute and Storage

Scaling Up

Scaling Out with Sharded Data

Scaling out with Data in Situ

Ingesting Data

Reverse Engineering a Web Form

Dataset Download

Exploration and Cleanup

Uploading Data to Google Cloud Storage

Loading Data into Google BigQuery

Advantages of a Serverless Columnar Database

Staging on Cloud Storage

Access Control

Ingesting CSV Files

Partitioning

Scheduling Monthly Downloads

Ingesting in Python

Cloud Run

Securing Cloud Run

Deploying and Invoking Cloud Run

Scheduling Cloud Run

Summary

Code Break

3. Creating Compelling Dashboards

Explain Your Model with Dashboards

Why Build a Dashboard First?

Accuracy, Honesty, and Good Design

Loading Data into Cloud SQL

Create a Google Cloud SQL Instance

Create Table of Data

Interacting with the database

Querying Using BigQuery

Schema Exploration

Using Preview

Using Table Explorer

Creating BigQuery View

Building Our First Model

Contingency Table

Threshold Optimization

Building a Dashboard

Getting Started with Data Studio

Creating Charts

Adding End-User Controls

Showing Proportions with a Pie Chart

Explaining a Contingency Table

Summary

4. Streaming Data: Publication and Ingest with Pub/Sub and Dataflow

Designing the Event Feed

Transformations Needed

Architecture

Getting airport information

Sharing data

Time Correction

Apache Beam/Cloud Dataflow

Parsing Airports Data

Adding Time Zone Information

Converting Times to UTC

Correcting Dates

Creating Events

Reading and Writing to the Cloud

Running the Pipeline in the Cloud

Publishing an Event Stream to Cloud Pub/Sub

Speed-up Factor

Get Records to Publish

Iterating Through Records

Building a Batch of Events

Publishing a Batch of Events

Real-Time Stream Processing

Streaming in Dataflow

Windowing a pipeline

Streaming aggregation

Using Event Timestamps

Executing the Stream Processing

Analyzing Streaming Data in BigQuery

Real-Time Dashboard

Summary