Kensu

Fundamentals of Data Observability

Implement Trustworthy End-to-End
Data Solutions

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.

Andy Petrella

Fundamentals of Data Observability

by Andy Petrella

Copyright © 2023 O’Reilly Media. 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.

Acquisitions Editor: Jessica Haberman

Development Editor: Gary O’Brien

Production Editor: Ashley Stussy

Interior Designer: David Futato

Cover Designer: Karen Montgomery

Illustrator: Kate Dullea

July 2023: First Edition

Revision History for the Early Release

2022-06-14: First Release

2022-08-01: Second Release

2022-09-21: Third Release

2023-02-21: Fourth Release

2023-05-02: Fifth Release

2023-05-26: Sixth Release

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

The O’Reilly logo is a registered trademark of O’Reilly Media, Inc. Fundamentals of Data Observability, 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 and do not represent the publisher’s views. While the publisher and the author have used good faith efforts to ensure that the information and instructions contained in this work are accurate, the publisher and the author 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-13323-8

Chapter 1. Introducing Data Observability

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 author at gobrien@oreilly.com.

Over the last hundred years, we’ve seen data literally take shape and transform just about everything we do, starting as a strategic asset and now becoming the core strategy itself.

Although they used to call themselves “data-driven” companies (and many still do), in a sense, all organizations aim to become data companies. Data is perceived as an asset, a product that can generate value for the business. Being “data-driven” is no longer enough.

For an organization to become a data company, they need to consider several transformations. One of them is their ability to scale their capacity to operationalize the generation of value from data - this has led to identifying a market gap called data operations (DataOps).

Improving data operations means you have to reduce the time to market – the time between discovering the potential value in data and delivering this value to consumers, a.k.a time to value. Then you have to maintain this value at the lowest possible cost – the resources needed to keep the existing consumers satisfied and extend the new ones. This is linked to the total cost of ownership, and one of the key factors is the cost of quality assurance – i.e., processes that maintain data quality.

Managing data quality is the most important barrier for organizations to scale data strategies due to a lack of ability to identify issues and remediate them appropriately.

Data quality is not a new topic. What is new is the dramatic shift in the scale and complexity of data, essentially invalidating older methods and best practices introduced before the introduction of DataOps. Only even considering tools such as notebook deployment in Databricks or dbt, integrated with automated tests, the time to deliver data projects has shrunk so that data teams can create data much faster. Consequently, the number of data projects continually grows, as also their complexity. Therefore when a data issue is detected, finding its cause is so challenging that the only option is to patch the data, which disrupts the data value chain, and generates debts.

Data Observability introduces new approaches, best practices, and capabilities that are needed to create the necessary visibility into data for companies to scale their data ambitions.

Before I dig further into what data observability is and offers at scale, let’s first look at how data teams are evolving and identify the challenges they face.

The evolution of data teams

To support the scaling of data value, data teams have started to play a critical role in the implementation of incorporating data into business operations.

This evolution of data teams is similar to how IT teams evolved in the 1950s¹—when dedicated teams and roles were created to incorporate computing into business operations.

Looking at Google Trends (Figure 1-1) of the term “data team” from 2004 to 2020, we can see that while data teams have always been around, the interest in data teams began increasing in 2014 and accelerated significantly between 2014 to 2020.

Figure 1-1. Fig 1. Google search trend for “data team.”

This is even though interest in “Big Data” began to decrease (as shown in Figure 1-2).

Figure 1-2. Fig. 2. Analysis of “Big Data” search term on Google Trends²

Obviously, this doesn’t mean that Big Data was not needed anymore, but, as Figure 1-3 shows, in 2014 the focus started to shift toward Data Science because its link with value-generation was more intuitive.

Figure 1-3. Fig 3. Google trends for search terms “data engineer” and “data scientist”.

However, even as interest began to climb in data science, it was clear that data availability was a roadblock to many data science projects.

Data science teams have not replaced big data teams. Instead, data teams have embraced analytics, and thus added this relatively new role of Data Scientist.

As data and analytics became more central to companies’ success, providing the right data to the right people remained a constant challenge. Gartner noted that starting in 2018, data engineers had become crucial in addressing data accessibility challenges and that data and analytic leaders must, therefore, “develop a data engineering discipline as part of their data management strategy.”

Hence, it became evident that a role dedicated to producing the data for downstream use cases was missing. This is why, since around 2020, as companies began to bring on engineers specifically to help build data pipelines and bring together data from different source systems, the search volume for data engineers has increased significantly. Today, as also shown by Figure 1-3, data engineering is trending toward catching up to data scientist in search popularity.

Note

Data Availability is one of the metrics driving Data Observability. There are important metrics and metadata relating to data availability, such as the time to live (TTL) of a data set, and the refresh rate (freshness) of a dataset. Still, there are also nuances we’ll get into throughout the book such as sourcing data, privacy and security ramifications, etc. Although, separating roles allows the different skills to build and the resources to be focused on the different phases of data projects.

Search trends are just one example that illustrates the creation and scaling of data teams. Additionally, these search trends highlight how these data teams started to become more organized, with members specializing in the scope of responsibility to implement new products and services based on data.

Challenges with current data management practices

As with any company transformation, data transformation and the associated creation of data teams posed the question of the teams’ location within the organization. Should there be one central data team? Should it be in IT? Or, maybe there should be one data team per business domain? Should they be under IT, or in each domain? But then, how do they collaborate, and how do they stay consistent, and efficient? Et cetera, et cetera.

Those questions are being addressed in a large number of ways; Data Mesh is one of the few (not to say the only one) that addressed them by rethinking data management at all levels including at the architecture, culture, and organization level.

However, independently of the position of data teams in the structure, let’s analyze the impact on data management when they are scaling.

Data management is a vast topic that is widely defined in the literature, for this analysis, I will concentrate on how data management is defined by DAMA³ in their Data Management Body of Knowledge v2 (DMBOK2)⁴ book.

In DMBOK2, data management comprises many areas, such as Data Architecture, Metadata, and Data Quality. All those areas are participating in leveraging the value of the data, alongside Data Governance which is meant to dictate the policies that must be respected to ensure the data value is generated efficiently and according to the core principles of the organization. This is very well represented in the Data Management Wheel of the framework, see Figure 1-4.

Figure 1-4. Fig 5. Data Management Wheel from DMBOK2

As data ambitions and data teams scaled, each area of the wheel has evolved to adapt to the different situations and needs. In fact, in the next section, I will cover how the increase in the importance of Data Governance is impacting other areas such as Data Quality, and Metadata for example.

Effects of Data Governance at Scale

Many challenges result from the evolutions of data management practices and technologies, but I’ll concentrate on one specific challenge: how to control at scale that the data culture is sustained, by respecting not only the data governance principles but also the definition and the implementation of the resulting policies. Data governance defines the policies, and each area is responsible for defining, planning, and executing the implementation.

However, at the implementation, many dependencies across areas are introduced, plus the fact that everything scales, without a harmonized, defined, and global control layer, any area that presents a default can break the machine.

This is why I am proposing not necessarily to change the abstraction presented in this picture but to extend it with an extra area, centered around data governance but extrinsic to it.

This new area is for Data Observability that will extend the data management practices of an organization with the ability to measure the health and usage of their data within their system, as well as health indicators of the overall system. Nevertherless, a more detailed definition of Data Observability will be explained in the section The Areas of Observability.

Having Data Observability at this position in the wheel makes its responsibility explicit to bridge the culture, its principles, and policies with its implementation and respect across all other areas.

The result is presented in Figure 1-5 below; however, it doesn’t mean there is necessarily a need for a data observability role, team, or division. It states that at scale, the control must become explicit as it becomes a challenge that, at a lower scale, could have been handled in a more ad-hoc fashion.

Figure 1-5. Fig 6. Data Management Wheel extended with Data Observability

In the remainder of this book, we’ll look at how observability extends from the IT DevOps landscape and how to apply it to data and data teams.

To understand why data observability is crucial and must be embraced, we’ll consider how data teams evolve, the impacts on the culture, the distributed responsibilities within data teams, and the many challenges that must be addressed and strategies applied to keep data teams effective.

Challenges of Scaling Data Teams

As companies look to scale their data usage, they also must scale their data teams. In this section, I will take you on the journey of growing data teams and highlighting why automation has become essential.

Consider a data team that starts with a single person tasked to do the data ingestion, data integration, and even the final report. For this person, a first project might consist of producing a view of recent customer acquisitions.

The data engineer would have to query and ingest the source data from the company’s Customer Relationship Management (CRM) system into the data lake, integrate it into the data warehouse, and create a report that gives the head of sales insight into the progress of the sales made over the past few weeks.

This first data team member is, in essence, a Swiss Army knife: covering both data engineering and data analysis responsibilities to deliver the project outcome (report).

In a larger data team, the skills required for this work are balanced between data engineering, data analysis, and business analyst. As a team of one, though, the person must master the technologies required to build and run the pipeline, reporting tools, and understand business KPIs. It is worth noting that each of these areas requires specific expertise.

So far, this person is happy, and the scope of work and responsibilities is as shown in Figure 1-6.

ERP

Pipeline

Report

Figure 1-6. Fig 7. A happy single-person data team and its work.

Of course, the process is not quite as simple as depicted above. For example, there should be an entire systems development life cycle (SDLC) to be followed to get the data into production and allow it to be available for analysis by the final user. At a very high level, however, this basic process outlined is what we mostly think of as “producing value” with a data project.

With only one team member and one data project, the whole data ecosystem is relatively under control.

The process is simple enough for one person, and the responsibility for the entire project is also attributed to a single person. Because there is just a single person on the team, they also have a close relationship with the business stakeholders—such as the head of sales—and have therefore built some domain knowledge around the use case and the business objective of the project. While this role is typically what a business analyst would perform, in this case, the data engineer takes on this role as well.

Thus, if the user raises any issues, it is clear who should work on troubleshooting. After all, he not only executed each step of the process but has knowledge of each area of the process.

However, as stakeholders become aware of the potential and value data can provide, more requests are made to this person for new reports, new information, new projects, and so forth. As a result, there is an increase in the complexity of the demands and ultimately, the individual reaches their limits of production capacity under the current setup. Consequently, it becomes urgent to grow the team and invest in specialized members to optimize productivity.

In the next section, we’ll see how this team starts growing as specific roles are created, such as data engineers responsible for providing the data to analysts, data scientists for building analytics and AI/ML models, and data analysts for interpreting the data for the business team’s use cases. The team members are working together to respond to business users’ requests who will use the insights to make business decisions. See Table 1-1 for the responsibilities and dependencies of each role.

Title/Role |  | Responsibilities | Dependencies
Software and Operation Engineer | Build and maintain the platform; manage security and compliance; oversee site reliability engineering (SRE) | Receives recommendations from other teams about what to monitor and what to consider as an error.
Data Engineer | Build the data pipelines and manage the data from deployment to production | Relies on IT to build the platform and set up the tools to monitor the pipeline and systems, including data observability tools
Analytic/data scientist | Build analytics and AI/ML models that can analyze and interpret data for the business team’s use cases | Rely on the data engineering team to build the pipeline to generate the data they will be using in their models 
Business/Domain Stakeholder | Sponsor use cases and use data analysis to make business decisions | Rely on all upstream teams (likely across other domains) to ensure data and analyses are accurate

Data Team (extended) roles, responsibilities, and dependencies.

There is also a shortage in data engineering, data science, and data analyst skills on the market, hence growing a team is extremely challenging.⁵

This is especially true at the beginning when everything needs to be built in parallel, such as the culture, the maturity, the organizational structure, etc. Thus, it becomes even more essential to keep the talent you have, and this means ensuring that their roles and responsibilities align with their skillsets. So, in the next section, we will discover how these roles are added, what they will be expected to do, and what will happen to the team over time.

Segregated Roles and Responsibilities and Organizational Complexity

The single-person team has reached its maximum capacity to build new reports and has added another member to get back on track with the pace of the incoming requests. As stated earlier, it is important to specialize the team members to maximize their output, satisfaction, and, of course, quality of work. So we now have a data analyst delivering the reports defined with the stakeholders, upon the data sets built by a data engineer.

The impact of segregating the roles is that the data engineer is now farther from the stakeholders and loses direct contact with the business needs. In a way, the data engineer loses part of the visibility about the final purpose of the project as well as part of the responsibility for the final outcome. Instead, the engineer will focus all efforts and energy on their own deliverables—the ETL, SQL, or whatever framework or system was used to build the data.

The team and its deliveries are represented in Figure 1-7, where we clearly see the distance and the dependencies taking place.

Figure 1-7. Fig 8. The team is growing, the complexity as well.

Figure 1-7 also shows the scope of produced work under the management by the data team after a few projects have been delivered. The data ecosystem begins to grow, and there is less end-to-end visibility, as it is scattered across both brains, and the responsibility across team members starts siloing.

Next, a data scientist is added to the team, as stakeholders are willing to explore automated decision-making without necessarily having a human in the loop, or overseeing the results and to scale the value generated by the data.

As shown in Figure 1-8, this adds more scale, and those projects will require more data at a faster speed and more complex as the whole process needs to be automated to generate the expected results.

Figure 1-8. Fig 9. Data team evolving toward the use of automated decision making with AI

As time passes and the whole system scales, issues inevitably begin to occur As noted in the Google paper, Data Cascades in High-Stakes AI⁶ there is a 92% prevalence for at least one data issue to happen in any project.

Because the system has evolved so much and so rapidly, these issues are hard to troubleshoot and resolve. What is happening? No one knows, in fact.

The lack of visibility within the system combined with its complexity makes it like a complete black box, even for those who built it (some time ago).

The organization’s data ecosystem, see Figure 1-9, has become like a “legacy system” in IT—only it happened faster.

Figure 1-9. Fig 10. Efficient data team delivering projects in production after some time

In further sections, I will drive you throughout the journey of handling those issues and their consequences on the data teams’ dynamics. Although, before that, let’s analyze the anatomy of such data issues.

Anatomy of data issues and consequences

Data issues are a constant and common challenge for most organizations. In one survey by Dun and Bradstreet,⁷ 42% of businesses said they have struggled with inaccurate data.

Data issues can be caused for any number of reasons, but the most common are:

Regulatory

Changes in data privacy or other data regulations may require changes in how data is collected, ingested, or stored, which can create unforeseen issues. For example, retail companies had to adapt their data collection strategies to use appropriately the information associated with the store cards.

Business demands

Different business use cases may require different configurations of the data. One business use case may rely on the shipping address and not anymore the billing address, for example. Therefore, the billing address may be removed and a business user may request that the address column be removed from the data. However, someone else using the same dataset may need addresses for their use case, so their analysis is now incorrect when this information is left off the data.

Human error

Often, data issues are caused by simple human error—someone accidentally deletes a field or column without realizing it. The data then becomes inaccurate, missing, or incomplete.

One of the most challenging aspects of what causes a data issue is that the engineers involved in creating the change to the data or application often don’t realize the implications of the changes they’ve made. And unfortunately, the issue isn’t usually discovered until the end of the data value chain. All in all, the whole point of this book is to teach how everyone can take part in the responsibility to react fast to any such situations.

Usually, as we discussed in the scaling data team story, it is business users who are running reports and realizing through a gut feeling and their own previous experience using the data that the numbers “don’t look right.”

But, at that point, it’s already too late. Business decisions may have already been made based on faulty data before the inaccuracies were discovered.

With no time to fix data issues, data teams scramble to figure out who has the knowledge and skills to help resolve the issue. Yet, it’s often not even clear who is responsible or what knowledge and skills you need to address the issue. Analysts? Engineers? IT?

And the responsibility can change from one moment to the next. Perhaps the analyst made a change in how some information is calculated that is now impacting the sales reports, but perhaps even before that, the data engineering team adjusted one of the connections supporting the sales analytics tool the business users use to run the sales reports.

To try and resolve the issue, everyone is relying on everyone else’s memory about what they did or didn’t do and are manually trying to find and fix the issue. No one has a clear understanding of what fields or tables affect downstream data consumers, and the only notifications they have set up are basic failure alerts.

The expense and time involved to resolve the issue and its negative impact on businesses’ productivity, sales, overall revenue, and even reputation are significant.

The Dun and Bradstreet⁸ report also tells us that almost one in five businesses has lost customers due to incomplete or inaccurate data. And, nearly a quarter of companies say poor quality data has led to inaccurate financial forecasts.

Constant data issues can lead to a lack of confidence in making business decisions based on data insights. In a recent study of 1,300 executives, 70 percent of respondents said they aren’t confident the data they use for analysis and forecasting is accurate.

Impact of data issues on data team’s dynamics

First and foremost, the data issue is detected by the user of the data who starts having doubts because either the data received seems odd compared to what was expected, or the results of the analyses are not what was anticipated. The direct side-effect of those doubts is the sentiment of mistrust that installs in the user’s mind against the team which produced the data, or even simply the whole data platform.

Note

This case could happen even if the data has no “issues”, because the data changes naturally as it represents the reality, which changes without notice - and so the user might not yet be aware. However, for simplicity, I will assume for the remainder of this study that the issue is real.

Figure 1-10 depicts the current situation, where one of the consumers is starting to have concerns about the data, and who imagines that the issue just discovered may have other, yet to be discovered, consequences in some or all applications that use it (a.k.a. data cascade, issue propagation, …).

Figure 1-10. Fig 11. A user not cool about what has been discovered, a data issue

In this scenario, users always detect issues at the wrong moment for them. They need the data at that moment, otherwise, they wouldn’t be accessing or checking the data anticipatively.

So the user is stuck and of course, becomes grumpy. Indeed, instead of working on the planned tasks, the user has to find out how the data will be “fixed.” Consequently, the delivery of the tasks is delayed, which will impact the stakeholders.

Moreover, finding how the data can be fixed is not straightforward as the received data might have been corrupted by any of the applications that produce it. At this point, the process to follow depends on how the (data) operations are organized, for example, creating an “incident” (a ticket) with a comment “data is not right ߘ䬠please fix ASAP”.

Eventually, the data has a data steward attributed to it. The data steward has the responsibility to unblock the user (at the very least) and to analyze the potential other consequences, which may result in more incidents.

To resolve such an incident, the data steward will have to involve the producer (or the producers…) who will need time to diagnose the symptoms and elaborate on a solution.

Well, this sounds easier than it really is. The incident needs to be reassigned to the producer, who is probably already busy with ongoing projects and unlikely to identify (the issue) instantly. That’s assuming they are even the right producer!. Instead, all producers will be contacted, probably summoned in a meeting, to understand the situation. They will challenge the “issue” (e.g., “are you sure this is an issue”?), and ask for more details, which loops back to the already angry user (who is most likely not going to be very cooperative).

Consequently, the applications that are touching this data and identified as potential culprits have to be thoroughly analyzed, which involves those tasks:

Access to the production data to manually verify and understand the data issue by running a series of exploratory analyses (e.g., queries, aggregation computations in spark, …). Given that, the grant to access the data might not be granted yet.

Repeating this operation by time-traveling the data (when possible) to identify when the issue started to show up (e.g., a delta table can help to do the time-traveling, but still the time needs to be found).

Access to the production logs of the applications (by SSHing on the machines or from the logging platform), to analyze its behavior and even review its history (version, etc), in case the application business logic has changed so did the results. Given that, like for data access, the logs might require special grants to be accessible.

Analyze the business logic to trace back to the data consumed by the applications to identify potential data cascades (issue propagation).

Repeat until the root cause is found, which data and applications need some repair, and finally execute the back-filling (to reestablish the truth).

Easy peasy, right?

So, let’s be honest, because time is critical (remember, the user is upset), it is likely that the root cause won’t be tracked but either one of these two temporary-definitive patches will be applied:

Run an ad-hoc script (or so-called data quality tool) to “fix” the data: which is totally getting the process out of control (e.g., outside the lifecycle of the data, etc.).

Update one of the applications to “clean” the received data: which is fixing the problem locally and is likely to have awkward side effects; such as removing or imputing null values that would change the distribution of the variables if the number grows up and have side effects downstream (e.g., bad decisions…).

In fact, this process, called “troubleshooting”, has puzzled many people, many of whom weren’t even needed for this incident. It is worth noting that while the troubleshooting was happening the scope of the issue has grown significantly, as shown in Figure 1-11.

Figure 1-11. Fig 12. People involved in the incident analysis (troubleshooting)

Moreover, the issue detected on this data may have other consequences (remember, data cascades!) on other projects, users, and decisions. Which has the effect to expand further the scope of the issue to all usages of the data.

Another process called impact analysis is somewhat close to the troubleshooting we’ve just covered, however, its goal is slightly different, as it is aiming to prevent issues or communicate them.

In fact, not only is the goal different, but an impact analysis is also much trickier and more sensitive because it requires requesting time from all users to check their data, results, analyses, or decisions.

Even worse, some may discover that decisions were wrong for a longer period of time, potentially as long as the issue has appeared, everything happened silently.

At this point, the scope of the data issue, considering both the troubleshooting and the impact analysis, is as big as Figure 1-12 shows. And the data steward has to resolve this as fast as possible.

Figure 1-12. Fig 13. The scope to resolve a data issue.

This is why we call data a “silent killer.” It slowed everyone down, destroyed all trust, generated stress, anger, and anxiety without raising any alerts, or “exceptions” (as in software). In fact, data doesn’t raise alerts, or exceptions yet, because there is a way to enable this capability we will see in the next chapter.

In this analysis, I have essentially highlighted the challenges around the time and the people involved in the process and the significant efforts wasted on ad hoc patching which just creates more technical debt. Despite the importance of those challenges, there is another inestimable resource that we lost instantly at the time when the issue was detected: trust in both the data and the data team.

This trust took months if not years (e.g., for sensitive AI-based projects) to build, but because nothing had been anticipated to maintain it, it fell apart like a house of cards, within seconds. This kind of scenario and the resulting consequences on the general mood lead to discouragement and turnover of highly talented team members.

Hence, we can identify many questions raised throughout the incident management process that are mood-destructors:

Who is responsible for this issue?

Why am I the one that:

Discovered the problem?

Is being questioned about it?

Why is the user telling me the data seems inaccurate?

Is the data really inaccurate?

What apps could be causing this?

What upstream data could be the cause?

What are the unknown consequences of this issue (i.e., what other projects is this data used in that could also be impacted)?

Understanding why these questions are raised allows us to identify the very source of the problem, and it goes beyond any data issues and their causes. The challenge to address is not data itself, it is the lack of clarity about accountability and responsibility across the data team members and stakeholders, strengthened by the lack of visibility into the data processes in the wild (i.e. production, during its use).

Consequently, when the questions are raised by or to data teams, they create and reinforce barriers at the creation of value using data as they will:

Take significant time to resolve during which the data is still unusable by the consumer.

Require significant energy and imply that there is a diminished capability to solve the problem at the source and that a local “patch” will be the likely default solution.

Provoke anxiety and exacerbate the loss of confidence in deliverables, data, and suppliers.

This is where data observability plays a key role—helping to generate greater visibility into the health of the data and the data ecosystem and better assign (distribute, decentralize) both accountability and responsibility.

Scaling AI Roadblocks

In the previous section, we covered the challenges that appear when scaling data teams. In this section I will focus on one of the most critical achievements that any organization is expecting from their data teams, scaling their AI capabilities.

However, these same types of visibility challenges that limit value when issues arise also create significant challenges when it comes to implementing AI.

In an analysis performed by Gartner see Figure 1-13 the most important roadblocks faced by companies in developing their data and AI programs included data volume and/or complexity, data scope or quality problems, and data accessibility challenges.

Figure 1-13. Fig 13. Gartner survey’s results about AI Roadblocks

The good news is that data observability helps with these roadblocks, but before we get into how, let’s look a bit deeper at what the primary barriers are and why.

The survey results indicate that technology-related challenges account for 29% of issues and the complexity of existing infrastructure accounts for 11% of the issues. Interestingly, almost half (48%) of respondents selected challenges that have a relationship with a lack of visibility due to the complexity of the system in place and a lack of clarity on who was responsible for remedying the issues.

Let’s take a look:

Security/privacy concerns or potential risks or liability

Nearly one-fifth (18%) of respondents to Gartner’s survey picked these issues as their biggest barrier to AI implementation. Security of risk management is all about knowing who, why, and for what purpose the data will be used for. There is also a potential risk or liability if the outcomes from the data are inaccurate because the data itself is wrong and lead to bad business decisions. Finally, there is concern that the outcome may not meet certain ethical constraints.

Data volume and complexity

This issue was the top barrier for 8% of respondents. The complexity and the size of data require a lot of experimentation to understand and derive value from the data. Because of a lack of visibility on experimentations that were performed such as profiling and wrangling, these experimentations are repetitive on the same big and complex datasets–This takes time and effort.

Data scope and quality

For 6% of respondents, data quality issues were the top barriers. If the data quality is unknown, then it’s very difficult to have any confidence in the outcome or final results produced from the data.

Governance issues or concerns, lack of understanding of AI benefits and uses, and difficulty finding use cases

A total of 16% of respondents felt that one of the above issues was the biggest challenge in implementing AI. Data governance is a big issue because documentation is manual and time-consuming, which means it’s not always done properly and therefore its overall impact and value are not always apparent. But without good data governance, the quality of the data fed into AI algorithms could be impacted, and without visibility on the quality of the data, stakeholders may worry about the security of the data and whether the AI outputs are accurate.

So far, I have taken you on the journey of scaling data teams and AI, and we have identified several challenges such as lack of visibility (data pipelines, data usages, …), of clarity about responsibility and accountability, resulting in distrust and chaos, leading themselves to loss of interest or confidence into data ambitions. In the next section, I will describe what data observability is, and how this approach and capability solve those challenges.

Data Observability to the Rescue

Up to this point, I’ve discussed the challenges faced by data teams as they grow and the roles a lack of visibility and clear responsibilities play in making it difficult for organizations to scale their data and analytics strategies.

However, this is not the first time we’ve encountered such challenges, the closest example is in IT when it scaled rapidly, which led to developing the DevOps culture.

DevOps has evolved over the years, as more IT practices (e.g., the service mesh) have required more best practices and associated tooling to support them efficiently.

The most well-known example of such best practices is probably CI/CD but also observability at the infrastructure or application level has become part of any architecture to give more visibility and confidence in their applications and systems while also speeding time to market and reducing downtime.

Associated markets have therefore emerged and grown to support these observability requirements. There are a variety of companies that have developed DevOps-related services and technologies at all levels of maturity of the IT environment (e.g., Datadog, Splunk, New Relic, Dynatrace, etc.).

In IT, “Observability” is the capability of an IT system to generate behavioral information to allow external observers to reconstruct (modelize) its internal state. By extension, continuous observability allows an external observer to continuously modelize the internal state.

Fundamentally, an observer cannot interact with the system while it is functioning (e.g., we can’t log onto the server), it can only observe information that can be perceived, which are therefore called “observations”.

Now, let’s discuss what those observations are.

The areas of Observability

An IT system is complex by nature as it is composed of several categories that can drastically expand in number, such as infrastructure, cloud, distributed, machine learning, deep learning, etc.

In this book, however, I’ll stop at getting too fine grain, but will aggregate the categories of an IT system that can be “observed” into several areas, and one of them is related to data, as represented by Figure 1-14.

Figure 1-14. Fig 14. Areas of IT Observability

These areas are not totally independent of each other, as they have a lot of interactions to encode the complexities of the system. This is why a Venn diagram seemed to be the best representation.

Before covering in detail the “data” area, the “data observability” then, let’s review the others briefly first:

Infrastructure

Using infrastructure log metrics you can infer the performance characteristics associated with internal infrastructure components. Proactive actionable alerts can be set when there is a failure or certain performance parameters aren’t met.

Application

Observing application endpoints, versions, open threads, number of requests, exceptions, etc., can help determine how well the application is performing and identify if or why there are issues.

User/Purpose

It is useful to understand and “observe” who is using or implementing applications, what the purpose of a project is, and the goal of the project. This helps to understand the most frequent projects or goals, detect duplicate efforts, or compose centers of expertise.

Analytics

Observing analytics, from simple transformations to complex AI models, helps identify and learn from the ongoing usages of data and the insights generated from them.

Security

Observing security-related operations such as modifications of grant accesses or roles, or metrics on which roles are used more often than others, gives visibility on the efficiency of the security and areas of improvement.

Some of the above areas have already been covered largely in the DevOps literature, however, we are focusing specifically on data observability. Therefore, we conclude that data observability can be defined as such:

Data Observability is the component of an observable system that generates information on how the data influences the behavior of the system and conversely. An observable system is a system having the “Observability” capability.

It is worth noting that Gartner has defined data observability as the following, which aligns well with the above definition:

Data Observability is the ability of an organization to have a broad visibility of its data landscape and multi-layer data dependencies (like data pipelines, data infrastructure, data applications) at all times with an objective to identify, control, prevent, escalate and remediate data outages rapidly within acceptable data SLAs.

Data observability uses continuous multi-layer signal collection, consolidation, and analysis over it to achieve its goals as well as inform and recommend a better design for superior performance and better governance to match business goals.

The Gartner definition also discusses how data observability can be used (e.g. prevent issues). However, I didn’t include this as part of my definition because I want to focus on what it is not what it does (also, I don’t define an apple as a fruit that can satisfy my hunger).

That said, it is important to know the benefits of data observability. We’ll dive into data observability use cases in more detail in the next section.

Nevertheless, both Gartner and I agree that there is an important “multi-layer” or “multi-area” component to be taken into consideration. I have, however, used the term “areas” because layers imply there is some independence, which, in reality, is not the case.

From this definition, we can look at the dimensions composing data observability as a result of its interactions with the other areas of observability.

I will start with the main dimension formed from observations related to the sole dataset. These are the core, or intrinsic, observations. They are essentially metadata such as the fields (e.g., columns, JSON attribute), the format (e.g., CSV), the encoding (e.g., UTF-8), and to some extent, the definitions of the information available.

They allow an observer to understand (mainly) the structure of the data and how it changes over time.

If we were to stop here, we wouldn’t be able to leverage data observability to its maximum capacity. For this, we must connect those core observations to the other areas to give the observer the following benefits.

Infrastructure

Identify where the data resides physically (e.g., file path on a server, the server hosting a database - ideally its connection string), and it could impact the data itself or the other areas cited below.

Application

Be aware of what components are storing or using the data (e.g., a transformation script, a web service), it could also include the code repository and the version.

User/Purpose

Contextualize and ease the knowledge handover with information about how was involved in the data production or consumption, its security settings, and in which projects (which have purposes) the data brings show value.

Security/Privacy

Control how liable and appropriate are the data collections, accesses, and usages.

Analytics

Understand how value is created with the data, through transformation (lineage), AI (machine learning training, predictions, …), or even simply migration.

As for any new concepts, especially those that touch the organizational culture, ways of working, and technologies, it is crucial to understand their use cases. Therefore, in the next section, I will cover the most common and valuable data observability use cases.

How data teams can leverage Data Observability now

The main use cases of Data Observability are currently oriented towards data issue management. However, as data continues to proliferate and its uses expand, there will be many more use cases to come.

Low Latency Data Issues Detection

The more synchronized data observability is with the application (its usage context), the smaller the delay between the issues and their detection will be. In fact, data observability can be leveraged at the exact moment of data usage to avoid any lags between monitoring and usage. This will allow you to detect data issues as quickly as possible, helping you to avoid having data users find issues before you.

Leveraging data observability this way reduces the time to detect (TTD) issues as data engineers are alerted in a timely manner because any data usage issues are observed in real-time (aka synchronized observability).

Efficient Data Issues Troubleshooting

In most organizations, when data issues arise, data engineers spend a lot of time trying to figure out what the problem is, where it originated, and how to fix it. And every step of the process takes a lot of time. With data observability, the time to resolve (TTR) is much faster because there is visibility into the entire system thanks to contextual observability, which provides information on the data itself and the context of its usage. This enables data engineers to fix issues before they impact downstream business users.

Preventing Data Issues

When implemented as part of the entire development lifecycle, including production, data observability provides continuous validation of the health of the data and the data ecosystem. Continuous validation can perceptibly improve the reliability of the applications and prevent data issues, overall lowering the total cost of ownership.

Decentralized Data Quality Management

SLAs can manage and ensure data quality, just as they are used in IT DevOps to ensure reliability and other key metrics. This new managing data quality requires data observability, which on the one hand, can provide synchronized (near-real-time) and continuous validation of the data, which further improves the efficiency of any SLOs in place. But more importantly, on the other hand, data observability will allow SLA and SLOs to be defined at the granularity of the usage, and the context (the application, for example). This capability solves one of the most important roadblocks of data quality management programs, the definition of SLAs by owners, stewards, or SMEs who have a hard time defining a single set of constraints that will supposedly meet all usage expectations. Hence, they cannot come up with a single (central) set of SLAs as each use case is likely to perceive the SLAs differently. The key difference with data SLAs is that they can take a very large number of forms that feels quickly infinite; for example, you could have SLAs for the min representation, number of nulls, number of categories, skewness, quantile 0.99, etc. for a single field from a random CSV file. Hence, leveraging data observability to decentralize the SLAs in a contextualized manner (the usage) is key to managing data quality and defining a culture of accountability and clear roles and responsibilities.

Complementing Existing Data Governance Capabilities

Remember the DAMA-DMBKO2 data governance framework from earlier? Because data observability is part of the architecture and surrounds all of the areas of data governance, it provides visibility into all of the different components that interact at the data level. This enables data teams to automatically create documentation using the same kind of data, data storage, and data modeling and provides greater visibility into the different data models that exist, what data has been published to the data catalog, which analytics have been run on what data, and which master data was used.

The future and beyond

By better understanding the different use cases for data observability, you should now be able to understand how data observability can be used to help optimize your data systems as well as data teams.

In the coming chapters, I’ll go even deeper, detailing how to set up these use cases and covering best practices to capture the information necessary for each use case in the most efficient and valuable manner.

¹ Mary K. Pratt, “CIO (Chief Information Officer),” TechTarget, accessed April 11, 2022.

² Cf What is Data Observability report

³ The global data management community - https://www.dama.org/cpages/home

⁴ DAMA-DMBOK: Data Management Body of Knowledge v2

⁵ https://quanthub.com/data-engineer-demand/

⁶ https://storage.googleapis.com/pub-tools-public-publication-data/pdf/0d556e45afc54afeb2eb6b51a9bc1827b9961ff4.pdf

⁷ https://www.dnb.co.uk/content/dam/english/dnb-data-insight/DNB_Past_Present_and_Future_of_Data_Report.pdf

⁸ https://www.dnb.co.uk/content/dam/english/dnb-data-insight/DNB_Past_Present_and_Future_of_Data_Report.pdf

Chapter 2. Components of Data Observability

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 author at gobrien@oreilly.com.

As introduced in Chapter 1, data observability is an area of (IT) observability intersecting its other areas such as applications or analytics. In this chapter, we will cover how data observability, and its intersections, can be added to a system where data must be observed.

As discussed in previous chapters, data observability gives observers a broader spectrum of observations to interpret the internal state of the system by combining all areas. However, this combination can become a challenge itself if some precautions are not respected. This chapter will give you a deeper understanding of what observations are and what they should contain.

We will first review the three channels where observers can access observable information to interpret the internal state of the system in relation with data. Then, we’ll see how this observable information can be organized to simplify this interpretation, its model. The chapter will conclude with the third component, that stretches the observability beyond information generated by the system, but mixes it with expectations observers may have about the internal state.

Channels of data observability information

The first component of data observability is the channels that convey observations to the observer. There are three channels: logs, traces, and metrics. These channels are common to all areas of observability and aren’t strictly linked to data observability.

Below are definitions of each of the three main channels of observability. You are likely already familiar with these channels, but if not, there are hundreds of books and blogs that delve deeper into defining them. If you do want to read more on the topic, I recommend the book Distributed Systems Observability¹, which dedicates all of chapter 4 to defining these channels. I also recommend part two of Observability Engineering² as well as the REF.

Logs

Logs are the most common channel of observation produced by the IT system. They can take several forms (e.g., line of free-text, JSON) and are intended to encapsulate information about an event. A line of a log (typically logs are a stream of lines), is the result of the act of logging.

In IT, logging is a decades-old best practice, especially in infrastructure, application, and security. Logging has been used to not only debug but also optimize IT systems or processes. There are even developed standards for logs, such as Syslog, that specify the log structure and even allow heterogeneous infrastructures to be controlled by a central system.

While logs are crucial to capture information about the behavior of a system, it is difficult to use logs to recreate a multistep process. This is because logs comprise all activities within the system and the logs of a process are likely interwoven with other concurrent processes or scattered across multiple systems (e.g., distributed system, service mesh).

Traces

Traces allow you to do what logs don’t—reconnect the dots of a process. Because traces are a representation of the link that exists between all events of the same process, they allow the whole context to be derived from logs efficiently. Each pair of events, an operation, is a span which can be distributed across multiple servers.

Traces, with their spans, are an efficient way to follow operations across services and servers, as information such as the server, the service, the timestamp of the event are conveyed. So an observer can easily browse the logs of the service, on the server, at the given time to analyze the logs of the specific event for which they need to observe.

In practice, however, spans are also conveying logs that might be relevant for the observer to analyze on the spot, instead of having to reconnect the trace information. Even though observability systems allow integrations between traces and logs to simplify further this process.

Note

A different form of trace in the data and analytics context is the data lineage (see section REF). Although data lineage is not strictly speaking a trace (because it doesn’t connect events), the data lineage connects data sources to encode the provenance or the generational model. However, in certain situations, data lineages can be used to help troubleshoot a process or discover opportunities to optimize.

The last channel of observations is metrics, which is closely connected to logs and traces, as both can contain metrics. However, because of their simplicity, metrics have a significant advantage over logs and traces.

Metrics

Every system state has some component that can be represented with numbers, and these numbers change as the state changes. Metrics provide a source of information that allows an observer not only to understand using factual information but also leverage, relatively easily, mathematical methods to derive insight from even a large number of metrics (e.g., the CPU load, the number of open files, the average amount of rows, the minimum date).

Due to their numerical nature, they need to be recorded. This is why metrics were once part of logs as well as present in traces. Because their usefulness is so straightforward, over the years, we have seen standards to simplify the publishing or collection of metrics independently of regular logs and traces.

Based on what I’ve described above, it may sound odd to talk about logs, metrics, and traces as being produced by data (a number such as “3” cannot log, measure, or trace anything, it is a number). This is why it is so vital to consider data observability as an area that intersects others, such as applications and analytics, which implies that the information produced by logs, metrics, and traces from several areas must be connected. Let’s tackle this in the below section.

Observations model

Because observations can be captured from different channels, in different formats, and are related to several areas of observability, it is important to model them, at least minimally. In the following sections, I will present a model to encode the dimensions covered by data observability that support the solution for the use cases presented in Chapter 1.

I want to emphasize that the model I’m proposing is one I have seen work in many projects and use cases. Though the model may change over time to include more complexities or encode more relations, I intend to show you in detail how the observations from the different areas can work together easily as long as the bridges across them are well enough anticipated to avoid reverse engineering or approximations.

This model will, at the very least, give you a viable starting point to generate information that represents the state of the system you need to observe. However, you can consider it as a core model which you could extend with additional, potentially custom, needs. In chapter 3, I will explain the different strategies you can use to generate automatically most of the information of the model, and, in chapter 4, the associated ready to use recipes for several common data technologies (e.g., python pandas, apache spark, SQL).

Note

It is true that other areas of observability have not necessarily introduced such a model yet, and this is limiting their ease of use. Having logs, metrics, and traces coming from many infrastructures or applications without a clear structure to recombine them (the intersections) makes the process of generating insights about the internal state cumbersome.

The information are therefore recombined in a best effort manner, using a posteriori analyses which depends also on human intervention.

At the time of writing, Data Observability is raising out of the water, therefore my intent here is to avoid repeating the mistakes made in other areas, due to the youth of Observability, with the introduction of a model to 1) accelerate its adoption 2) ensure its ease-of-use.

Ideally, ISO specifications will be created, endorsed, and widely supported, hopefully based on this model, in the meantime, let’s use common sense.

In Figure 2-1 below, the model is designed as a graph to define the entities providing some information about the state and how they are linked to each other. The links are clearly identifying the intersections between data observability and other areas. For example, in the section below, you will learn that a data source (data observability) is linked to a server where it is accessible (infrastructure observability), hence the link itself is part of the intersection of these two areas.

Figure 2-1. Fig 2-1. Data Observability Core Model

The model presented in Figure 2-1 aims at structuring data observations to maximize the overall interpretability for an external observer, and to provide a common language that can be used independently of the people, the time, and technologies.

To improve the readability of the model, it has been split into the three following spaces:

The physical space links observations with tangible entities. Even though a server might be virtual or a user is a system or applicative user, the physical space represents a space that can be physically checked.

The static space represents the entities that are changing relatively slowly in opposition to the dynamic space. It represents a state of the system that could be built manually.

The dynamic space is part of the system that is evolving or changing so fast that it cannot be consolidated, documented, or built manually. This is where automation is not even a question–but a necessity.

Within each space, there are entities introduced to enable some observability. Let’s go through each of these entities to understand what their purpose and how their relation to other entities play a role in data observability.

Physical Space

In this section, I will cover the information in the Data Observability area that can be considered tangible: the server and the user (or their virtual alternatives). Let’s get started with the server first.

Server

The server is related to the machines which are running the IT system being observed. It is meant to provide information about physical appliances. However, in the era of cloud computing, it can also represent a container, a virtual machine, or similar that opens the door to access infrastructure observations.

Because I am defining the core model, I kept the server simple without encoding containers, virtual machines, and clusters for example – which can be extensions of the model. However, the core information about the server must help the observer to tap into other systems (e.g., Datadog) to analyze the infrastructure status.

To ensure this capability to the observer, the type of information the server must convey includes IP, container id, machine name, or anything that can identify the underlying infrastructure appropriately.

In [Link to Come], we’ll see an example of how and when the server helps the observer.

User

The user is probably the easiest actionable information that can be used by an observer to improve their understanding of a situation. This is because the user mainly has knowledge and use of the system.

Like the server, the user can be a virtual abstraction, such as an identifier that doesn’t hold personal information (e.g., GitHub account id) or a so-called system or applicative user such as a root or spark-user, that gives the observer a hint on the impact of the user on the behavior of the system (e.g., security) or its liability (e.g., privacy).

In other words, the user is a way for the observer to connect the status of the user with the purpose of observability.

In later sections related to the applications within a system I’ll discuss the advantages of having such information on hand.

While I have limited the physical space to users and servers, we may think of other elements such as GeoLocation that can also be part of the physical space.

However, I recommend starting with a simple model to get the core visibility needed, then eventually growing it with additional use cases. That said, GeoLocation does fit nicely in the user and infrastructure observability areas.

Now let’s address the static space, which includes most of the exclusive data observability entities.

Static Space

Whilst the physical space allows the observer linking events about data to “things” that can be accessed (even virtually). In this section, I introduce the static space that gives the observer the ability to analyze the status of the system at rest, or where the time has less influence than the physical space on its status (e.g., the existence of a server, the code change of a user). The entities are related to data and applications, and their structural evolutions.

Data Source

The data source provides information about what data exists and how it is used–whether that’s a simple CSV file, Kafka topic, or a database table. The local file path, the topic name, and the table name are the kinds of information that are typically part of the data source.

The data source is present in the static space because data sources do not change rapidly. Of course, files can be copied or moved to other locations, but the model would consider the copied or moved files as new data sources. Renaming a table in a database is also similar to moving it. It’s important to note though, that renaming and moving data sources can have big consequences. Applications can fail as they expect the data to be at a specific location, but the location may change in production because the data has been moved (e.g., `mv`). In such cases, there will be several data source entities, one for each data source location, as in fact moving a data source from one location to another is copying then deleting it. This action of copy then delete is actually a transformation which will itself be represented by its own entity, the lineage introduced further in this chapter.

The data source is also an element of observation that resides in the main component of the data observability area and is linked with the server, which is part of the intersection with the infrastructure observability area.

Below are some use cases highlighting where you can gain observability of the data source:

Data migration to/from another server

Data duplication or redundancy

Data availability on a server (e.g., server migration or upgrade )

Data access (i.e. if a machine is physically connected to access the data)

However, the data is not only a source, it is a repository for several types of information. In the next section, we’ll explore what this means.

Schema

While the data source introduced above provides information about how data can be accessed, it doesn’t cover what type of information the user may find in the data.

The Schema entity provides visibility into the structure of the data source³. The schema is an essential component of metadata (others would be definition, tags, etc) and conveys the information about fields (or columns), potentially deep (embedded), available in the data source. Each field of the schema has at least a name and an associated type, which could be native (e.g., string, int, address) or conceptual⁴ (e.g., address, SKU_ID) or both.

Even though the schema is in the static world, it is still a changing entity, as a database table can have columns added, removed, or renamed. However, the schema makes it relatively manageable, although cumbersome, to keep track of the changes.

The schema is linked to the data source because it allows the observer to identify the types of information available in a data source.

Because the schema of a data source is likely to change, the model can encode these changes by keeping all versions of the schema in two different but non-exclusive methods. Each method supports interesting use cases:

Each modification of the data source’s schema creates a new schema entity, along with the time of change or version.

Each consumer of the data source creates a schema entity related to its own usage of the data source.

Interestingly, the schema, depending on the semantics presented above, is an observation that is either part of the (a.) main component of data observability or (b.) an intersection between data and analytics observability.

At this point in the model, we are capable of recovering decent visibility of the data. However, we still lack a fair amount of information about how the data plays out with the remainder of the system, especially applications and analytics. The remainder of this section will focus on how to achieve data observability within these two areas.

Lineage

The Lineage entity, or more precisely the technical data lineage, is probably the most difficult, and rarest, information to uncover. There is a lot of literature on how a lineage can be used (REFs?)(I’ll list several at the end of this section), but there is little about how to collect it concretely.

Lineage, literally the line+age, refers to the direct connections between data sources. The lineage of a data source is therefore the set of direct upstream data sources and their lineages. The technical data lineage can be either at the data source or field level. Considering that the schema level is more complex to generate than at the data source level as it provides information on how data sources’ schemas are connected to each other. That results in a connected graph where each field of a data source has its own lineage that connects it to the fields of the upstream data sources, as shown in Figure 2-2.

Figure 2-2. Fig 2-2. Example of technical data lineage - in black = data source level, yellow = field level

The connections between data sources are created using applications that execute transformations of inputs (sources) into outputs (results). These transformations can include:

A SQL query issued by a Java web server transforming a table (input) into a JSON HTTP response (output)

SQL generated by a reporting tool computing and presenting (output), a key performance indicator from a series table (inputs).

A Spark job that creates a view in a data warehouse (output) from a number of Parquet files (inputs).

A Kafka application using a streaming API that consumes a topic (source) and produces another topic (output).

A machine learning script training a model (output) on a feature matrix created from a CSV file (input).

Even a human copying figures from a sheet of paper (input) into an Excel sheet (output) is a lineage.

The information contained in the lineage is essentially the mapping between input and output data sources and/or schemas.

To encode this mapping with the data source and schema, the model presented in Figure 2-1. shows that the lineage is linked to the schema which enables the following use cases:

[Use Case To Come]

[Use Case To Come]

Although conceptually simple, the complexity of lineage is because it is:

Cumbersome to document: All data sources must be documented, and each operation applied to all fields as well.

Likely to change: even a simple modification of a SQL query will change the data lineage.

Numerous: There are so many transformations happening continuously on a company’s data. Even a simple CRUD application is composed of many transformations.

Despite these complexities, I have left lineage information as part of the static space because a lineage relies on humans who transformed data to create their desired output.

When considering lineage outside of the main component of data observability, we can see that it also fits well with:

Analytics observability: Lineage contains the structure of the analysis and how it is performed with data.

User/purpose observability: Lineage encodes what can be achieved with the data sources.

Application observability: Lineage is executed by an application, and a data application exists to execute lineage.

There’s a lot to explore when speaking about the contribution of lineages to user/purpose observability and application observability, which we’ll explore further.

Application

The application gives access to another part of the data observability because it is key to providing visibility of the data. In fact, the applications are at the center of all information: they are the product of the user, the container of the analytics, the consumer and producer of data, and they run on infrastructure while also configuring or implementing the security.

It is essential that information about the application is recorded so the observer can understand which system is at play and the associated logic.

I tend to consider the application like a door to the other areas of observability, because the application is relatively central to everything. In fact, referring to an application can be as simple as:

The groupId: artifactId of a Spark job

The path to a workbook in Tableau

The organization/module_name of a python script

The URN of a step in an orchestrator descriptor (e.g., GCP Workflow) scheduling an SQL

etc.

Figure 2-1 shows that the Application does not refer to anything but still has the most central position in the observation model, it must be executed. I will explain how this is done in [Link to Come].

One could argue that lineage should be directly connected to the application. For example SQL joining two tables in Oracle and creating a view in Snowflake is coded in the application that will run it, hence creating a direct connection between both entities.

However, I believe that this view couples Data and Application Observability areas too strongly in the static space. There are several cumbersome instances where this direct connection is completely inappropriate:

Lineage modification: When the lineage is modified and a new lineage is created, the initial lineage is no longer part of the application. In this case, the lineage should be connected with the Application Version (see below).

Lineage is generated at runtime: When the lineage is generated at runtime, such as SQL created on the fly, all the SQL is not existing in the application until it runs and is used. In this case, the lineage would be connected to the runtime (execution) of the application (see further Application Execution in the Dynamic Space).

Of course, the application is not disconnected from the lineage, it just won’t link to it directly. Instead, it plays the role of a container by embedding most of the observation model via the two components, the Application Repository and Application Version.

Application Repository

The Application Repository is often seen as the outlier in the model presented in Figure 1., because it gives the information about where the source of the application resides (e.g., source code, report configuration).

At first glance, there seems to be very little, or no connection to the data. However, the information provided by the Application Repository is clearly intended to encode the connection between Data Observability and Application Observability.

Because an Application can have its source moved around, the Application Repository can represent its latest location (hopefully not several at the same time) and gives a hint about where it resided in the past. This notion of temporality is something that I have chosen to encode in a different entity–in the Application Version.

One of the primary usages of the repository is to give the observer the direct location where they should be analyzing the system. For example, if there is an issue with the generation of a data source by an application, knowing where the application repository is tells the observer exactly where to look. However, they will also need to know the version of the application to analyze the situation appropriately at the time the issue happened, which is where Application Version is useful.

Application Version

Application Version is the final entity –and one of the most critical as it is the glue between the other entities– in the static space. The Application Version is the entry point in the Observations model that refers to the exact version an application was running when an issue occurred with the data.

The version can be the version number (e.g., v1.0), a git hash (e.g., d670460b4b4aece5915caf5c68d12f560a9fe3e4), or even simply a timestamp (e.g., 1652587507).

The version allows the observer to browse the history in the Application Repository for the running Application under supervision. The version is also useful when conducting root cause analysis. If the version is too old, it may indicate that the application has not been (yet) updated. If the version is unknown/newer, it may indicate that a bug in the logic has been introduced or an unknown transformation has occurred.

You’ll see that the version is not directly connected to the lineage or any entity closer to the core Data Observability entities (e.g., Data Source, Schema). Thus, to create a bridge between Application and Data Observability, we must take into account what is executed, read, transformed, and produced. This is what the Dynamic Space is for.

Dynamic Space

The Dynamic Space gives the observer the capability to leverage the behavior of the system using the data. Its main purpose is to create visibility on what is called the runtime, which is highly dependent on the status of the environment in which the system is at play (i.e., its runtime context).

Also, the status of the environment is very sensitive to events that occur during runtime, such as the data changing due to real-world events, the application being killed due to a virtual machine reboot, etc. In other words, it is very dynamic and unpredictable (at least deterministically), which is why I call it the Dynamic Space.

The composition of the Dynamic Space is described in the structure that follows, starting with the Application Execution.

Application Execution

The Application Execution entity is key to providing the observer with information about the application that is running and its use of data.

As previously noted, the application is the artifact that opens the door to greater visibility of other components within the observation model because its source describes the analytics (e.g., lineage) using the data. By capturing the real runtime and its execution, the application execution conveys information to the observer such as:

An ID: This allows the observer to identify which Application Observability information is relevant to review (e.g., in an application logging system).

The starting date: This allows the observer to know, for instance, the period during which the aforementioned information is relevant.

The current configuration and how it’s changed across executions: This allows quick identification of incorrect settings.

The application execution directly connects to other entities, such as the application (what), the server (where), and users (who). Because the observer is aware of the application execution, and its connection to the behavior of some data usage within the application (see Lineage Execution and Data Metrics), the observer will be able to:

Connect to the server to analyze how it may influence the execution, or even better, use the server observability information to gain this visibility without connecting to it (e.g., splunk, elk).

Know exactly which applications are running or used to run on a server, and analyze its performance or load.

Identify easily and quickly who has performed the execution and thus is likely to be able to help describe the situation.

But what about the connection with the application? The model presented in Figure 1. doesn’t connect application execution to the application, because what is most important for an observer to know about the execution of an application is its version. Additionally, because the version refers to the source code, you can infer that it is also referring to the application.

Having the information about which version is being executed is extremely powerful. This information provides instantaneous insights, such as the version deployed is not the latest, or which changes may have caused an issue (e.g., git diff).

Also consider that an application is likely to be executed in different versions across different servers (e.g., environments), making the information described above even more valuable.

Lineage Execution

The Lineage Execution entity is the less obvious, but also the most complex, to create because it will likely have a large number of instances.

A Lineage Execution simply gives information about the execution of a lineage. Therefore it can represent many different things, such as the execution of an SQL job, a Python transformation, a copy command, machine learning training, etc. Lineage Execution is meant to provide the observer with explicit information about how a lineage behaves, including how often and what time connections to data sources occur.

I wouldn’t say that a Lineage Execution conveys a lot of information intrinsically but it is a keystone of the observation model. Without Lineage Execution, most of the use cases of data observability presented in Chapter 1 are fantasy, or rely on best effort reconciliation, which inevitably, leads to spurious correlations (mainly while reconciling the Data and Application Observability areas).

A lineage execution is rich in information about its connections to other core entities such as Lineage and Application Execution. If the link to Lineage is clear, then the observer can tell which lineage is observable and can analyze the link further.

A lineage can’t be executed alone because it only represents the connection between data sources (schemas) not how these connections are implemented. This information is available in the code, which can be observed using the Application Version. For example, a data transformation could have been implemented in Spark RDD, then in Spark DataFrame, then in Spark SQL, then a materialized view in Snowflake, etc.

To understand how a lineage is executed, the observer has to use information about the application execution. You can think of lineage execution as an internal program that is started by the application, for example an application creates an entry in a table when certain events are consumed from Kafka, then it executes the related lineage.

Moreover, an application is not limited to running a single lineage or a lineage only once or even always the same lineages. Because of this diversity, giving the observer visibility of this extreme (nondeterministic) situation is essential - as it is unthinkable that this complexity could be held in a person’s brain.

Due to the centrality of Lineage Execution in the model, the observer gains a holistic visibility (but not limited to) on:

Which Lineage is executed by…

Which Version of…

Which Application in…

Which Repository consumes and produces…

Which Data Sources having…

Which Schema executed by…

Which User on…

Which Server.

In graph lingo, we call this a traversal (collecting information while walking a graph). I call this a freaking mind-blower.

But we can extend further this visibility with data metrics. One of the exciting things about Data Metrics is that it has the potential to support data quality in a bottom-up, scalable, and real time fashion.

Data Metrics

Data Metrics is the specification of Metrics in Observability. Data metrics is often the most common suspect when discussing observability, as it is easily leveraged in analysis to infer or predict (future or dependent) events - e.g., using their temporality dependencies with multivariate time series analytics, Markov Chains, and the like.

In this context of Data Observability, I am focusing on the metrics that we can relate directly to data, such as descriptive statistics (see below for more examples). However, the data metrics entity comes with a subtlety that can have a huge impact on its usefulness–the metrics are always attached to their analytical (e.g., transformations) usage.

The role of Data Metrics is to give information to the observer about what the values of the data look like under certain conditions during their consumption or production. There are at least three types of metrics that can be considered:

Univariable: a metric related to one of the fields/columns of the observed data source. For example:

Descriptive statistics of a column (e.g., min, max, nulls, levels)

Distribution (e.g., density, skewness)

Formula (e.g., the sum of squares, product)

Multivariable: the result of the combination of several fields/columns. For example:

Formula (e.g, the sum of the products of two fields).

Combined (e.g., Kolmogorov-Smirnov)

Extrinsic: the result of the aggregation of fields with fields from other data sources. This metric is tricky, and can even be considered a borderline metric, as it is too close to a KPI or a result.

It is important to note that these metrics are not intended to replace the exploration analysis required during the analytical phase of a data project (e.g., data science, report). Those metrics are meant to allow the data to be understood in the context of their usage, not before. In other words, the primary usage of the data metrics is not to provide potential users with the initial insights they may find.

Nevertheless, data metrics are key for their potential users to gain trust in the likelihood of the data being reliable, because if the data source is reliable for the other usages, there is a good chance that their own usage of the data will also be reliable. Remember that trust (particularly related to data quality) requires time and is stronger when the reliability of the data is tied to real and understandable usages.

Note

The likelihood in the sentence above is important because it is not fully granted that if a data source is reliable for the applications of user X, it will also be reliable for the applications of user Y. Users X and Y are likely to use the data differently. For example, imagine if user X computes the arithmetic mean of a field and user Y its geometric mean. In this case, how would a single unexpected zero impact each model⁵? And what if the data source is suddenly empty, how would that also change the impact on the different use cases⁶?

Hence, the data metrics must be linked to the data source it is exposing the behavior (the correct term should be status) to the observer. In Figure 2-1, I show that data metrics are linked to the Schema, this is because the metrics are highly dependent on the fields present in the data source during its usage.

Finally, those metrics have to be linked to the usage of the data source, which is represented by the Lineage Execution, which represents when the data sources combination is executed. This link comes with a never-seen-before power in observation models–giving the observer the ability to understand the status of a data system.

Expectations

So far we have covered the types of information and the entities that give an observer the ability to understand how data behaves in a system.

In this section, I am going to address the third component of data observability, which pertains to creating visibility when the data behaves as expected. As you’ll see, the expectations can come from different parts of the data ecosystem (e.g., users, applications, …), and if expectations are visible to an observer, it gives the observer key insights into how the observed status of the data compares to how it is expected to behave.

By setting specific expectations, just as tests do in software development, the observer can better understand the behavior of the data aligns with expectations. Moreover, if some expectations are not met, the observer knows instantly what requires their attention, as, again, tests reports indicate to software developers where they need to focus their attention.

So they give essentially visibility about what the observer expect to happen, and they represent either a positive or a negative event. For example, knowing that an expectation, such as address always starts with the number, is true for 99% of the entries, ensures that the data that can’t be used is only 1%, there is no incidents, it just provides the observer two informations: the expectation itself and if it is true or false. On the other hand, if a transformation is known to fail if a secondary is not complete, in this case, when it becomes false, an incident can be clearly created, triggered and an associated decision can be made (e.g., not executing the transformation at all).

When setting expectations, you will want to both set rules for how the data should behave as well as detect anomalies. The next section will look at both of these areas of Expectations as well as how an application can become its own observer and make decisions regarding the status of the data’s behavior.

Rules

A Rule is a common tool used in development and monitoring, it is basically a function that evaluates the state of the data. If the current state passes the rule, it will return a “true” response, and if it doesn’t, it will return a “false.” The function’s logic is often quite simple, making it so that interpreting the result is easily relatable to the inputs. For example, rules using a machine learning model to return the result are closer to an anomaly detection than rules.

Note

The Oxford’s definition of “rule” is One of a set of explicit or understood regulations or principles governing conduct or procedure within a particular area of activity.

I don’t consider a trained machine learning model to be explicit, even if a fully explainable one would be borderline.

Therefore, rules for data observability would be created using information conveyed by Schema, Data Metrics, Lineage, etc. For example, if a Data Metric reports the number of null values for the field age, a rule could be the number of null values for age must be below 10, or less than 5%, or even can’t increase by more than 0.01% between two executions of the same lineage. Another example of using the Schema information would be the age field must be present and of X type of integer.

In fact, there are two categories of rules:

Single-state based

These rules only act on the information about the state of the system at a given time, such as in the above example where null ages must be below 10. The state information could also be combined to create intrinsic checks, such as comparing the number of null values to the total count (e.g., no more than 10%).

Multi-state based

These rules require more than one state, such as the states of one or several periods of time, continuous states, or a sampling. Multi-state based rules could even include a list of states randomly drawn from the available states.

While the entities of the Observation Model are computed or extracted from available information, the rules are less factual and require more insight to create. The insights needed can come from several sources such as:

Human knowledge and experience

History of the system

Similarities with other systems

Let’s look at two ways to make rules.

Explicit Rules

Explicit rules are mainly created by users as a representation of assumptions that they are building up and which they expect to hold true when their applications will run and be used in production.

Those assumptions come mainly from the following channels.

Experience

A person’s experience with the data or the addressed use case is a great way to generate rules because it is based on past experience or applied knowledge. I am referring to intuition or gut feelings that are the result of years of experience forming assumptions based on use cases or practical experience (e.g., networking, etc). However, it doesn’t mean that those assumptions are always reliable. A person may be biased or misunderstand the full situation. Nevertheless, it is still worth considering these types of “instinctual” assumptions as they are far better than nothing and, more importantly, they express the expectations of the user. Then, if those rules are not met, thanks to the observability of the system, the team has the opportunity to perform an analysis to determine the rules’ validity and potentially discard or adjust them.

Exploration

While working with the data to create the pipeline, the report, features, and other types of data transformations, a person will be learning, or at least better understand, the data at hand⁷. Therefore assumptions will be formed and integrated into the application based on these learnings. For example, if the data has no duplicates or a turnover always above zero, if you turn them into assumptions, you are likely not going to make them explicit or may not be specified in the business logic to be implemented–although those assumptions are a great source for rules. One way to help this process for developing rules could be to create data profiles (using profilers). Profiling data generates a series of static rules inferred from the data provided. However, these rules need to be used with caution and reviewed thoroughly, leveraging the experience of the team to select the most appropriate ones while avoiding introducing rules too close to the data at hand and therefore likely to fail rapidly in production. The usage of data profilers will be detailed further in Chapter 3.

Discovery

When application and data are in use, new cases can be discovered, such as unknown-unknown (unexpected cases) or corner cases (not considered in the business logic). When discovered, they are generally considered as incidents that need to be analyzed, and most likely fixed. As discussed in Chapter 1, there are several processes that must be conducted such as root cause analysis, troubleshooting, and impact analysis, to determine and fix the issue. Each of these analyses present opportunities to introduce new rules discovered along the pipeline.

It is also important to note that rules can be outdated as the world is constantly changing (e.g., products are boycotted, laws are adapted, platforms are upgraded). Consequently, the maintenance of these rules must be considered. I often encourage teams to perform a quick review of the rules, especially for experience-based ones, with every new version of the application using or producing the data at the very least. Validating rules justly requires a simple feedback flow; however, there are alternatives such as using assisted rules.

Assisted Rules

In the previous section, I introduced rules created entirely as the result of human analysis. Here, I will introduce an alternative method, which allows rules to be discovered.

Although rules cannot be based on non-explicit behavior, they can still be discovered using simple analysis (e.g., heuristics) or even learned (e.g., association rule mining). Such rules are created with the assistance of the observations. In this way, they are “data-driven.”

The power of assisted rules is not only the capability to create new rules, but also to update existing ones, even the explicit ones. Done efficiently, assisted rules lessen the burden required to maintain rules in the long run.

The way to introduce assisted rules in a data observability system is by keeping humans in the loop. Because rules are structurally comprehensive, they can be reviewed by people who have the knowledge to tune, accept or reject them. Validating rules in this manner is important to increase the level of trust in the rules, and ultimately, the accuracy of the issues the rules detect. For example, an assistant system can estimate (e.g., based on self-correlation) that the amount seems to take its value around 19,945.62 with a standard deviation around 783.09, and therefore propose a few rules such as following some distributions– which could also either estimated from observations, or be an observation itself like the Kullback–Leibler divergence with a few distributions a priori. The possibilities are limitless, as long as we have enough observations at hand. That said, assisted rules should, over time, also leverage knowledge accumulated during the discovery processes, that is the rules created after the fact, which provides information on how to anticipate unknown-unknowns.

The opportunity to recommend and update rules is an advantage over automated anomaly detection, which I will discuss in further detail. However, such anomaly detection is, by definition, not involving the human much, the lack of control they provide and their randomness are not generating the level of certainty about what is under control. To increase and strengthen this feeling, which is the goal of Data Observability, I will first tackle the relationship between rules and Service Level Agreements (SLAs).

Connection with SLA/SLO

Service level agreements (SLAs) and service level objectives (SLOs) have been around for quite some time to establish contracts and are a way to avoid subjective disputes about the expected quality of the service. They help build a sustainable relationship between producers or service providers with their consumers and their users.

These relationships also exist in the data world. Data teams are, in a service fashion, producing data for dedicated use cases or to be made available for future consumption.

Thus, both data teams and data consumers/users have expectations, even if they are unspoken, undocumented, and not intuitively shared when it comes to data. This has created the issues I covered in the first chapter. SLAs and SLOs are good solutions to eliminating many of these issues.

Service level agreements for data are contracts between the parties involved comprised of expectations that can be met, and if not, a fine or other penalty is applied. The fine/penalty is potentially as important as the length of time that elapses in which the expectations are not met. Hence, operational metrics such as “time to detection” and “time to resolve” are key for sustainable success.

To establish an SLA in data, it is a prerequisite to have the input of the users to make sure the quality of the service will be high enough for them to actually use the data. The more users taken into account in the SLA, the greater the service, the higher the service levels, and the better the relationship is - as long as the associated cost to ensure them is acceptable.

There are several complex facts that need to be considered when establishing the SLA for data:

How to define the constraints

The user may have a hard time defining the important constraints due to the number of possible constraints. Considering that each field in a data source can have 10+ associated metrics (e.g., for age there can be min, max, mean, std but also the quantiles, the estimated distribution, the null values, the negative ones, etc).

Data usage will vary

Each user will have a different use case for the data, resulting in a different set of expectations and thus constraints.

High number of users

With self-serve data and analytics, the number of users is exploding, thus creating more data use cases and more constraints.

This is where data observability is essential. Data observability observations can be used as Service Level Indicators (SLIs) by providing enough information about both the data and user behavior based on which rules can be identified, assisted, and maintained.

Because both producer and consumer generate these observations, it becomes easier to find a consensus as to which KPIs from the SLIs that should be used to determine which SLAs are possible for the data team to meet. This will set the bar for the SLOs across all SLAs while keeping the door open for other SLIs to be used later during discovery of SLAs. In the meantime, both consumers and producers can use them as SLOs (or simply informative, through a notification system for example).

Without data observability the scope is too vast for both parties, which prevents them from committing to a set of expectations without sound evidence of their effectiveness, and a way to improve over time.

Automatic anomaly detection

Rules are a powerful way to ensure data matches expectations in the context of where the data is used. However, rules have a dependency on people to create and validate them (at least, this is what I strongly recommend). This dependency, on people’s time and availability, is undoubtedly a bottleneck to scale exponentially with data validation. Although, companies reaching this stage need to have already reached a high maturity level.

Also, assisted rules have a constraint on their structure as they must be explicit and fully explainable, which may be a weakness over other more opaque solutions, such as learning-based rules.

Considering that rules are meant to intercept expectations that are not met, another method we haven’t discussed yet is automatic anomaly detection using the data observability information available.

An anomaly is the state of the system which is different from the expected states. This is the connection between expectations explicitly encoded as rules and expectations implicitly discovered when unmet.

In our case, we have the states of the system corresponding to the history of the data observability information based on the core observation model. Hence, those states can also be seen as a stochastic process with many random but connected (non independent) variables (each instance of the entities could be considered as a variable).

Automatic anomaly detection, therefore, is the process of leveraging some of the learning-based methods listed below to infer or predict when a state should be considered an anomaly.

The methods that can be used nowadays are numerous and include machine learning:

Supervised learning

This requires labeling data observability information (or the state itself) to work properly. Hence, it requires time to collect enough data to train the algorithm as well as time from the team to label (with high accuracy) the anomalies.

(semi-)Unsupervised learning or statistical analysis

This method can help identify or predict potential abnormal states. These states can then be presented to the team for labeling. However, it is cumbersome to label the predictions as anomalies automatically, even in extreme cases (e.g., outliers), without a second review.

Reinforcement learning and active learning

This method is quite advanced and still requires some time from the team. Not only would agents need a lot of simulations to be validated, the active learning is also essentially based on the human in the loop.

My intent is not to dig into each method, because it would require at least another book and there are already enough papers and books on each of those topics. However, let me emphasize that these methods, although capable of scaling in theory, still come with downsides.

The cold start problem

In order for anomaly detection to be efficient, data and people are needed to ensure the learning phases are reliable.

The black box worry

The above methods generate anomalies automatically but with little information about the reasons. Consequently, the team must reconstruct the context and understand the reasons.

The performance challenge

Anomaly detection can’t be 100% accurate. Therefore, their performance needs to be under control to avoid alert-fatigue, where the team starts to ignore the detected anomalies.

Coming back to the black box worry, it is important to note that nowadays a lot of effort is put into making machine learning models explainable (which is, in my humble opinion, a component of analytics observability). This is a challenge that is not fully solved, especially for methods such as deep learning. However, keeping an eye on the evolution of this capability is a must to reduce the effect of the black box worry.

I’ll wrap up this section on anomaly detection by also mentioning another method rarely considered to address the black box worry, and moreover, provides an interesting mix of rules and anomaly detection – the learning of finite state machines. This method could be promising under circumstances where data observability information is growing in volume, especially if the rules having triggered are also labelled as true or false positives.

Prevent Garbage In - Garbage Out

If there is one cause for which the data community should be fighting it is the ban of the “Garbage in-Garbage out” excuse. This common excuse is generally used as the last line of defense when a data producer is being asked why the data is wrong. Often, with too much confidence, the explanation provided is, “you know, my application that produces the data works as expected, it follows the business logic requested, but it was provided with garbage data, so of course, it produced garbage results.”

Although this excuse is used as an explanation or a way to waive the blame, it does little to placate the user who still has no real solution at hand, so it is considered, appropriately, as an excuse for not meeting data quality expectations.

In other words, it is a “dodge” that creates more harm than good. An opposing view is that if an application can be aware garbage comes in, it should be forbidden to generate garbage out. To do this the application needs to be able to

Assess provided data as garbage

Qualify the created data as garbage

The good news is that the third component of data observability, the expectations, addresses these exact requirements.

With Data Observability implemented, here are the different scenarios that are possible:

Incoming data is classified as garbage before the application generates its results. The application can then either:

Not proceed and avoid the garbage out

Make best efforts to clean the data, ensure the cleaning process is observable, and then proceed

Incoming data is classified as garbage based on the qualification of the results. The application can then either:

Roll back the data and log it

Log why the data is garbage, due to a posteriori discovery of garbage in

Like in the first case, try to clean the incoming data or its output, still ensuring those processes are observable

The logic described above could be implemented using the two, non-mutually exclusive procedures. The first is the use of conditions.

Pre/Post-Conditions

Conditional statements are commonly used in programming, such as when implementing an HTTP endpoint. For example, if the endpoint’s business logic is expecting a JSON payload encoded in UTF-8 that contains a non-null integer age field under the customer top-level field and these conditions are not met, the application can decide to return an error to the client or be lenient for certain cases such as the encoding by re-encoding before proceeding. Also, if the HTTP endpoint’s side-effect is to update a table and then get the current state of the data, but it gets several rows instead of one, it can take actions such as rollback, delete, or simply return an error letting the client know about the potential garbage data it sent.

What I just described is something very natural for any programmer, it is even more than a best practice, it is intuitively used. Therefore, the idea would be to simply also use this pattern when creating data applications, pipelines, and so forth.

The interesting advantage of implementing pre and post conditions directly within the data application is that it is well guarded without depending on other external components to ensure its viability. Of course, the system used to build data needs to allow such practices, which have not systematically been addressed in some systems, especially the low-code no-code (e.g., ETL, reporting tools).

Note

In recent years, not only because of the influence of DataOps, which is a set of methods and tools to go to production faster, the data system has reduced the entry-level required by people to use the data. Therefore, the practice of setting of pre/post conditions is not necessarily well-known or seems like an additional unnecessary burden – as unit tests used to be considered in the early 2000s.

On the other side, the whole world is putting a significant focus on the culture of accountability and respect for data.

In my opinion, accelerating time to production, lowering the skills, and increasing the accountability are not playing well together, unless the data systems allow the people to also mature and take some responsibility. Some technologies have partially included this concept, such as dbt, which includes data tests but with the current limitation that tests are not executed in production. This is addressed further in this book (REF).

In the next section, I describe the circuit breakers, which conceptually are not so different from the conditions described here, but structurally give an alternative when the system does not (yet) have first-class support for the conditions.

Circuit Breaker

The circuit breaker originates from electrical circuits. In electrical systems, the circuit breaker is a safety device designed to protect an electrical circuit from damage⁸. This device closes the circuit to allow the electric current to flow until certain conditions are met that could break the circuit. In such an instance, the device breaks the circuit (opens it), forbidding the current to circulate until the underlying issues are fixed and an action is taken to close the circuit again.

This idea has been reused in software architecture by Michael Nygard⁹ and supported by Martin Fowler¹⁰, where an additional component wraps a function to prevent its execution under certain conditions. The prevention is as simple as returning an error directly when the wrapper is called and the circuit is open.

In a data pipeline, this may be more of an alternative to pre/post-conditions, especially in the case where the user has no capabilities to introduce conditions into the system used to create the pipeline, such as in low-code-no-code systems or restricted interfaces (whether graphic or API). Therefore, the system has extra components in the pipeline (circuit) to break it in case conditions are not met. In fact, implemented as such, circuit breakers are similar to pre/post-conditions external to the applications.

The nice thing with a circuit breaker is that they seem to be easier to add after the fact (especially good for legacy systems). However, they do come with downsides. For example, if the conditions depend on the steps before and after (e.g., case specific SLA/Os, custom checks, …), then the circuit breaker is yet another application to maintain and align with the steps’ logic¹¹. Hence, circuit breakers are likely to be present before and after each step, which makes the pipeline three times as long as it would have been if the conditions had been added in the applications directly.

¹ Cindy Sridharan, Distributed Systems Observability, O’Reilly 2018

² Majors, Fong-Jones & Miranda: Observability Engineering, O’Reilly 2022

³ The name Schema relates to concepts such as XML-Schema, JSON-Schema and alike.

⁴ As in ontology’s concepts.

⁵ The user Y ends up with the value 0, whilst the value of X has probably not changed a lot (especially if there are many values).

⁶ Boom… division by 0. At least, Y would get an error this time :-).

⁷ Data at hand refers to the data available during the development phase.

⁸ https://en.wikipedia.org/wiki/Circuit_breaker

⁹ https://learning.oreilly.com/library/view/release-it/9781680500264/

¹⁰ https://martinfowler.com/bliki/CircuitBreaker.html

¹¹ This is likely to result in troubles…

Chapter 3. Getting Started with Implementing Data Observability

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 author at gobrien@oreilly.com.

As explained previously, data observability combines technology and people to gather information about the state of a system and expectations of that state. It then uses the information to make the system more adaptable or resilient.

This chapter explains how to apply the data observability practice. It starts with a method to introduce data observability in your day-to-day data work and shows how to minimize the impact on efficiency. Then, it elaborates on a process that will subscribe to the delivery life cycle, such as continuous integration and continuous delivery (CICD).

As with any emerging practice and technology, to increase data observability adoption, lower the barrier to entry is needed, but still make people part of the solution and process is crucial. To this point, you’ll learn several ways to reduce efforts to generate observations and understand when to introduce them at the right moment of the development life cycle.

Data Observation Collection Strategies

Chapter 2 explained the sources and types of information that help the observer. But how do you generate and collect the information from these sources?

It starts with data observability at the source. This concept means the applications—the sources of data generation—are given the ability to generate data observations following the model covered in Chapter 2.

Pure data and analytics observations—such as data sources, schemas, lineages, and metrics—correlate to read, transform, and write activities. The strategies described in this section address these activities by explaining what to consider when running them.

Generating Data Observation at the Source

The baseline of this method is adding instructions when developing your application, whether in Python, Java, or any other language. Developers can add lines of logs that contains the information needed to generate visibility about what their code is doing. The instructions use the channels—that is, the logs, metrics, and traces or lineages—covered in Chapter 2 to convey the observations that can be centralized in a logging system.

In this section, you learn how to create data observations in the JSON format that can be collected (written), for example, to local files, a local service, and a distant (web) service. In addition to the JSON format, you can use another format, such as AVRO or Protobuf, if you share a model (cf. Chapter 2) across applications in your environment.

This ability to collect data observations in the same model on a centralized platform is key to generating the value of data observability at scale, such as across teams. That’s why using the data observability core model is important to aggregating data observations with ease, especially along lineages.

Let’s get started with the strategies to generate data observations. The first one is the most tedious but essential to understand because it’s often used as a fallback when higher-level abstractions (covered later in this chapter) are too constrained. Even then, you’ll learn how to generate each entity of the data observation core model and which observations to convey.

Low-level API

This strategy of using a low-level application programming interface (API) that requires the most of your time and involvement because you are responsible to create every observation explicitly. However, this strategy also gives you the greatest flexibility because it doesn’t involve any higher-level abstractions.

The development, especially during exploration and maintenance, requires a rigorous approach. You must also bring the changes applied to the logic or behavior of the application in the reported observability information. Examples of this information include connecting to a new table, generating a new file, or adding a new field.

In the following sections, you go through a complete example of data applications written in Python that generate data observations alongside their data usage. First, you see the applications without the data observability capability and learn what they do and what can happen in production. Next, you see how to add the instructions to generate data observations and learn what they are used for. Then, you gain insights into the pros and cons of using this strategy.

The pipeline that drives this explanation is available in the GitHub repository https://github.com/kensuio-oss/kensu-public-examples, in the Financial_Data_Report/python_nodo_code folder. The repository has two options:

Python/pandas, which is used in this example.

Apache Spark (PySpark).

For simplicity, the pipeline (Figure 3-1), which manipulates only CSV files, has two applications: ingestion and reporting.

Figure 3-1. [Insert Caption]

The ingestion and reporting applications transform the data from the “Daily stock prices” into the Buzz stock report and the AppTech stock report. Both applications use Python and the pandas library to read, transform, and write the results.

The ingestion application reads the CSV files for the daily stock prices that the Stock Market team provides each month. The team places the files in a folder structure by year and month and then merges the prices into monthly views stored as a separate file as shown in ingestion code below:

import sys
month = sys.argv[1]
year = sys.argv[2]
import pandas as pd
Apple = pd.read_csv(
    "../datasources/%s/%s/Apple.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
Buzzfeed = pd.read_csv(
    "../datasources/%s/%s/Buzzfeed.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
EURUSD = pd.read_csv(
    "../datasources/%s/%s/EURUSD.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
Microsoft = pd.read_csv(
    "../datasources/%s/%s/Microsoft.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
iMetal = pd.read_csv(
    "../datasources/%s/%s/iMetal.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
AppTech = pd.read_csv(
    "../datasources/%s/%s/AppTech.csv" % (year, month),
    parse_dates=["Date"],
    dtype={"Symbol": "category"},
)
monthly_assets = pd.concat([ Apple, Buzzfeed, EURUSD, Microsoft, iMetal, AppTech ])\
.astype({"Symbol": "category"})
monthly_assets.to_csv(
    "../datasources/%s/%s/monthly_assets.csv" % (year, month), index=False
)

Example 3-2. Code 3.0.1

The ingestion application processes two arguments: the month and the date. Then, it runs several lines of almost identical instructions to read the CSV file for the following stocks:

Apple

BuzzFeed

EURUSD

Microsoft

iMetal

AppTech

The application reads the files from the “../datasources/%s/%s/ … % (year, month)” folder at a location that depends on the year and month being processed.

During exploration, our data engineer determined that the files have two fields that need to be preprocessed. The “Date” field must be parsed as a date, and the “Symbol” field must be treated as a category.

The application reads and processes the daily stocks before making them available in memory. Then, application merges them into into a single variable, which in this case is a new pandas DataFrame, containing all categories in Symbol.

The script writes the result into the same folder as a new file named monthly_assets.csv. This way, other analysts or data engineers can start there if they need all daily stock prices for further analysis.

After this first script runs and the file is available, the remaining pipeline scripts can run to generate the “BuzzFeed” and “AppTech” stock reports. This example has only one script, the reporting Python file:

import sys
month = sys.argv[1]
year = sys.argv[2]
import pandas as pd
all_assets = pd.read_csv("../datasources/%s/%s/monthly_assets.csv"%(year,month),parse_dates=['Date'])
apptech = all_assets[all_assets['Symbol'] == 'APCX']
Buzzfeed = all_assets[all_assets['Symbol'] == 'ENFA']
Buzzfeed['Intraday_Delta'] = Buzzfeed['Adj Close'] - Buzzfeed['Open']
apptech['Intraday_Delta'] = apptech['Adj Close'] - apptech['Open']
kept_values = ['Open','Adj Close','Intraday_Delta']
Buzzfeed[kept_values].to_csv("../datasources/%s/%s/report_buzzfeed.csv"%(year,month),index=False)
apptech[kept_values].to_csv("../datasources/%s/%s/report_AppTech.csv"%(year,month),index=False)

This application performs the following actions:

Reads the data source created by the previous application for a particular month.

Filters out the data for the two reports it creates: BuzzFeed and AppTech.

Computes the “Intraday_Delta” scores, which are the daily stock evolutions to report.

Reports the “Open”, “Adj Close”, and the computed score “Intraday_Delta” values in the dedicated files for each stock.

The codes are available from code repositories, such as the GitHub repository mentioned before. However, both applications are likely independent of each other because they’re developed by different individuals, teams, or domains.

To run the pipeline, deploy and orchestrate the applications carefully, with respect to the dependencies between the reporting and ingestion applications. That is, you must run the ingestion application successfully before you run the reporting application. Otherwise, the pipeline will fail. To run ingestion before reporting you will use orchestrators, such as Luigi and Airflow. The orchestrator, which is the third application you need to configure, turns the implicit dependency between the applications into an explicit one, although the applications still aren’t aware of this dependency.

Hardcoding a pipeline solves problems, like ordering and scheduling. But it creates other problems because it’s a new asset to manage, involving responsibility, lifecycle, management, and synchronization. These aspects can become constraints when creating a pipeline because some teams must stay independent. Or because new applications are created at a later stage, they must follow the principles of reusable or self-serve data.

Also, as mentioned, the reporting application can run only after you’ve successfully run the ingestion application. That is, you must meet two conditions before running the reporting application: first, run the ingestion application, and second, make sure it runs successfully. What does successfully mean?

To answer this question, you need to analyze each application individually and review cases when ingestion can’t be considered successful. Unsuccessful ingestion results when one of the following types of failures occurs:

Explicit failure, which causes the application to crash. This type of failure is easy for the orchestrator to handle if you use one because it’s a flag not to trigger the next application—the reporting application in our example.

Silent failure, in which the application finishes, for example, without an error code or log. Because it doesn’t run as expected, you must consider the notion of expectations introduced in Chapter 2.

An observer of the ingestion application would face explicit failures like:

File not found errors: Occur when any of the data files, such as Buzzfeed.csv, aren’t available in the folder because it was renamed or changed to lowercase, or the file wasn’t created before running the ingestion application.

Type errors (TypeError): Occur if some values aren’t coercible to match the types provided to the read_csv function, such as when the Symbol should be a category.

Hard-coded name errors: Occur when any of the fields explicitly used in the code to access the values, such as Date and Symbol in this case, aren’t present or the name has changed.

File system errors: Happen when the files aren’t readable or the folder isn’t writable for the user who runs the application.

Memory errors: Happen when the files increase in size to the point the memory allocated to the application is no longer sufficient.

System errors: Are triggered if the disk doesn’t have any more space to write the aggregated result.

But from the perspective of the engineer who observer it, the following examples indicate silent failures:

The “Date” column isn’t parsable as a date because it’s malformed, the pattern changed, or the time zone isn’t consistent. In this case, the column isn’t a datetime anymore but an object. This application doesn’t use it yet, but it might use it later (see below).

The “Date” column contains values but not for the current month. All values are past or future dates.

The “Date” column contains values in the future because the generator might have run later and generated information in the future compared to the month in which it’s processing. This situation can generate duplicates later or inconsistent values for the same dates and can likely fail some aggregations.

The “Symbol” category changes for various reasons, such as because of a typo or a capitalization change, length, or reference. This category is used across the files and written in the output file as a category.

Also, the reporting application might regard the ingestion as failed because it caused the following failures:

The monthly aggregated file wasn’t written when the reporting started, making it unavailable.

Any of the fields used to filter or in the arithmetic aren’t available or their name changed.

The same errors occur with read/write access, size, and space.

Or these silent ones, but first, all silent failures of the ingestion files are also valid here, because they are propagated, then:

The ‘APCX’ and ‘ENFA’ symbols aren’t available in the provided file.

‘Adj Close’ or ‘Open’ are missing values, which generates garbage in the ‘Intraday_Delta’ output. This problem can also be an explicit failure, in which case the values become pandas NAs.

The aggregated file doesn’t contain information about the current month, but the dates are in the past.

Because any of these failures can occur, you must have clarity when they happen and—even better—anticipate them earlier to prevent propagation if they are detected earlier (see Fail Fast), such as those identified in the ingestion application.

The first part is about adding observability to our application. The next section covers the expectations.

But first, let’s look at Figure 3-2, which shows how, by using a low-level API, you can generate the visibility you need to manage errors more easily when they occur. It produces the observations according to the model introduced in Chapter 2.

Figure 3-2. [Insert Caption]

In this diagram, you’ll notice color entities. The green colored Data Sources highlight the observations generated by the ingestion application about the data it generates, and by the reporting as it is reading it. So, that specific Data Source observation in green represents explicitly the data dependencies between the applications.

In fact, both applications generate several similar observations, which represent all dependencies that tie them together, in Figure 3-2 are also highlight the following identical observations:

In Yellow: the Server used to retrieve the data from.

In Orange: the same application user is executing applications.

In Light-Blue: the schema of the data generated by the ingestion read by the reporting.

Let’s dig into what we need to add to the application’s code to generate the observations shown in Figure 3-2. Because the code is in Python, we’ll use the logging module to print the observations that are encoded in JSON. At the end, we’ll review the example issues mentioned previously and see how the observations in the code help.

In looking at the ingestion application and its execution context, the observations shown in Figure 3-3 are reusable for the reporting application and any others.

Figure 3-3. [Insert Caption]

Insert the following code at the beginning of the file to generate the observations for the ingestion application:

# retrieve information about current user
import getpass
app_user = getpass.getuser()

# retrieve information about application code from git
import git
# - the remote git repo
code_repo = git.Repo(cur_dir, search_parent_directories=True).remote()
# fetching the latest commit
commit = git_repo.head.commit
# - the latest commit hash is used as version (note: we could use something from setting.py alternatively)
code_version = commit.hexsha
# - the latest commit author is considered as the user connected to this version
code_author = commit.author.name

# checking if there is any changes in the current git cloned
if git_repo.is_dirty():
    # if the code is dirty with flag the version and overwrite the commit author with the current user
    code_version = code_version + " (dirty)"
    code_author = app_user

# retrieve application name as the file name
application_name = os.path.basename(os.path.realpath(__file__))

# retrieve information about server on which this application runs on - example, using IP
import socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.settimeout(0)
server_socket.connect(('10.254.254.254', 1))
server_ip = server_socket.getsockname()[0]
server_socket.close()

# retrieve starting time of the application
import datetime
application_start_time = datetime.datetime.fromtimestamp(timestamp/1000).isoformat()

Example 3-4. Code 3-1

These extra code instructions create variables for the observations, but nothing is done yet with them. To log the information as mentioned previously, we use a JSON representation of the information model encoded as follows:

application_observations = {
    "name": application_name,
    "code": {
        "repo": code_repo,
        "version": code_version,
        "author": code_author
    },
    "execution": {
        "server": server_ip
        "start": application_start_time,
        "user": app_user
    }
}
import json
logging.info(json.dumps(application_observations))

Example 3-6. Code 3-2

This code logs the JSON that’s formed with all the observations created so far. However, this section is about using a low-level API for data observability. So, we can introduce the observations and add the functions that will allow us to standardize the approach across the ingestion and reporting applications and any other Python applications.

To create an API, we create a model that mimics the observation core model in JSON, turns each entity into a class, and converts relations into references. That means the Application entity will have an Application class with a property name that would hold the name of the file as an application_name variable. Also, the Application Repository entity will be encoded as ApplicationRepository class that has a property location set with the git remote location and a reference to an Application instance. This structure will help build the model and generate a JSON representation more easily that will be reusable and lead to standardization. Here’s how these two classes look:

# make classes JSON serializable: https://stackoverflow.com/a/38764817/193246
from json import JSONEncoder
def _default(self, obj):
    return getattr(obj.__class__, "to_json", _default.default)(obj)
_default.default = JSONEncoder().default
JSONEncoder.default = _default
# Class encoding the Application Repository entity
class ApplicationRepository:
    location: str
    application: Application
    def __init__(self, id: str, location: str, application: Application) -> None:
        self.location = location
        self.application = application
    def to_json(self):
        return {"location": self.location}
# Class encoding the Application entity that refers to its repo
class Application:
    name: str
    def __init__(self, name: str, repository: ApplicationRepository) -> None:
        self.name = name
        self.repository = repository
    def to_json(self):
        return {"name": self.name, "repository": self.repository.to_json()}
# instantiationinstantiation and reporting (log)
app_repo = ApplicationRepository(code_repo)
app = Application(application_name, app_repo)
logging.info(json.dumps(app))

Example 3-8. Code 3-3

An additional benefit of encoding concepts into API classes is giving them the responsibility to propose helpers to extract associated observations as shown in the following code:

# Class encoding the Application Repository entity
class ApplicationRepository:
    location: str
    application: Application
    def __init__(self, id: str, location: str, application: Application) -> None:
        self.location = location
        self.application = application
    def to_json(self):
        return {"location": self.location}

    # Add static method to fetch the git location
    @staticmethod
    def fetch_git_location():
        import git
        code_repo = git.Repo(cur_dir, search_parent_directories=True).remote()
        return code_repo

# Class encoding the Application entity that refers to its current repo
class Application:
    name: str
    def __init__(self, name: str, repository: ApplicationRepository) -> None:
        self.name = name
        self.repository = repository
    def to_json(self):
        return {"name": self.name, "repository": self.repository.to_json()}

    # Add static method to fetch the name as the file name of the current script
    @staticmethod
    def fetch_file_name():
        import os
        application_name = os.path.basename(os.path.realpath(__file__))
        return application_name

# instantiationinstantiation and reporting (log)
app_repo = ApplicationRepository(ApplicationRepository.fetch_git_location())
app = Application(Application.fetch_file_name(), app_repo)
logging.info(json.dumps(app))

Example 3-10. Code 3-4

This strategy might be a straightforward way to implement the model. However, we prefer another approach that weakens the links between entities. In the previous example, all information will be logged all in one JSON with entities spread into the tree of information that has Application at its root. This encoding forces us to create all observations before logging the root one, which is the Application instance. The Application constructor would become something like:

class Application:
    name: str
    def __init__(self, name: str, version: ApplicationVersion, repo: ApplicationRepository, execution: ApplicationExecution, server: Server, author: User) -> None:
        pass

Example 3-12. Code 3-5

To avoid this complexity and constraint, a better way is to reverse the dependencies between the entities. Instead of having the Application contain its ApplicationVersion or ApplicationRepository, we’ll create the Application alone and then add a weak reference to it from within the ApplicationVersion and ApplicationRepository. The following example shows how this model would look:

import uuid
class Application:
    id: str = str(uuid.uuid4())
    name: str
    def __init__(self, name: str) -> None:
        self.name = name
    def to_json(self):
        return {"id": self.id, "name": self.name}
    # Add static method to fetch the name as the file name of the current script
    @staticmethod
    def fetch_file_name():
        import os
        application_name = os.path.basename(os.path.realpath(__file__))
        return application_name

class ApplicationRepository :
    id: str = str(uuid.uuid4())
    location: str
    application: Application
    def __init__(self, id: str, location: str, application: Application) -> None:
        self.location = location
        self.application = application
    def to_json(self):
        return {"id": self.id, "location": self.location, "application": self.application.id}
    # Add static method to fetch the git repo
    @staticmethod
    def fetch_git_location():
        import git
        code_repo = git.Repo(cur_dir, search_parent_directories=True).remote()
        return code_repo

# instantiationinstantiation and reporting (log)
app = Application(Application.fetch_file_name())
logging.info(json.dumps(app))

app_repo = ApplicationRepository(ApplicationRepository.fetch_git_location(), app)
logging.info(json.dumps(app_repo))

Example 3-14. Code 3-6

With this model, we can log the observations individually—the two calls to logging.info—reducing the amount of information to retain. Because we need to recompose the entity relationships, we introduce the id variable to reduce the amount of information to log and observations to link. By using the logs, id can reconstruct the links in the model like the dependency between the ApplicationRepository and the Application by its id, because it has been already been logged.

In this example, the application generated id locally, resulting in a poor design that will make it inconsistent across several executions. To circumvent this problem, we must define a functional id that can identify the entities across executions, deployments, and applications. This notion is the primary key in modeling. You can use the primary key as input in a hashing algorithm, for example, that will generate the id in a deterministic fashion using hashlib in this case.

The following code excerpt shows how to use the primary key to generate the id consistently by using md5 for example. We’ll use this strategy throughout the rest of this chapter when generating entities.

import hashlib
class Application:
    # primary key
    name: str
    id: str
    def __init__(self, name: str) -> None:
        self.name = name
        self.id = hashlib.md5(self.name.encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "name": self.name}
    # Add static method to fetch the name as the file name of the current script
    @staticmethod
    def fetch_file_name():
        import os
        application_name = os.path.basename(os.path.realpath(__file__))
        return application_name
class ApplicationRepository:
    # primary key
    location: str
    # primary key (the repo is then unique for a given application in our case)
    application: Application
    id: str
    def __init__(self, id: str, location: str, application: Application) -> None:
        self.location = location
        self.application = application
        self.id = hashlib.md5(",".join([self.location, self.application.id]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "location": self.location, "application": self.application.id}
    # Add static method to fetch the git repo
    @staticmethod
    def fetch_git_location():
        import git
        code_repo = git.Repo(cur_dir, search_parent_directories=True).remote()
        return code_repo

# instantiation and reporting (log)
app = Application(Application.fetch_file_name())
logging.info(json.dumps(app))

app_repo = ApplicationRepository(ApplicationRepository.fetch_git_location(), app)
logging.info(json.dumps(app_repo))

Example 3-16. Code 3-7

Note

INSERT GITHUB REF API: Chapter 4 covers the implementation of API principles for several languages and tools. They rely on the INSERT GITHUB REF API on GitHub that encodes most of the entities presented in the core data observation model. If you use the API as is, it will simplify your process to define your model encoding.

The Server, User, and ApplicationRepository entities are pretty simple. They are included here as a thought experiment to move faster to the other observations of the main data observability components, which are Data Source, Schema, and Data Metrics. Figure 3-4 shows the observations we must log.

Figure 3-4. Fig 3-4

The observations to generate are related to the data source that’s read and written. Then, the transformation (lineage) is addressed. In the ingestion code, many sources are read by using the pandas read_csv function.

The observation will generate is the file path, which is provided as input to the read_csv function. Also, we could consider generating observations about the fact that read_csv is used without any specific parser attributes, such as sep, which tells the data must be in the CSV format.

Also read_csv isn’t given any information about the header. Therefore, the first non-blank is expected to be the column names, which will help create an observation about the data that the ingestion application accesses.

The type of columns is another piece of information that can be observable. pandas infers most of the numeric types. However, strings, categories, and dates are kept as objects. The read_csv function provides two hints: the Date is date information, and the Symbol is categorical data or a string in this case.

With the values, pandas creates a DataFrame, which is a sort of table we can use to catch the information listed so far. Moreover, the DataFrame also gives us the opportunity to compute additional intrinsic information about the values themselves, such as descriptive statistics. We use the describe function to compute these basic statistics on numerical values and other types like categorical or date values by using the argument: include='all’.

Note

Describe function: You have likely used the describe function many times when exploring and developing your application. This function is essential for knowing what you must do with the data to achieve the goal. This exploration, which is based on a set of assumptions about data expectations, leads to understanding the data. Therefore, logging the result of the describe function gives you the information to understand the data later if the assumptions about it hold true in the future.

Finally, the ingestion application writes the monthly assets CSV file by using the to_csv function on the DataFrame that holds the values to write. This function gives almost the same level of information as from using the read_csv function. We can derive the file path, columns, and types.

How data sources are read and written hints before one of the previously introduced cases happens. Now, let’s see how to use the ingestion application to efficiently generate those hints. First, we’ll see what classes we can introduce to generate the entities and which helpers we might need.

The first class is DataSource, which is relatively straightforward to model. We mainly want to know which data source the application is reading or writing. Therefore, it will model the path to the file and extract the server from it.

class DataSource:
    # primary key
    location: str
    # primary key
    server:Server
    format: str
    id: str
    def __init__(self, location: str, server:Server, format: str = None) -> None:
        self.location = location
        self.server = server
        self.id = hashlib.md5(",".join([self.location, self.server.id]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "location": self.location, "server": self.server.id}
    # Add static method to extract the server information from a file_path
    @staticmethod
    def extract_server(file_path):
        # overly simple, quick and dirty heuristic
        if file_path.startswith("file:/") or file_path.startswith(".") or file_path.startswith("/"):
            return Server(Server.get_local_ip())
        # can handle ftp, hdfs, https, ...
        return None

Example 3-18. Code 3-7

The only helper added was to extract the server from a file path, which must be extended over time to handle more cases. Also, id is generated based on the provided path, which is relative (starts with ./), and should be expanded to use the absolute path. Also, if we use a database table, for example, using a method other than read_csv, we’d need to add other helpers to handle connection strings.

Next, we can model the Schema that the application sees for a Data Source as follows:

class Schema:
    # primary key
    # fields is a list of (`name`, `type`) - each entry is a field / a column information
    fields: list[tuple[str, str]]
    # primary key => so fields are linked to this data source only in our model
    data_source:DataSource
    id: str
    def __init__(self, fields: str, data_source: DataSource) -> None:
        self.fields = fields
        self.data_source = data_source
        linearized_fields = ",".join(list(map(lambda x: x[0]+"-"+x[1], sorted(self.fields))))
        self.id = hashlib.md5(",".join([linearized_fields, self.data_source.id]).encode("utf-8")).hexdigest()
    def to_json(self):
        from functools import reduce
        jfields = reduce(lambda x,y:dict(**x,**y), map(lambda f: {f[0]: f[1]}, self.fields))
        return {"id": self.id, "fields": jfields, "data_source": self.data_source.id}
    # Add static method to extract the fields information from a pandas DataFrame
    @staticmethod
    def extract_fields_from_dataframe(df:pd.DataFrame, hints: dict):
        fs = list(zip(df.columns.values.tolist(), map(lambda x: str(x), df.dtypes.values.tolist())))
        return [(f[0], hints.get(f[0], f[1])) for f in fs]

Example 3-20. Code 3-8

The Schema class has the fields property, which models the columns of the CSV file. In this case, we can extract it from the metadata of the pandas DataFrame: columns and dtypes. We chose it to represent the fields as a list of pairs (field name, field type).

The last class for this part is DataMetrics, which will model the metrics of the files that the ingestion application reads and writes. However, it’s an incomplete version of the class because it encodes only the relationship with Schema. It must be extended to ensure it provides visibility into the data metrics for a specific usage. This purpose requires a lineage that comes later in this section. The current class might look like the following example:

class DataMetrics:
    # primary key => the metrics is attached to the schema that was just read, as the data source could change
    schema:Schema
    # metrics is a list of names and values - which in this case will be only numerical.
    # We'll first represent the numerical values returned by `describe`
    metrics: list[tuple[str, float]]

    # ⚠️⚠️⚠️⚠️ SOMETHING IS MISSING => see next section when we introduce the metrics context: Lineage ⚠️⚠️⚠️⚠️

    id: str
    def __init__(self, metrics: list[tuple[str, float]], schema: Schema) -> None:
        self.metrics = metrics
        self.schema = schema
        self.id = hashlib.md5(",".join([self.schema.id]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "metrics": jfields, "schema": self.schema.id}
    # Add static method to extract the metrics information from DataFrame's describe
    @staticmethod
    def extract_metrics_from_dataframe(df:pd.DataFrame):
        d = df.describe(include='all')
        import math
        import numbers
        metrics = {}
        for field in d.columns[1:]:
            msd = dict(filter(lambda x: isinstance(x[1], numbers.Number) and math.isnan(x[1]) is False,
                                map(lambda x: (field+"."+x[0], x[1]), d[field].to_dict().items())))
            metrics.update(msd)
        return metrics

Example 3-22. Code 3-9

The metrics were chosen in this simplified form to represent only the numerical descriptive statistics as a list of pairs (metric name, numerical value). These observations will convert directly from the result of the describe function called on the DataFrame that holds the data values.

Now that we have defined all entities as classes, we can update our code to ensure it generates the observations appropriately as shown in the following example:

apple_path = "../datasources/%s/%s/Apple.csv" % (year, month)
apple_hints = {"Date": "date", "Symbol": "category"}
Apple = pd.read_csv(apple_path, parse_dates=list(dict(filter(lambda x: x[1] == "date", apple_hints.items())).keys()),
                                dtype=dict(filter(lambda x: x[1] == "category", apple_hints.items())))
Apple_data_source = DataSource(apple_path, DataSource.extract_server(apple_path)))
logging.info(json.dumps(Apple_data_source))
Apple_schema = Schema(Schema.extract_fields_from_dataframe(Apple, apple_hints), Apple_data_source)
logging.info(json.dumps(Apple_schema))
Apple_data_metrics = DataMetrics(DataMetrics.extract_metrics_from_dataframe(Apple), Apple_schema)
logging.info(json.dumps(Apple_data_metrics))

buzzfeed_path = "../datasources/%s/%s/Buzzfeed.csv" % (year, month)
buzzfeed_hints = {"Date": "date", "Symbol": "category"}
Buzzfeed = pd.read_csv(buzzfeed_path, parse_dates=list(dict(filter(lambda x: x[1] == "date", buzzfeed_hints.items())).keys()), ,
                                      dtype=dict(filter(lambda x: x[1] == "category", buzzfeed_hints.items())))
Buzzfeed_data_source = DataSource(buzzfeed_path, DataSource.extract_server(buzzfeed_path)))
logging.info(json.dumps(Buzzfeed_data_source))
Buzzfeed_schema = Schema(Schema.extract_fields_from_dataframe(Buzzfeed, buzzfeed_hints), Buzzfeed_data_source)
logging.info(json.dumps(Buzzfeed_schema))
Buzzfeed_data_metrics = DataMetrics(DataMetrics.extract_metrics_from_dataframe(Buzzfeed), Buzzfeed_schema)
logging.info(json.dumps(Buzzfeed_data_metrics))

# [...]

monthly_assets_path = "../datasources/%s/%s/monthly_assets.csv" % (year, month)
monthly_assets.to_csv(
    monthly_assets_path, index=False
)
monthly_assets_data_source = DataSource(monthly_assets_path, DataSource.extract_server(monthly_assets_path)))
logging.info(json.dumps(monthly_assets_data_source))
monthly_assets_schema = Schema(Schema.extract_fields_from_dataframe(monthly_assets), monthly_assets_data_source)
logging.info(json.dumps(monthly_assets_schema))
monthly_assets_data_metrics = DataMetrics(DataMetrics.extract_metrics_from_dataframe(monthly_assets), monthly_assets_schema)
logging.info(json.dumps(monthly_assets_data_metrics))

Example 3-24. Code 3-10

Note

Code evolution: This version of the code requires tremendous changes and an extra layer of instructions. Reviewing this code to understand what it does helps give a sense of what we need to observe and how. <Cross-ref> shows a more simplified version that introduces a simple function that removes all duplicated codes.

Also, the next sections introduce many strategies you can use to reduce the number of changes and work to the point where data observation generation is almost automated and invisible.

We now have code that generates the observations we need to observe the data behavior when running the ingestion application. However, plenty of code is almost identical. This version reduces the noise with a function and uses the subtleties of the language to limit the changes by using the kwargs expansion:

def observations_for_df(params: dict, df: DataFrame) -> None
    hints = dict(**{d : "date" for d in params["kwargs"]["parse_dates"]}, **params["kwargs"]["dtype"])
    ds = DataSource(path, DataSource.extract_server(params["args"][0])))
    logging.info(json.dumps(ds))
    sc = Schema(Schema.extract_fields_from_dataframe(df, hints), ds)
    logging.info(json.dumps(sc))
    ms = DataMetrics(DataMetrics.extract_metrics_from_dataframe(df), sc)
    logging.info(json.dumps(ms))
apple_read_params = {
    "args": ["../datasources/%s/%s/Apple.csv" % (year, month)],
    "kwargs": {"parse_dates": ["Date"], "dtype": {"Symbol": "category"}}
}
Apple = pd.read_csv(*apple_read_params["args"], **apple_read_params["kwargs"])
observations_for_df(apple_read_params, Apple)
buzzfeed_read_params = {
    "args": ["../datasources/%s/%s/Buzzfeed.csv" % (year, month)],
    "kwargs": {"parse_dates": ["Date"], "dtype": {"Symbol": "category"}}
}
Buzzfeed = pd.read_csv(*buzzfeed_read_params["args"], **buzzfeed_read_params["kwargs"])
observations_for_df(buzzfeed_read_params, Buzzfeed)
# [...]
monthly_assets_write_params = {
    "args": ["../datasources/%s/%s/monthly_assets.csv" % (year, month)],
    "kwargs": {"index": False}
}
monthly_assets.to_csv(*monthly_assets_write_params["args"], **monthly_assets_write_params["kwargs"])
observations_for_df(monthly_assets_write_params, monthly_assets)

Example 3-26. Code 3-11

In this section, we work with files, but what if you have to deal with tables instead, using SQL to read, transform, and write the data? If SQL is used directly to perform these operations, it contains the same operations as a Python code would have. That is, it reads tables, transforms them, and eventually writes them, for example, an insert of a sub-select. However, SQL doesn’t provide many capabilities to generate information in-the-middle, such as metadata extraction or metrics. Another application, which the database server should be responsible for, runs the SQL. Learn more about this concept in Chapter 4.

Remember SQL holds a lot of information that your application can leverage. It has a list of tables, columns, types, and transformations, for example, that you can extract to generate the same observations as above. We recommend using any SQL because it’s the result of several experiments and iterative steps. It implicitly relies on all assumptions made about its components during these steps, providing crucial visibility about them.

Note

SQL encoding: SQL is procedural logic—like the Python logic used here. However, it’s encoded in a domain specific language (DSL) that’s processed by the server, which represents the compiler and executor.

After adding the ingestion capability to generate data and application observations, we can work on another area that will provide even more information—observations about the analytical part— when the application is deployed. For this part, we address the connection created between the data, down to the column level, which is the Lineage entity as shown in Figure 3-5.

Figure 3-5. [Insert Caption]

As discussed in Chapter 2, the part of the model that connects the lineage—the data flow and transformations in this case—glues the observations from the main component of the Data Observability area to the Application component. It’s in the Analytics Observability area, at the intersection of all three areas.

To encode this part, we must generate observations about which data sources produce other data sources. For the ingestion, the stock CSV files help generate the monthly_assets CSV file. Going one level deeper, ingestion concatenates all values from all columns and then writes the result in a file. The lineage at the field level is direct, meaning the input and output column names are identical and an output takes only its value from this column for each input data source.

Code 3-12 shows how to encode the reporting application to generate the lineage---OutputDataLineage---with a simple definition of the entity as a dedicated class to express the dependencies with input data sources:

class OutputDataLineage:
    # primary key - the schema of the data source output by this lineage
    schema: Schema
    # primary key - each schema of data sources used as input are mapped at field level as a dict (OutputFieldName, list[InputFieldName])
    input_schemas_mapping: list[tuple(Schema, dict)]
    def __init__(self, schema: str, input_schemas_mapping:list[tuple(Schema, dict)]) -> None:
        self.schema = schema
        self.input_schemas_mapping = input_schemas_mapping
        self.id = hashlib.md5(",".join([self.schema.id, linearize(input_schemas_mapping))]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "schema": self.schema, "input_schemas_mapping": self.input_schemas_mapping}
    # Generate a mapping using a simple heuristic that use field names only to create dependencies
    @staticmethod
    def generate_direct_mapping(output_schema: Schema, input_schemas: list[Schema]):
        input_schemas_mapping = []
        # get output field names only
        output_schema_field_names = [f[0] for f in output_schema.fields]
        # let's analyze each input schema and see which ones have a direct link with the output
        for schema in input_schemas:
            mapping = {}
            # loop over the input fields
            for field in input_schemas.fields:
                # if the input field is in the ouput fields
                if field[0] in output_schema_field_names:
                    # then we map them
                    mapping[field[0]] = [field[0]]
            # if at least one map has been added
            if len(mapping):
                # then we create a connection with the output schema
                input_schemas_mapping.append((schema, mapping))
        return input_schemas_mapping
    def linearize(input: list[tuple(Schema, dict)]):
        # left blank for the purpose of this example.
        pass

class DataLineageExecution:
    # primary key
    lineage: OutputDataLineage
    # primary key - links the current execution of the transformation with the execution of the application that runs it
    application_execution: ApplicationExecution
    # primary key - we make the choice to add the start time in the PK as we may have several runs of the same lineage
    start_time: datetime
    # NB: we could have a "end_time" as well which may not be in the PK perhaps.
    def __init__(self, lineage: OutputDataLineage, application_execution:ApplicationExecution) -> None:
        self.lineage = lineage
        self.application_execution = application_execution
        self.start_time = datetime.datetime.now()
        self.id = hashlib.md5(",".join([self.lineage.id, self.application_execution.id, self.start_time)]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "lineage": self.lineage, "application_execution": self.application_execution, "start_time": self.start_time}

Example 3-28. Code 3-12

Each input is mapped with the output at the field level, connecting them in this way: output_field -> list of input fields for each input. The generate_direct_mapping helper binds the data sources using an overly simple heuristic that maps output data sources when field names match. This strategy fails in most real-life use cases, especially with aggregations, which require more conscientious tracking to manage all connections. However, you can avoid this situation by using one of the strategies discussed in the next sections.

The lineage execution—ApplicationExecution— makes the connection with application-related observations.

Also, the model in Chapter 2 linked the Data Metrics entity to the execution of the lineage to give visibility about the data when it’s used or modified. Therefore, we adapt the DataMetrics model to represent that link by adding the lineage_execution property:

class DataMetrics:
    # primary key => the metrics is attached to the schema that was just read, as the data source could change
    schema:Schema
    # primary key => the lineage execution to ensure the metrics is "orphan"
    lineage_execution: DataLineageExecution

    metrics: list[tuple[str, float]]

    id: str
    def __init__(self, metrics: list[tuple[str, float]], schema: Schema, lineage_execution: DataLineageExecution) -> None:
        self.metrics = metrics
        self.schema = schema
        self.lineage_execution = lineage_execution
        self.id = hashlib.md5(",".join([self.schema.id, self.lineage_execution.id]).encode("utf-8")).hexdigest()
    def to_json(self):
        return {"id": self.id, "metrics": jfields, "schema": self.schema.id, "lineage_execution": self.lineage_execution.id}
    # Add static method to extract the metrics information from DataFrame's describe
    @staticmethod
    def extract_metrics_from_dataframe(df:pd.DataFrame):
        d = df.describe(include='all')
        import math
        import numbers
        metrics = {}
        for field in d.columns[1:]:
            msd = dict(filter(lambda x: isinstance(x[1], numbers.Number) and math.isnan(x[1]) is False, map(lambda x: (field+"."+x[0], x[1]), d[field].to_dict().items())))
            metrics.update(msd)
        return metrics

Example 3-30. Code 3-13

Now all pieces are ready to generate the observations shown in Figure 3-5. The following code shows a possible ingestion script:

# Observations high-level helpers -> move to a framework
def andLog(val):
    logging.info(json.dumps(val))
    return val
# generate observations about the application and return what is reused: the ApplicationExecution
def observations_for_application():
    application = andLog(Application(Application.fetch_file_name()))
    application_repo = andLog(ApplicationRepository(ApplicationRepository.fetch_git_location()), application)
    application_version = andLog(ApplicationVersion(ApplicationVersion.fetch_git_version(), User(ApplicationVersion.fetch_git_author()), application_repo))
    application_execution = andLog(ApplicationExecution(application_version, User(User.fetch.current_user()), Server(Server.fetch_ip())))
    return application_execution

# Now the helper generate observations for DataSource and Schema,
# but only prepares the generation of the DataMetrics for when the lineage execution will be available
# Hence why the function now returns the entities
def observations_for_df(params: dict, df: DataFrame):
    hints = dict(**{d : "date" for d in params["kwargs"]["parse_dates"]}, **params["kwargs"]["dtype"])
    ds = andLog(DataSource(path, DataSource.extract_server(params["args"][0]))))
    sc = andLog(Schema(Schema.extract_fields_from_dataframe(df, hints), ds))
    logging.info(json.dumps(sc))
    def generate_metrics(lineage_execution: DataLineageExecution):
        return andLog(DataMetrics(DataMetrics.extract_metrics_from_dataframe(df), sc, lineage_execution))
    return (ds, sc, generate_metrics)

# Script starts here

application_execution = observations_for_application()

apple_read_params = {
    "args": ["../datasources/%s/%s/Apple.csv" % (year, month)],
    "kwargs": {"parse_dates": ["Date"], "dtype": {"Symbol": "category"}}
}
Apple = pd.read_csv(*apple_read_params["args"], **apple_read_params["kwargs"])

buzzfeed_read_params = {
    "args": ["../datasources/%s/%s/Buzzfeed.csv" % (year, month)],
    "kwargs": {"parse_dates": ["Date"], "dtype": {"Symbol": "category"}}
}
Buzzfeed = pd.read_csv(*buzzfeed_read_params["args"], **buzzfeed_read_params["kwargs"])

# [...] other inputs

monthly_assets_write_params = {
    "args": ["../datasources/%s/%s/monthly_assets.csv" % (year, month)],
    "kwargs": {"index": False}
}
monthly_assets.to_csv(*monthly_assets_write_params["args"], **monthly_assets_write_params["kwargs"])

# finish with all observations generated and sent
observations = {"inputs": [], "output": None}
observations["inputs"].append(observations_for_df(apple_read_params, Apple))
observations["inputs"].append(observations_for_df(buzzfeed_read_params, Buzzfeed))
# [...] for other inputs
observations["output"] = observations_for_df(monthly_assets_write_params, monthly_assets)
# generate lineage based on inputs for the output
lineage = OutputDataLineage(observations["output"][1], OutputDataLineage.generate_direct_mapping(observations["output"][1], [i[1] for i in observations["inputs"]]))
lineage_execution = DataLineageExecution(lineage, application_execution)
# updating metrics
for o in observations["inputs"]: o[2](lineage_execution)
observations["output"][2](lineage_execution)

Example 3-32. Code 3-14: π

In Code 3-14, notice that the observations generated for the inputs shifted to the end. We made this implementation choice for the simplicity of the example. An upside is the additional computations are done at the end without impacting the business flow. A downside is that, if something fails in between, no observations about the data sources and their schema are sent. Of course, it is possible to avoid this situation,with some adaptations the code. Also, for ‌this introduction to a low-level API, we have to add some boilerplate to generate the right information that may sounds like noise, proportionally to the business code. Tough, keep in mind that the script had no logs initially. In general, logs are added sporadically to generate some information about the script’s own behavior, which is what we’ve done, but for the data.

Before moving on with the analyses of how the observations help with the issues mentioned at the beginning of this section, we reuse what we’ve done so far for the ingestion application to address the reporting application by using the following code:

import sys
month = sys.argv[1]
year = sys.argv[2]
application_execution = observations_for_application()
import pandas as pd
all_assets_read_params = {
    "args": ["../datasources/%s/%s/monthly_assets.csv"%(year,month)],
    "kwargs": {"parse_dates": ["Date"]}
}
all_assets = pd.read_csv(*all_assets_read_params["args"], **all_assets_read_params["kwargs"])
apptech = all_assets[all_assets['Symbol'] == 'APCX']
Buzzfeed = all_assets[all_assets['Symbol'] == 'ENFA']
Buzzfeed['Intraday_Delta'] = Buzzfeed['Adj Close'] - Buzzfeed['Open']
apptech['Intraday_Delta'] = apptech['Adj Close'] - apptech['Open']
kept_values = ['Open','Adj Close','Intraday_Delta']
Buzzfeed_write_params = {
    "args": ["../datasources/%s/%s/report_buzzfeed.csv"%(year,month)],
    "kwargs": {"index": False}
}
Buzzfeed[kept_values].to_csv(*Buzzfeed_write_params["args"], **Buzzfeed_write_params["kwargs"])
apptech_write_params = {
    "args": ["../datasources/%s/%s/report_AppTech.csv"%(year,month)],
    "kwargs": {"index": False}
}
apptech[kept_values].to_csv(*apptech_write_params["args"], **apptech_write_params["kwargs"])
# finish with all observations generated and sent
obs_all = observations_for_df(all_assets_read_params, all_assets)

obs_buzzfeed = observations_for_df(Buzzfeed_write_params, Buzzfeed)
lineage_buzzfeed = OutputDataLineage(obs_buzzfeed[1], OutputDataLineage.generate_direct_mapping(obs_buzzfeed[1], [obs_all[1]]))
lineage_buzzfeed_execution = DataLineageExecution(lineage_buzzfeed, application_execution)
obs_all[2](lineage_buzzfeed_execution)
obs_buzzfeed[2](lineage_buzzfeed_execution)

obs_apptech = observations_for_df(apptech_write_params, apptech)
lineage_apptech = OutputDataLineage(obs_apptech[1], OutputDataLineage.generate_direct_mapping(obs_apptech[1], [obs_all[1]]))
lineage_apptech_execution = DataLineageExecution(lineage_apptech, application_execution)
# ⚠️ this is the second time we generate a metrics for all_assets.
obs_all[2](lineage_apptech_execution)
obs_apptech[2](lineage_apptech_execution)

Example 3-34. Code 3-15

By adding the observations this way, we keep the modifications similar to what we did in the ingestion application. This approach enables us to build habits and abstractions, such as a framework, that will reduce increasingly more changes—almost an implicit law in development.

Also, keep in mind the following points:

We reused the OutputDataLineage.generate_direct_mapping helper as-is to create lineage mapping between the outputs and input. However, it won’t work because we aggregated Adj Close and Open from the monthly_assets.csv file into the new Intraday_Delta column. Because the fields don’t have the same name, the “direct” heuristic won’t pick up this dependency.

A warning message is displayed about reporting the same observations a second time for the monthly_assets. We did this because we encoded lineage per output. We now have two lineages, each producing output for the report_buzzfeed.csv and report_AppTech.csv files. Because, the output reuses the same data as the input (filtered), we must report how the input looks like for each output to avoid them appearing as duplicates. As an alternative, we could reuse the observations or adapt the model to address this duplication. You might consider the following options instead:

If we change our strategy to read every time the data is accessed instead of loading it in memory, then if the data changes between the two writing operations, the observations won’t be identical anymore. If one output has issues, we prefer to synchronize the observations of the input with this lineage. The probability of this situation increases when you consider each writing operation can take hours, not seconds.

The reporting application generates all outputs at every run, but refactoring later can change this and make it parametrable. For example, only one output might be created, such as BuzzFeed. Therefore, each reporting data is generated by independent runs. In this case, the observations already represent this appropriately, so we don’t need to adjust the logic. In other words, sending the observations of a given input as often as it is used to create an output represents the reality rather than trying to optimize for the sake of reducing data that might look like duplicates.

Let’s address the first point and ensure that the lineage can represent the real connections between the data sources. To do this in a simplified manner, we update the helper function as shown in Code 3-16 with extra information that provides non-direct mappings for each input. Later sections present strategies to handle this common use case in a much easier, efficient, maintainable, and accurate way, such as by using monkey patching.

# Generate a mapping using a simple heuristic that uses field names only to create dependencies
# Now each Schema is accompanied with hardcoded extra mapping in case direct is not enough
@staticmethod
def generate_direct_mapping(output_schema: Schema, input_schemas: list[tuple(Schema, dict)]):
    input_schemas_mapping = []
    # get output field names only
    output_schema_field_names = [f[0] for f in output_schema.fields]
    # let's analyze each input schema and see which ones have a direct link with the output
    for (schema, extra_mapping) in input_schemas:
        mapping = {}
        # loop over the input fields
        for field in input_schemas.fields:
            # if the input field is in the ouput fields
            if field[0] in output_schema_field_names:
                # then we map them
                mapping[field[0]] = [field[0]]
        # updating the mapping with the provided extra mapping if any
        mapping.update(extra_mapping)
        # if at least one map has been added
        if len(mapping):
            # then we create a connection with the output schema
            input_schemas_mapping.append((schema, mapping))
    return input_schemas_mapping

Example 3-36. Code 3-16

The following code shows the final observations part of the reporting application:

# finish with all observations generated and sent
obs_all = observations_for_df(all_assets_read_params, all_assets)
# manually mapping the "Intraday_Delta" column
intraday_delta_mapping = {"Intraday_Delta": ['Adj Close'], 'Open']}
obs_buzzfeed = observations_for_df(Buzzfeed_write_params, Buzzfeed)
lineage_buzzfeed = OutputDataLineage(obs_buzzfeed[1], OutputDataLineage.generate_direct_mapping(obs_buzzfeed[1], [(obs_all[1], intraday_delta_mapping)]))
lineage_buzzfeed_execution = DataLineageExecution(lineage_buzzfeed, application_execution)
obs_all[2](lineage_buzzfeed_execution)
obs_buzzfeed[2](lineage_buzzfeed_execution)
obs_apptech = observations_for_df(apptech_write_params, apptech)
lineage_apptech = OutputDataLineage(obs_apptech[1], OutputDataLineage.generate_direct_mapping(obs_apptech[1], [(obs_all[1], intraday_delta_mapping)]))
lineage_apptech_execution = DataLineageExecution(lineage_apptech, application_execution)
# ⚠️ this is the second time we generate a metrics for all_assets.
obs_all[2](lineage_apptech_execution)
obs_apptech[2](lineage_apptech_execution)

Example 3-38. Code 3-17

Now, we can deploy, run, and monitor our pipeline by using the observations they generate each time they’re run. The low-level API required quite a bit of extra effort and conviction to achieve. Nevertheless, we are satisfied with the results. Every minute spent on those tasks will give us 100 times avoidance of revenue loss—following the total quality cost rule 1-10-100— when the issues happen in production.

Again, let’s look at the issues mentioned in this section, starting with the ingestion application failures:

Input files not found.

The DataSource observations are sent every run. After they’re read, the failing run are therefore not sending any of them. Even for someone without any knowledge about the application logic, it’s clear the files used until now are missing.

Type errors at read.

The Schema observations are sent and contain the field names associated with their type. Therefore, the expected type is clear for the observer without going to the application or the files for the previous months.

Errors due to missing fields.

The same observations as for “Type errors at read” help the observer quickly identify which fields were present in the previous runs that are missing in the current one.

File system errors.

The exception thrown by the pandas library normally provides the path that resulted in an error. The remaining information for the observer to identify the issue is which server this path applied. The IP provided in the Server observations associated with the DataSource gives instant visibility about which server is related to that path.

Memory error.

This issue happens mostly when the data increases suddenly. Many cases can be considered, but they are mostly handled intuitively by the observer using DataMetrics observations that contain the number of rows, the schema with the number of fields, or the number of DataSources. Regardless, it might require the observations to be sent earlier than at the end of the script:

One of the input files has a larger size than before.

The file is detected easily because no DataMetrics are available for it.

The output has grown a lot because all files have grown.

The size difference is detected because the DataMetrics for the outputs are missing. Also, the DataMetrics for the inputs show an increase in the number of rows.

File system space errors.

These errors most likely happen when writing the output, considering we’re addressing data observability cases here. Therefore, the same information as for “Memory error” gives the observer instant visibility about why the space is no longer sufficient and which files couldn’t be written.

The “Date” is not parsable as a date.

In this case, the Schema has the type observation for the date field change from “date” to something else, such as “str” or “object.”

“Date” column doesn’t contains values for the current year/month.

The DataMetrics observations include the minimum and maximum time stamp, which gives instant visibility into the difference between the execution time and the available data.

“Date” column contains values in the future.

Easy, the same observations as for the previous failure due to missing values for the current month will give this visibility.

“Symbol” categories changed.

If we only consider numeric DataMetrics, we can quickly identify this case by using the number of categories that would grow in the output file. One or some of the files wouldn’t be consistent anymore as they would refer to different categories..

Then, we must consider how the reporting application might regard the ingestion application as failed and what the reporting application observer could use:

The monthly aggregated file is not available.

The DataSource observation or DataMetrics wouldn’t have been sent by the ingestion application.

Aggregation uses missing fields, such as “Close”.

The Schema of the monthly data sent by the ingestion application would miss those fields too.

Errors with read/write access, size, and space.

The same solutions apply to the reporting observer as for the ingestion observer. There is no bias of information across teams or team members.

‘APCX’ and ‘ENFA’ symbols.

The number of categories reported by the ingestion observer has changed, giving hints in some cases. However, we can extend DataMetrics to also report non-numerical observations and report the categories.

Missing values in ‘Adj Close’ or ‘Open’ leading to abnormal numbers.

The DataMetrics “number of nulls” typically provides these in a straightforward fashion as if any of those number of nulls are greater than zero, the computation of the Intraday_Delta will return NAs.

Wrong date in the monthly assets.

The same solutions used by the ingestion observer related to Date failures apply here. For example, it uses the minimum and maximum value for the “Date” column compared to the currently reported month.

We’re now ready to handle various situations when the data is the source of a problem. Without this knowledge, these situations would require long, high-stress meetings to understand them and exhausting hours or days spent debugging that could turn into months because we can’t access the data in production.

The issues discussed so far are ones we know can happen. However, many other issues are likely to propagate throughout the pipeline for which we know little about yet—the unknown-unknowns. For example, the values of one of the stocks are for a previous month due to errors in the data during the CSV export. This problem will happen for at least one of your applications and others like it.

For this kind of unknown-unknown, data observations mustn’t be restricted toonly specific cases. The applications must report as much observations as possible—with possible constraints of computation resources and time—to generate visibility about the anticipated or unmet situation. In this example, distribution of the monthly stock values would be useful later for comparison with other months and could provide hints about whether the values are equal or similar.

The advantage of using low-level logging is to have full flexibility over what you can generate as visibility. Examples are custom metrics and key performance indicators (KPIs).

Because all jobs aren’t equal, every company, project, and application has its own specific details. You’re likely to control specific metrics whether they are linked to the consumed data or produced data. Such metrics for a table could be the sum of the number of items multiplied by the cost per unit minus the amount received from a web service. The result must always be greater than zero. This can be easily encoded but has to come from the engineer in any language they use.

Another reason for customizing observations is KPIs—the numbers stakeholders request that are important to the underlying business. KPIs are often reported periodically or computed on demand and used at random or fixed intervals. However, their meaning is so strong that stakeholders have high expectations for them, with little to no time to wait for corrections. Therefore, you must also create visibility about how KPIs evolve over time because, if a stakeholder has doubts about it, time starts ticking the moment they ask you about its correctness. To increase responsiveness, you must know generally how the KPIs evolve, identify how they change before they do, and understand why they change based on their historical values and lineage.

As you might have anticipated during the reporting application update, defining the API— the model, encoding, and functions—isn’t a task for each application. Rather, you must standardize and reuse it across applications. Standardization reduces the amount of work per application. More importantly, it enables the observations to be homogeneous, independently of the application, to simplify the observers’ work in matching the behaviors of other applications that participate in the pipeline they are analyzing.

Standardization is also helpful for reusing entities across applications, such as the assets_monthly DataSource, which is the output of the ingestion application and the input of the reporting application. With the homogenous representation of the observations, you can consolidate the status of the entire pipeline by reusing the entities across applications.

Part of the architecture to support data observability must include creating an external system that aggregates the observations to build a global view in a systematic way. By having observers that rely on this aggregated view and act upon it, the system can evolve to replace some of the actions performed by the observers, which is where machine learning comes into play.

Note

Aggregator service as OpenAPI: An example of an aggregator service that’s implemented as an OpenAPI is available in the following GitHub repository:TODO. The API used behind the Python library was given as a reference for the low-level API earlier in this section. Also, we’ll reuse this service in Chapter 3, which includes recipes for common data engineering tools.

Although low-level logging is hyper-customizable and deals with great flexibility, its adoption can be impacted by the amount of effort it has, at least initially. We made the same argument for tests. Nevertheless, we can simplify the complexity of usage at this level. And we should look for alternatives to complement it, while ensuring the adoption of data observability across teams and their members. The following sections address this topic, starting with an event-based system.

Abstraction Strategies

The previous section explained how to add data observability to applications that deal with data by using a low-level API. This approach can become tedious, even for simple processes, leaving the task to the final user—the data engineer or practitioner

Most of the work, such as the instructions, can be abstracted into a higher-level framework that reduces the amount of effort, as long as you dedicate enough time to it. Like all best practices, we can also omit for the sake of reducing time to market, even though we’re just going to deploy something in which we have little or no confidence.

To avoid falling into this trap again, we can automate various best practice strategies or techniques–this is an already adopted practice in other areas such as Application Observability, where most commong observations are generated by default such as the number of requests for a web service, the memory usage of a computation engine, the version of the application (cf. META-INF). Automation always comes with less flexibility because it addresses common tasks. However, you can combine these tasks by using the low-level API to extend with your specific needs.

These techniques are used heavily in application observability, for example, where they apply naturally. Players, like DataDog, Splunk, and Opentracing, use them in the various agents they develop.

The first strategy—event listeners—leverages events the tools you use for your data work might trigger.

Event listeners

Depending on the framework or tools you use to transform or manipulate data, some have likely implemented several concepts that allow the internal behavior to be reacted upon. When some events happen, the technology enables you to intercept and operate on them. The ability to understand which behavior can be intercepted would give you the opportunity to create visibility about them in a more automated, systematic, and standard way.

In the context of data observability, the events related to the model that represents the internal system explained in Chapter 2 are of particular interest. Such events are generally available for reading or writing a data source and running a transformation.

The components to leverage these events are commonly named listener or handler. You can create and register them on the event bus, which is the system that propagates the triggered events:

Some technologies can trigger events that contain all information at once, or enough information, so you can generate the model almost fully in a single line of log or action, such as with Apache Spark (REF to Chapter 4).

When possible, the event-based agents are powerful as you can integrate them automatically at the framework level or in the tools’ servers, publishing most of the information without involving much time.

Of course, this concept works only for creating standardized information. Any customization, such as metrics, algorithms, or descriptions, must be introduced by application or usage. The other types of observation that require involvement are the rules related to expectations. They leave little room for standardization while keeping their efficiency and performance. Generic rules, which apply to all cases, are too vague to capture custom or corner cases. Applications are different.

Logs are events

This special case of events is interesting to consider as a fallback when the framework, application, or system you use doesn’t provide the flexibility you need to create your goal for data observability. Those systems are called black-boxes and include some legacy systems.

However, if those systems are closed, they are likely generating some logs, such as SQL executed or modules applied. The logs create some visibility about the system as part of application observability. However, they might eventually contain enough information to recreate part of the data observability out-of-context by patching the system externally.

Aspect-oriented programming

The aspect-oriented programming strategy is somewhat similar to the event listener strategy as the idea is to react to some behavior to extend it, for example. It’s often used for configuration purposes because it enables components to be integrated into a more declarative fashion. Aspects are like triggering events when functions are going to be executed or have been executed.

The capacity to react before and after the functions are executed is what you can use to intercept behaviors and expose them as part of data observability. Examples include running a component to read a file or running an anonymization procedure.

Java application observability agents, such as open tracing for CDI/Spring applications, rely heavily on this approach. Open tracing, for example, automatically captures access to an HTTP endpoint or database. The idea is to reuse, in this case, as much as possible what already exists and is in place (see Chapter 4).

An aspect is defined by using the following components:

Aspect: A dedicated behavior that spans several running classes, such as a DataObservabilityAspect defines the behaviors to introduce data observability.

JointPoint: The moment of a point-cut method or property, such as before access or after execution.

Advice: The behavior to run at a JointPoint.

PointCut: The point to enable and run an Advice. It’s generally defined as a regular expression that matches the class, method, and property names to run the associated Advice at the defined JointPoint.

Typically, aspects are implemented using in two different types of weaving to add capabilities into existing code, which we expand on in the next sections with other options:

At compile time: The aspect definitions help inject new code during the compilation phase. It’s done once.

At runtime: The aspects update code while it’s running. It’s done at each execution.

Code enrichment

Other alternatives enable data observability in a more automated way from the code that you are developing. The code enrichment strategy involves manipulating code to add behaviors without relying on existing capabilities that were introduced beforehand as introduced in previous sections.

You can achieve code enrichment by using runtime manipulation or using code generation.

Runtime manipulation

Enriching the code at runtime enables the application to be extended with additional capabilities, such as data observability when the application is run. This strategy is useful and efficient when adding generic capabilities that require few interventions or customizations by your engineers. Nevertheless, the team in charge of deploying and monitoring the application can configure such customizations. However, the customizations aren’t close to the application’s business function but rather to its generic behavior.

You can update an application at runtime in several ways. One of the most popular dynamic languages is Python, which uses monkey patching.

Monkey patch

When your application uses dynamically typed languages, such as R or Python, instead of statically typed languages, such as Java or C++, the variables have types or structures that might be checked when the code is run.

You can apply monkey patches in the window where the types can be extended between the application start and the usage of the different types. The term monkey patch comes from guerilla patch, which refers to sneakily changing code.¹

Monkey patching doesn’t require setting up used libraries, such as installing a new version, or updating the application code much to generate the data observations because those updates can be applied at runtime.

Currently, most data-related libraries haven’t yet been conceived to generate data observations and, therefore, offer little visibility into what they do. They’ll likely be updated over time to add this capability. But, in the meantime, we must find workarounds, which are called polyfills² in web development.

The following code shows how to monkey patch a function in Python:

// code without
// code of the monkey patch and the original call

We need to generate data observations for the three activities. The first one is the reading activity. To illustrate what we needed, we have a well-known pandas library in Python that has many extensions for various data sources.

Imagine you have to read a CSV file that you want to make observable with monkey patching. The following code shows what you need to do:

// code reading without
// code that generates DO

For the write activity, the code is pretty similar to the read activity. However, other information, such as metadata or metrics before or after the write, might be worth generating.

// code writing without
// code that generates DO

For the transformation activity, some technologies don’t prepare the transformation before applying it (called a lazy evaluation like in Spark). Instead, they use a combo of for-loops and manual manipulations. In these cases, it can be difficult to keep track of all transformations automatically from read to write. But, it’s not impossible. And if it’s challenging in an automated fashion, it’s at least as challenging manually, although working on the automation must be done only once and by a few people.

Continuing with the pandas, if your application filters the data read and combines it with others, you can track these operations to make the whole transformation observable. The following code shows how:

// code transforming without => reuse what we used in Low Level API
// code that generates DO

However, we have another case we haven’t handled here, which is when the transformations are multi-threaded. In this case, we must keep a correlation between the different operations happening asynchronously. We do this by tracing each step with a correlation_id that corresponds to the transformation that’s being run. This tracing enables easier reconciliation later. However, a specification already exists that can simplify this process called opentracing. This case is rarer in dynamic languages used for data purposes, so we don’t cover it here, though we’ll see some examples in practices in the next chapter (REFs).

Bytecode instrumentation

Dynamic languages offer a lot of flexibility, but that’s not the case with other languages, such as Java. Types are checked at compile time, which generates the bytecode—an intermediate representation of the code—that the virtual machine runs.

In this case, the compiled code isn’t something we can easily manipulate, because its representation isn’t identical to the original code. Because it’s so convenient to allow the code to be updated with polyfills, you can use another technique that’s a bit trickier than monkey patching.

This technique is called bytecode instrumentation or compiled code manipulation. In Java, its low-level API is called ASM—a reference to Assembler—and has been used for decades to build tools such as test coverage analyzers or mock generators. With ASM, you can take any code running in Java and inject it with your own additional behavior. That is, you can extend Spring Data, Flink, Spark, and alike at the bytecode level to generate data observations, without changing the original framework. Although ASM is low level, you can instead use a more convenient abstraction called javassist.

While the low-level bytecode injection is the most flexible, we don’t go any deeper because most cases are possible with aspect-oriented programming. In reality, aspect-oriented programming is an elaborate version of bytecode, such as AspectJ. However, it’s at a higher level of abstraction to remove most of the complexities to inject code at the price of adding a new dependency to your application—the aspectj.jar file.

Therefore, you can apply something similar to monkey patching to inject the generation of data observations, and it will require fewer (guest) code changes because it happens in the background. In fact, Python also has a bytecode you can inject.³ You can also implement what was possible by using monkey patching at a lower level by injecting the compiled code without wrappers.

For example, if you create a class to run SQLs, you can extend the methods that read, parse, and access columns and alike to trace the schema and the usage of the columns.

Code generation

As introduced at the beginning of this section, you can generate data observations by using code generation. However, this strategy doesn’t apply to enriching existing code from existing applications or libraries. Nevertheless, flexible alternatives are available to generate the code to run by considering new cases you encounter or simply the code itself.

For example, say your application uses SQL to query and update database tables, with the capability to generate code. A piece of code can be generated to parse the SQL in-place, which in turn generates code to run observations for each table that’s touched, such as generating other SQLs to gather metrics or metadata.

The following sections cover three options to generate code, starting with the usage of abstract syntax tree (AST).

AST manipulation

Regardless of the language you use to process data, it goes through a series of steps that turn your code into forms until a processor can run it. The first two phases are lexical analysis followed by a syntactic analysis. The latter phase generates an abstract syntax tree, which represents the structure of the code and encodes the execution of your logic.

An AST is a structure that gives you the option to manipulate it as long as you respect its rules before passing it to the code generator, which turns the result into the machine code.

The most common compilers and interpreters, such as Java, Python, and C#, offer this capability of intercepting and modifying the AST. You can leverage them in your application or framework to generate observations based on the code you’ll run.

To do so, you can visit an AST that contains read, write, and transform operations, as well as generates code blocks to create observations about those operations.

This code isn’t limited to advanced programming languages, such as Java or C. You can also introduce it for any other high-level abstraction, such as SQL and its variations or custom Domain Specific Languages (DSLs). To do this, use the ANother Tool for Language Recognition (ANTLR) technology, which handles the language complexities and gives you the ability to generate an AST or create your own representation.

Annotations, decorators, and attributes

Previously, we explained the possibilities of generating observations by using explicit logging, which requires your code to include all instructions to this effect.

A good practice of programming is to decompose your code into smaller blocks without too many responsibilities or functions. You can label each function with its responsibility, such as merging data sets or updating a field with a computed value. The label is similar to a document that describes how the function will behave.

Because the label has a well-known semantic, you can use it to update the code of the function with additional code that will generate observations about its behavior.

To leverage such capabilities, you can use a programming concept called annotation in Java, attribute in C#, or decorator in Python. These annotations are generally called syntactic sugar because they are removed from the generated machine code (or bytecode) after the compiler or interpreter processes them by using plug-ins. For example, in Java, annotation can be a retention at RUNTIME, which enables the runtime to access them using reflection.

The generated code handles most of the complexities based on the inputs and outputs of the annotated function. For example, it can infer the connections, create metrics about them, and extract metadata. In fact, the complexity of generating the observations is encapsulated in the annotations, making the code easier to reuse and maintain.

Macros

The last strategy to avoid writing most of the boilerplate over again and to increase adoption. Also, the quantity of observations is based on macro programming⁴ that enables code to be preprocessed and injects code in-place at compile time.

Macro programming isn’t always available for all languages. If all languages—even SQL—have functions, macros might not be supported. The key difference between a macro and a function is that a macro generates code that depends on the code it’s applied on. In a way, it’s meta-programming, because the idea is to write code that takes code as arguments and generates it as output during pre-compilation phases.

You might notice it’s relatively close to the AST strategy. The idea is to use a code representation, manipulate it, and then generate the enhanced code with the extra capabilities, which, in our case, is adding observation generation. AST happens outside the code itself, but macros are supported natively by the language and offer a simpler alternative, such as quasi quotations in R.⁵ The macros can process the lexical representation (tokens) or the abstract tree of the code, making it more capable.

Although it’s also safer than other alternatives, such as bytecode or AST manipulation, you must first ensure the macros are hygienic to avoid capturing existing variable (term) names.

The advantage of a macro is that it runs faster because it’s injected in the guest code and is supported by the language. Unlike annotations, which are added as function labels, macros are added within the body of the code.

High-level Applications

Over the years, data management activities, such as data integration and orchestration, have been abstracted into higher-level solutions to lower the entry barrier and increase efficiency.

Such solutions hide most of the complexities to develop common tasks like connecting to a database or sequencing several tasks. Therefore, the level of control you have on such applications is much lower than when working on your own code base.

The main operations remain the same conceptually, meaning a data application has to read (extract, input), transform, and write (load, output). Abstracting them generally takes the form of dedicated components that share the same structure for each operation and are implemented to hide the specificities of each type of input (such as a file on S3), transformation (such as SQL), and writing (such as in a Kafka topic). The resulting application must be orchestrated to run and collect the results.

Today, such solutions don’t yet fully embrace the principles of data observability that are introduced so far in this book. However, they can generate some visibility about how they operate. This capability gives us the opportunity to reconcile what’s available and create global visibility across the different systems, but still the task remains to complement what’s not yet available. Such high-level applications aren’t responsible for the entire end-to-end process but intervene as a step.

Let’s look at the types of high-level applications for data engineering, which are categorized into low-code and no-code.

No-code applications

No-code solutions for data provide a user interface that enables users to manipulate data without having to use any programming. They’re built to offer this capability to people who have few or no programming skills.

At the time of writing, these applications aren’t yet fully data observable. Also, because they offer a lot of flexibility, several data access, transforms, and creations activities happen fast without all the functionality that’s associated with producing the data observations and easing the description and validation of expectations.

Most of these systems operate as black boxes, meaning the code isn’t accessible or generated as it is for the low-code ones described later. Therefore, we’re limited in how we can fulfill our requirements. You can use the following workarounds until they embed the data observability features:

Logs to reconstruct observations.

Like any applications, no-code applications have logs that include some information that you can use to only reconstruct—even partially—some of the necessary observations. Of course, this means the application part, but also the lineage might be present. However, metrics often aren’t present.

Application or configuration files and APIs.

Often such applications store the data pipeline configuration in a file or make them available by using an API. This approach enables the opportunity to analyze the flow and generate the lineage and data sources from it at a minimum. With this information, you can build an additional component to gather the data metrics. However, the ability to catch when the pipeline is run is necessary to catch the right context of usage.

Execution or logs control board.

Such applications also generally present an interface, or logs, to review what is being executed. They enable you to intercept the executions and combine them with the options presented previously. Plus, you can develop an additional component to access the data sources at that moment to gather their metrics.

External data observability platform.

Expectations are harder to introduce at this level. In general, an external data observability platform is preferred to leverage the observations gathered in previous steps. However, an alternative is to extend the pipeline defined in the tool with additional checks that are added directly at the beginning and end, for example. This works if the tool has this feature.

Of course, no-code solutions don’t give much room to introduce data observability. But what was true for the frameworks you use in your code is also true here: we must provide data observability as a supported feature. In the meantime, we are simply trying to go faster and complete what hasn’t yet been implemented.

Low-code applications

The low-code alternatives require more technical skills to no-code systems and aim to help developers to work faster on common tasks. This section addresses the data pipeline creation task using code generation and workflows to define the processing steps.

Code generators

At a certain level in an environment, many tasks are often included in data pipelines, such as instances of the classic extract, transform, load (ETL) components: the readers, transformers, and writers.

Low-code platforms offer engineers the ability to generate the pipeline code but keep them focused on the business logic rather than, say, boilerplate code. Although a software framework could also handle this, such platforms as Talend have gained a lot of traction thanks to the user-interface (UI)-based environment.

The code generated by using Talend is Java and can be handled like any other code base that can be revisited or extended, for example. However, these platforms allow such extension beforehand with a system that gives the engineer a way to define their own behavior and keep their focus on the business logic.

Because the code is generated, we can use most of the strategies introduced in the first part of this section to give the platform the ability to generate observations across the whole process. Over time, those building blocks are likely to natively include the generated data observations generation. Therefore, when we define this type of extension for an open source platform, we can simply propose it as part of the original component instead of as a new one.

Workflows

Workflows might sometimes generate code, but it’s not guaranteed you’ll have access to it or room to extend it with data observability. However, describing a workflow in low code involves several steps when using templates that you can add in a JSON, YAML, or similar format to explicitly configure the read, transformation, and write functions.

We can customize these templates as long as they follow the workflow tool constraints. Therefore, we can add our own behavior by extending the existing templates that are available with data observability and then register them as new ones. For example, read_csv can be read_csv_with_do.

For templates based on code, such as defining a Spark step, we can extend the underlying Spark component with the agent to create the observations. We can also extend the template to add the extra capabilities to compute specific metrics or introduce expectations.

Orchestrated SQL templates enable users to write SQLs with additional configurations, such as database server or credentials. You can adjust the template to intercept the SQL and use its configuration to build, for example, the lineage, data sources (tables), and compute metrics on the fly.

For APIs that handle user queries, such as expressed in SQL, an alternative way to introduce data observability automatically is to introduce a proxy API, which mimics but extends the original API. The proxy passes the query as-is to the original API and returns the result with additional data observations, even though the generated observations might already be published.

The good news is a strategy is always available that enables us to retrieve some part of visibility and flexibility to avoid falling into the trap of building data applications without control over the data.

Differences with monitoring alternatives

Practices can be assimilated to data observability because they share the goal of monitoring the data quality. However, they do not address the same process, because data observability aims to detect data issues and provide the necessary information to find the right solution.

We could reconstruct the context information by mixing data quality monitoring information with observations, such as infrastructure and applications, provided independently by the other systems. For legacy systems, this might be an alternative, because it wouldn’t be considered viable to update the applications with data observability. However, it only consists of a best-effort strategy that relies on the reconstruct system to have enough information at all times, and to accurately reconstruct the internal data usage.

Choosing this strategy to implement a long-term solution to data reliability is an anti-pattern of data observability. The practices introduced here don’t involve all the different actors, such as data engineers who are often outside of the process or who have potentially different goals.

This section doesn’t detail the data quality (monitoring) practices, because various materials are available that describe them in a proper way, including DAM-DMBOK2 as explained in <>. Those data quality practices rely on experts, such as business analysts or subject matter experts (SMEs), to define quality constraints (or rules in this domain) that consist of metrics (or KPIs) and controls. The constraints are then scheduled against the data sources to validate that the content matches them. They are expectations, although they’re defined generally or independently of their usage. If that’s not the case, a large number of constraints satisfy the different needs. They come with management downsides, because over time, maintenance increases the program’s total-cost of ownership.

Setting up this battery of constraints can be relatively easy with the help of some open-source systems whether for the relational database or files, such as Apache Griffin or AWS Deequ. However, the additional resources needed to install and manage this external system might be important. You must consider them before moving on this path. For example, AWS Deequ requires an Apache Spark cluster and a scheduler or orchestrator, which increases the cost of ownership linearly.

Note

Avoid mismatched discoveries: Monitoring only after you create or update the data (a.k.a. at rest), will likely lead to mismatched discoveries which are desynchronized with the real event, that is the real moment in time the system deviated.

An evolution of these practices, recently coined as augmented data quality, softens the dependency on people’s time or knowledge by introducing artificial intelligence (AI) in the mix. These practices still require resources and access to the data and its history. However, instead of sending predetermined metrics computations and controls, they analyze the data to:

Discover rules that can be validated, meaning they’re approved by the users of the same type as before.

Pin data considered as anomalies, first statistically, and, eventually over time using machine learning based labeled anomalies.

With this information about the quality of the data, we’re now aware of general data issues. To provide the necessary context for engineers to solve the issues, we need to rebuild this context using the information at hand to reverse engineer it. Luckily, we can use the observations generated by the following components, which follow the approaches covered in the first part of this chapter:

Operating systems produce many observations through syslog (Linux). Many agents, such as Prometheus agents, are also developed to complement those native capabilities.

File systems log many events through dedicated APIs that can be consumed. Such APIs include inotify in Linux, kqueue in FreeBSD, and ReadDirectoryChangesW in Microsoft Windows.

Most database servers provide additional facets of data usage to extract observations partially. Two types of information can be considered:

Audit logs: Extended Events&Triggers in MSSQL, and log statements and triggers in, for example, PostgreSQL and BigQuery audit logs.

Change data capture (CDC): Uses monitoring events sent by databases, monitoring APIs, or supplementing tactics, such as triggers, to generate a stream of all changes in the server, from table creation or schema change to new entries or value updates.

Application log generation best practice: Logs now generate large amounts of data—mostly untapped data. As recent as 2015 or so, metrics and traces are also now regularly generated by using, for example, an open-source solution such as open telemetry or Prometheus.

The next task is to consolidate all these sources of observations from different, unsynchronized periods to reconstruct the contexts as close as possible to reality. You must also consolidate the information from which observations must still be extracted.

Note

Log standardization: Application logs are poorly standardized across applications and developers. Because there’s no guarantee they’ll be produced or enabled, such as debug logs, this task is harder than we can imagine.

In addition, the observations from these components don’t cover the ones needed to observe the data behavior. Otherwise, the approach explained in the first sections of this chapter is applied. You might notice the following observations, for example:

The database or table used by the application is available in the traces of that application. The number of rows and columns might be available, but not the data since the default intent is to control the application behavior.

The SQL captured in the audit logs won’t give much information about the metrics of the used or created tables, or only limited metrics such the number of output rows or execution time, although rarely more. However, SQL systems should eventually change this way to provide more observability capabilities out of the box.

Restraining ourselves to the sole databases, we can also leverage CDC. An open-source implementation of this strategy, called Debezium (https://github.com/debezium/debezium), comes with the architecture shown in Figure 3-6.

Figure 3-6. https://debezium.io/documentation/reference/1.9/architecture.html

As Figure 3-7 shows, CDC generates as many new streams of information that can be turned into observations of the database server and its components, such as tables. It also requires some resources capable of handling potentially big and fast data points. Even though the captured information can be elaborate and helpful, such as SQL queries to extract lineage, they are decorrelated to the usage. Therefore, the context must still be reconstructed to find and address changes that lead to issues.

All in all, these strategies aren’t suitable to embrace data observability and its benefits, such as data issues prevention and ease the analysis of issues. This strategy can’t work by only consuming what’s produced by the system as-is. Therefore, we need to find other alternatives to fill in the gaps, of which the most important or most lacking one is metrics. We might be tempted to reuse an old data quality or data catalog pattern, called the Kraken anti-pattern (see Figure 3-7), which consists of crawling, defining service-level indicators (SLI) or KPI, and then scanning.

Figure 3-7. The Kraken anti-pattern used in an architecture implementing data observability for a data pipeline

Implementing Expectations

Like Chapter 2, we end this chapter with expectations. As data practitioners, expectations are key to the success of our data initiatives, especially their operational maintenance and keeping our stakeholders’ trust. Introducing them in our applications is an important step of their development and maintenance, just as important tests and documentation are.

This section highlights moments of the application lifecycle where we must introduce expectations and the associated practices to discover and implement them. Expectations also represent pieces of knowledge or interpretations of the constraints based on understanding the context. They have their maintenance process just as tests and documentation do.

To conclude the chapter, we present two examples of overarching practices that combine the benefits of generating observations at the source and introducing expectations. But first, let’s see how to define expectations and add them to our data applications.

Introducing Expectations

As mentioned in Chapter 2, expectations represent data constraints or usage that we must respected, If we don’t follow them, we must be aware and take action. This section explains how to turn the expectations into rules—how to define and implement them—and the actions we need to take.

Before jumping in, let’s see what a rule looks like. You can implement a rule as a function that takes observations as inputs and computes a Boolean condition in its simplest form. So, when you develop an application, you then add instructions that use the observations and run the rule. The following code shows, in Python, how two rules validate the expectations that the values of email in the table are non-empty and well-formed:

TODO python code

If you implement a pipeline where you’re restricted in your code capacities, such as by a low-code platform, you have to implement the rules as an extension of the practices introduced in the related sections above, likely in a declarative fashion.

The declarative approach is a good practice over implementing rule logic and running it in the applications themselves. It enables you to give an external system this responsibility, which helps with standardizing, sharing, and notifying, for example. Also, this system can extend usage of the observations to more use cases if it keeps the historical observations. For example, it can detect patterns in the observations and suggest new rules or update them (see the maintenance section).

You must also consider those expectations and have the opportunity to add them. This best practice, like generating observations, is one you have to learn and master over years. To help you, let’s take a look at the opportunities that come naturally or that you can establish as part of the development process.

Shift-left data quality

As your organization has grown, it’s likely to have initiated a data quality process, or program, you can use to nurture the data observability process. In most cases, data governance initiatives drive data quality programs. These initiatives are often led by chief data officers who are closer to the business, have knowledge about how the data is used to make decisions, and therefore have important expectations based on their business knowledge and experience.

Over the years, data quality programs have leveraged this knowledge to introduce in-house or existing data quality tools, data validation points, or data control points. These controls access the data to compute metrics and apply checks. The computations or checks might fail, or in some cases the data is reputed wrong.

But in most of the cases, these programs are detached from the data engineers. They’re meant to be used by less technical people, which creates a distance with the knowledge and lowers the opportunity for data engineers to use it. Data governance programs have tried to close this gap by introducing a dedicated role—the data steward. This role has many responsibilities, such as maintaining data quality and streamlining communication between the different actors who use it.

Because these practices happen farther downstream, it’s often too late when failures are discovered, as mentioned in Chapter 1. You can break this pattern and communicate with your stakeholders or use the tools mentioned previously to integrate metrics and controls earlier in your own work. This notion is referred to as shifting-left.

You can shift left by introducing a step in the data project to capture existing data expectations from downstream users. Or you can review the existing controls and implement them directly in your applications, making them closer to, and synchronizing them with, the data usage. You can think of shift left as functional tests in software development that determine whether the business functions are working as expected. However, in data observability, shifting left consists of production validations based on the observations generated and validated as the application runs.

The shift-left step at the beginning of your project might not be enough. Often additional cases are discovered after application deployment, for which observations aren’t yet generated. In this case, you can use the data quality tools described previously to define new control points if you can’t fix or update the application in a timely manner.

However, you must communicate this need properly. This way, when you or someone else is tasked with “fixing” the application, you can implement the controls as observations and expectations. We say “fixing” because the application might not be seen as having a bug, and instead we might have to deal with a corner case that requires extra cleaning or business logic. Consequently, you can remove the controls from the data quality tools, reducing the number of dedicated resources to validate the data and maintenance of the orchestration.

Shifting left isn’t a one-time action to capture all the knowledge or migrate the data quality tools toward data observability. Rather, it’s an ongoing process and collaboration.

Corner cases discovery

Data application development entails choosing the data to connect and creating the transformations to apply, such as data selection, filtering, grouping, and joining. These choices aren’t random, but rather are the result of analyses, explorations, trial-and-error, and tests until satisfaction is reached.

All these activities generate knowledge about the data and the outcome, which conceptually is a large graph of interconnected information. This knowledge is mainly the result of assumptions we’ve accumulated during these activities, such as data completeness, available fields, and data freshness.

Because the data graph is so large, you must keep in mind the following considerations:

It’s an incomplete representation of reality.

It’s impossible to implement the code needed to handle all corner cases.

It’ll fade over time as you deploy the project and move to another one.

The first point indicates you might have made false assumptions that led to a misunderstanding. It also means you haven’t seen these cases because they aren’t represented in the data sets used during development. The bad news is these situations are hard to anticipate because they are classic unknown-unknowns. However, when you generate enough observations by default in your applications, not necessarily only those linked to known cases, you increase the probability of having some visibility when those cases occur.

The two latter points give you more to think about because they represent your known-knowns and known-unknowns at the time. However, they tend to fade away over time if you don’t do anything to persist them as expectations right away. Adding expectations to them gives someone else—such as a maintenance team—the opportunity to deal with these cases, so you can focus on your work, instead of jumping back on this one to fix it.

Why isn’t it always possible to implement the code to validate all cases? Well, there isn’t any good and valid reason but the time to value and laziness inherent to each of us. The reality is that adding code to handle all those corner cases can be boring. Also, because their probability is low, we want to demonstrate the fastest possible value, so we focus on the main cases and leaves those apart. If this wasn’t true, Null pointer expectations or similar would never be thrown because everyone checks that all values aren’t null.

Also adding code for each case considerably increases the complexity of the application. Therefore, it’ll increase its maintenance or even the probability of just rewriting the whole thing the next time someone—even yourself—looks at it.

Consider this example. You read a table with an email field that’s highly important for the customer-360 table you’re building. But your application logic returns an error if some values are null because you extract the domain from it by using a substring. You notice the number of null values is really low when you develop your application. You assume that’s the case, because you have so much data, so you decide to drop all rows with a null email and move on. Instead, you could have issued a query to another table that represents the customers’ business information, such as their employers, which would help you discover the domain with extra logic. However, it’s quicker and simpler to just drop them—“for now at least,” you think.

The point here is: “okay fine, fair enough.” But you must add an expectation that this number of nulls cannot increase dramatically or should stay below a certain percentage. This approach will save you countless hours of dramatic debugging sessions.

Lifting service-level indicators

When creating and maintaining trust in the data and associated applications you have deploy, you must conceive this trust as a combination of your own with the trust of your producers and consumers. Trust doesn’t happen in isolation. Rather, it’s a team effort, because your work is part of a larger picture that involves other people’s work with multi-layered dependencies.

This concept isn’t specific to data applications but is a well-known fact in IT and other contexts, such as the food chain, where trust must be continuous along the entire chain. Considering the importance of this trust, contracts are discussed and signed off to create the notion of accountability, which should lead to greater respect and consciousness to stay within the contract’s acceptable boundaries.

These contracts provide a way to create a relationship between the contract constraints and the cost associated with ensuring them. Consequently, it helps to create an environment where trust isn’t infinite—well, as long as the available money isn’t infinite. You must build the number of controls that must be respected so it’s viable and known by all parties. Without this type of contract, consumers will expect that everything is always 100% perfect.

In IT, these contracts are called service-level agreements (SLA), with their sidecars—the service-level objectives, a sort of contract with yourself. SLAs translate easily into the data world because they consists of determining rules based on arguments called service-level indicators (SLIs) as input. For data observability, the SLIs are nothing but observations.

If you’ve already defined SLAs, the associated expectations are available, in a limited, finite amount. If not, introduce them as early as possible. To help define them, use the available observations to propose to your data consumers (and data producers) a panel of observations that you can choose from to create the first set of SLAs. This concept is referred to as lifting observations into SLIs.

Using data profilers

When starting a new data project or application, a common issue is the lack of a priori confidence about what to expect from the input or output data. This situation can result from:

A lack of knowledgeable people, such as business or domain experts

Usage of newly acquired internal or external data

Various factors about the data such as its speed and size

Little input is available on what expectations can help you, including rules in your work. This case is somewhat of a cold start issue.

Fortunately, data profilers can be a source of a priori expectations. You’re likely to use them anyway during the exploration phase mentioned in the Corner cases discovery section.

A data profiler, which is often mixed with a data quality tool, “[examines] the data available from an existing information source (such as a database or a file) and [collects] statistics or informative summaries about that data.”⁶

It’s a lighter version of the process we introduced when we explained how to create data metrics. However, in this case, the data profiler is an independent tool. Leverage data profilers when possible even to produce data metrics within your application in production.

Examples of data profiler tools include Talend Open Studio, OpenRefine, Apache Griffin, or AWS Deequ. However, we recommend Great Expectations, which performs statistics and includes learning processes to discover rules, instead of only anomalies. You can use this tool at the beginning of your work on a data source or after you’ve created your own new data to generate rules for you to review that you can either accept or reject.

Note

Why you don’t want to auto accept rules: You might wonder why you can’t accept all rules automatically. Well, those rules will become one of the main sources of qualification of your work and how it’s seen by others. You don’t want to use rules that you don’t clearly understand the reasons for, consider as a representation of the current data, or anything else that would complicate the maintenance of your work in production.

Maintaining Expectations

The practices described in the previous sections to introduce and implement expectations in your work rely heavily on the status of the knowledge and data at a point in time. However, because data changes, the world changes, and knowledge improves, you must consider expectations as artifacts that require maintenance and reevaluation of their accuracy.

Consequently, you need to review the rules that could lead to an update process, like any application logic that might not match the reality of the business. You need to know when such maintenance is required, by introducing a periodic review process of the status, based on the observations to make the update process data-driven and less subjective. Without this review process, you end up having to do a post-mortem analysis of cases raised due. to missing or inadequate rules upstream, or when unknown-unknowns issues are encountered.

Having rules defined in a declarative way and stored in a separated system comes with additional benefits. You can make the update simply when reconfiguring a configuration file or updating the rules in the external system. Because this is more of an architectural detail, we address it in a later chapter.

Overarching practices

As explained, introducing rules in your applications has many benefits. However, you can benefit from other corollary practices.

Fail-fast and fail-safe

Fail-fast and fail-safe practices are efficient ways to increase confidence and keep the maintenance costs under control.

A fail-fast system follows an Agile methodology paradigm, where the system fails as fast as possible to avoid pursuing a flawed process. As quickly and early as possible, the associated practice moves on and tries with control points that highlight if the process reaches the expected level of quality or requires improvements. Following this practice, and making it clear to everyone, removes any stress or fear associated with deploying your project live, because you expect the first versions to fail…fast.

Moreover, finding the cause of a failure is easier in a fail-fast system because the system reports the failure with as much information as possible as close to the time of failure as possible.⁷

Therefore, expectations and their rules are key to turning your application into a fail-fast system. The iterations enable you to improve your applications and their results, as well as tune the expectations as you progress. Plus, implementing the observations and rules “at the source,” as explained in this chapter, enables the advantage of easing the cause analysis.

Similarly, a fail-safe system can also stop the process as early as possible, with a focus on avoiding the whole system from crashing (although you can still use it) and minimizing the effects on the other components.

For the data application part of the pipeline, introducing expectations as pre-conditions to execute logic or write (update) their results:

Stops their behavior early.

Provides enough observations to identify the reason.

Avoids “garbage-in-garbage-out” to invade the remaining of the pipelines.

However, the application might still be up and running and accept other tasks, because the failure was inherently due to the situation it ran into, not its behavior.

Simplify tests and extend CICD

After reading about expectations and the introduction of rules, a sense of similarity might come to your mind. You might think of rules as tests, such as unit or functional ones, because they usually represent examples of situations based on sample data and validate an implementation against its results. A test validates two types of outcomes: a thrown exception (or the code errored) or a match of the expected value.

With expectations and associated rules, we haven’t deviated from the test behavior and function. They were introduced to validate that the observations, which represent the status (by definition of data observability), match the expected values or behavior.

Therefore, making your application data observable by introducing those rules is similar to data testing in production, not data testing alone. But we’re not really testing in the sense of how testing is conceived in engineering, which is why we prefer to use the concepts validating in production or continuous validation.”

There’s a big advantage of leaving the rules in the main production code versus creating tests with the validation logic running against sample data sets (or pre-production data dumps). That is, by nature, the rules in data observability test the intrinsic logic, and not only the implemented logic, which is correct for certain predefined cases.

The other advantage of having the rules within the application is that it’s collocated with the logic. Therefore, when reviewing the implementation, you don’t need to jump between test cases and the logic to understand how it should work, or to update a wide range of tests. Of course, when the logic changes, you must refactor the rules, for example, but you don’t have to refactor the “test,” which is conceptually different.

Another advantage of this practice is you can’t deactivate the tests. Period.

Nonetheless, you still need tests. If we focus on the ones related to data observability, they can be largely limited to testing a single type of outcome in a situation where the rules are throwing exceptions. For this reason, tests are much simpler, because most, if not all, the test logic is in the main application source.

Consequently, in CICD, continuous integration means everyone saves their code in the main code base, and then tests and builds it. Continuous deployment refers to when the built deliverables are deployed automatically in several environments until production. This practice requires a certain level of commitment by your engineers, (especially for testing) and can be extended with continuous validation.

¹ https://en.wikipedia.org/wiki/Monkey_patch

² https://en.wikipedia.org/wiki/Polyfill_(programming)

³ Bagheri, Reza. “Understanding Python Bytecode.” Last accessed August 22, 2022. https://towardsdatascience.com/understanding-python-bytecode-e7edaae8734d

⁴ https://en.wikipedia.org/wiki/Macro_(computer_science)#:~:text=In%20computer%20programming%2C%20a%20macro,is%20known%20as%20macro%20expansion.

⁵ https://adv-r.hadley.nz/quasiquotation.html

⁶ Johnson, Theodore. “Data Profiling”. Springer, Heidelberg (ed.). Encyclopedia of Database Systems. 2009.

⁷ https://en.wikipedia.org/wiki/Fail-fast

Chapter 4. Integrating Data Observability in Your Data Stack

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 author at gobrien@oreilly.com.

In the previous chapter, you learned about the three components of data observability, and how to apply data observability in your day to day data work. In this chapter, we are going to get our hands dirty and put these concepts to work.

The goals of this chapter are to provide recipes that will help you integrate data observability in your pipelines and provide technical materials to up-skill your capabilities to personalize your frameworks and applications to make them data observable–then, as a good chef would do with any recipe: you will personalize it, extend it, improve it. I will explain the proper steps to follow, the purpose of each step, and how it works. Spoiler alert: it will get pretty technical from time to time–almost nerdy–but trust me, it’s worth it.

To give this chapter a logical flow, I will follow the Data Engineering lifecycle introduced by Joe Reis and Matt Housley in the Fundamentals of Data Engineering (Figure 4-1).

Figure 4-1. Fundamentals of Data Engineering, Chapter 1.: The Data Engineering Lifecycle¹

Data observability resides in DataOps as part of the undercurrents of data engineering. To generate the values discussed in Chapter 2, the data observability platform must be present in the data architecture and in the undercurrents layer. While I won’t discuss the Generation stage much, I am going to cover Ingestion, Transformation, and Serving and provide examples of how data observability supports their operations.

Data observability extends quite naturally to what the data engineering lifecycle produces, and the responsibility to satisfy the consumers is inherent to the mindset I have presented so far in this book. Therefore, I will also cover the data products built upon the served data, especially analytics (e.g., reporting) and machine learning.

Let’s start with the ingestion stage.

Ingestion Stage

The ingestion stage consists of moving data from the source, where business applications are deployed and used. Data engineers have the least amount of control of the applications, data, and the teams generating it. At times it’s similar to working with third-party vendors. So, ingestion is moving data from one system to another where we have control over it

Therefore, ingestion is the first line of visibility as this is where most downstream issues originate. Ingestion is, in fact, the stage when you can anticipate most datastrophes² and prevent dramatic propagations (ak.a., data cascades³).

The first step in ingestion is the connection with external or siloed data sources. It is at this stage where the data team loses the full control over the process. To regain its influence and more control, the data team must create higher visibility to clearly communicate their requests, backing them with facts such as metrics, trends, etc. Otherwise, the other party turns down or deprioritizes all requests and thus leaving the data team in a situation where the data must be continuously “cleaned”, instead of fixing the issue at the source.

As represented in Fig 2., ingestion is the first line of defense but it is also where there is less knowledge about what must be validated, as the purpose of data pipeline is to refine, combine, and generate more advanced insights. Moreover, the number of impacts fans out, as the data moves downstream towards many usages, which implies that data cascades starting here are greater (butterfly or snowball effect). Therefore, Because root cause analysis will often stop here, ingestion must be checked and updated regularly. Catching events in ingestion can avoid, like,85% of the issues downstreams, if not more. But because of lack of knowledge or it is hard to anticipate all use cases in advance without exploding the amount of checks, it will evolve as a consequence of post-mortems or because some metrics tracked anticipatively are already capable of highlighting anomalies. For example, an ingested data source could have one field used as filter in some transformations, or can be split in several groups to perform segmentations, or aggregated with another field from another data source to compute a business KPI. This list of usage can go over and over, and thus, the chances to implement all related checks at the very beginning (ingestion) is daunting if not impossible, actually I would almost say, irrelevant.

Figure 4-2. Ingestion first line of defense but with lesser knowledge usable for validation purposes

Adding anticipated metrics so early may be seen as useless though because the data is quite raw and there is little semantic knowledge attached to the information ingested. For example, one may be tempted to produce for each field in a JSON file a bunch of metrics such as min, max, stddev, mean, quantiles{1, 5, 25, 50, 75, 95, 99}, unique count, null values, skewness. So 14 metrics for each field to be computed and observed, for a JSON structure with 100 fields in its schema, it yields 1400 metrics, and so 1400 time series as the pipeline runs. Consider that the pipeline uses 10 data sources like this, you end up in 14000 timeseries, which can seem daunting and overloading the data observability platform with so much information that may never be used. However, in my opinion, their individual “expected value” is still huge in the event of an issue associated with the corresponding field.

Therefore, defining the semantic and usefulness of the observations is a challenge, so is the cost of producing, storing, and analyzing it. This challenge decreases in importance as we move down in the pipeline towards the business output, as the data gains in semantic and contextual usage, therefore, it becomes easier to generate observations to the outcomes (e.g., KPIs) as the number of options also diminishes.

Consequently, it is often preferred to start with a minimal set of observations (especially metrics) and increase over time the amount of observations published (especially metrics). This is mainly due to the associated cost and management overhead, if these are reduced to a negligible amount (as it will happen over time, ), the maximum of observations would be produced all the time to discover at the ingestion time. Nevertheless, this shouldn’t be surprising, as when a software goes to production, engineers decrease the level of logging from DEBUG to WARN, but can increase it during troubleshooting.

In the next section, I will cover some technologies to ingest data into the data platform and discuss how data observability capabilities can be added to each of them. But also I will cover what could have been provided by the technology to avoid users (us in this case) extending the capabilities ourselves.

Ingestion Stage Data Observability Recipes

In this section, I will highlight the Airbyte technology that performs ingestion by copying sources to destinations. Airbyte’s core engine is implemented in the Java language, and I will use an approach to make it data observable based on Java capabilities such as bytecode injection. This approach can be reused easily for any other technologies based on the JVM such as, for ingestion, Debezium⁴, Kafka connect, or Apache Flink. However, ingestion is not only limited to JVM technologies, for example, the Python ecosystem is rich of options (e.g., Airflow, Singer⁵). In fact, for python, the approach I will use in this section can be adapted to the python capabilities to achieve the same results, which I will introduce in the dbt section further.

Airbyte

Airbyte is an open source Java-based data integration platform that replicates data from the long tail of APIs, databases & files to data warehouses, lakes and other destinations.. Airbyte comes with a protocol to extract and load (ET) data in a serialized form (JSON messages) which is already available as open source connectors for many data sources. While ingested data, you can also normalize the messages using the embedded dbt (covered in the next section) which will, for example, linearize the deep structure in the ingested JSON messages (e.g., {“name” : {“first”: …} } normalized as name.first). Moreover, Airbyte has developer kits to implement easily and rapidly new connectors (or customize them).

The transfer of data between the source and the destination implements the Airbyte Protocol⁶ which describes a series of standard components and all the interactions between them in order to declare an ELT pipeline. All message passing across components is done via serialized JSON messages for inter-process communication. The processes are running inside workers, between which the data is transferred using stdin and stdout.

Figure Fig 3. Shows the architecture involving the workers and their interaction with the Airbyte scheduler that takes care of their lifecycle via the temporal service. And thus, the JSON messages are exchanged between the Source (e.g., PostgreSQL) and Destination (e.g., Snowflake) workers.

Figure 4-3. Airbyte Architecture Overview

Also Airbyte makes available some data observations via its API⁷, such as:

Get a connection: Application and Lineage…

Get information about a job: Application Execution and Lineage Execution

At the time of writing, Airbyte includes the following data observability capabilities via the MessageTracker⁸ which is a Java class that accumulates information (observations) about the execution of a connection between Source and Destination, such as:

Get the overall emitted record count.: Data Metrics for the input data source (Source).

Get the overall committed record count: Data Metrics for the target data source (Destination).

We could also consume data from the logs combined with API calls to create some of the observations. That would only partially cover the model introduced in Chapter 2 and miss important information, thus will not provide the value we expect from data observability.

This is true for all technologies I cover in this chapter, for the time being when Airbyte is not supporting data observations yet (which will most likely happen over time), let’s extend it to cover the missing pieces.

Airbyte Agent

Because it is likely that you already have Airbyte jobs (connections) deployed and running in your platform, the work needed to change all jobs to generate data observations could be daunting work. So in this section, to minimize the efforts, I am using a strategy that will inject data observations generation capabilities at a lower level, the Airbyte core as shown in Fig. 3”.

Figure 4-4. Airbyte data observability agent extending the Airbyte core engine

As explained in the introduction above, Airbyte has a component called a worker, which is responsible for the execution of the replication jobs. The worker’s code is in Airbyte’s GitHub as a module called airbyte-workers.

The worker executes the jobs configured in Airbyte the Replication and the Normalization. The first replicates the source into the destination (the Extract-Load parts of ELT) and the latter transforms the raw data created by Airbyte in the destination into the required form. If normalization has not been configured, the raw data will be replicated as-is to the final destination table.

The normalization step is not covered here however, as it will be covered in the next section so I will concentrate on the Replication jobs in this section.

The replication is defined in a Java interface called ReplicationWorker, which mainly contains a run function using a StandardSyncInput (configuration of the replication) to produce a ReplicationOutput (summary information of the process). Both objects are defined in the Airbyte config-models–the model objects of the Airbyte protocol defined using the Open API standard.

Because we want to enable data observability for the replication job, a good starting point is to generate the observations about the source, destination, lineage, and metrics. To do so, we’ll use Bytecode Injection(see Chapter 3) to avoid changing, recompiling and redeploying the Java code.

The code for this agent is available in the book’s repository⁹ and is composed of two Java projects:

airbyte-agent

Implements the generation of data observations. The airbyte-agent initializes the client that logs the observations, extracts metadata from source and destination, plus computes metrics for them.

airbyte-instrumentation

Implements the Java-Agent to intercept the execution of the run method before it is loaded into the JVM and to inject the airbyte-agent’s code to it.

If you need to extend another java based data system such as Airbyte, this combo of projects is giving you the starting point. Let’s see the key elements of each project to extend Airbyte with additional data observability capabilities.

The airbyte-agent project defines the following concepts.

Agent Factory

A factory that can generate new data observability clients.

Agent

The agent is performing most of the actions such as extracting metadata using the processors and complementing them when data is read (handleMessageRead), mapped (handleMessageMapped), and copied (handleMessageCopied)–including the generation of Data Metrics. But also, the agent is initialized with project information so the application and related entities of the core model can be generated.

Processor

A processor can be defined for the source and destination. It is responsible for the generation of metadata (Data Source, Schema) of the data sources being replicated.

Note

You will see this pattern quite often in the data engineering tools ecosystem, and I’ll come back to it often in this section. The source and destination in Airbyte are an abstraction that allows the tool to be extended and support new data sources, therefore, it defines a contract (API) to read and write to them. Therefore, for the agent to support exhaustively all types of data sources (e.g., CSV, BigQuery, MySQL, Snowflake, etc) needs to be updated to support each of them, which can seem complex especially as new data sources are continuously added to the Airbyte catalog. .

Alternatively, if this responsibility is introduced in the Airbyte’s abstraction layer (Source and Destination contracts), then data observability would be implemented by-design, and thus removing the need of this whole section about the agent..

While the Airbyte agent isn’t in the code base, this project extends Airbyte’s capabilities by using its internal concepts to inject it.

To inject this code, we need to look at the airbyte-instrumentation project, which produces a JVM-Agent capable of executing code before the application’s main is executed. To do this, it uses a JVM capability, called the premain class.

The Premain class¹⁰, provided by the jar attached to the Airbyte Worker JVM using the -javaagent property, is a class that is supposed to be executed before the main class, as it names suggests it (pre-main). The Premain class is a special class the JVM knows that it must call its static method premain before the main one , this gives us the opportunity to inject the airbyte-agent code into the ReplicationWorker before it gets used.

Here is the signature of the premain method:

public static void premain(String agentArgs, Instrumentation inst)

Example 4-2. Code 0.1. Signature of the JVM-Agent’s premain method.

The Instrumentation argument passed to premain by the JVM is what is used to transform the ByteCode as it is loaded. To achieve this, we can use its method addTransformer to which we’re providing an instance of a ClassFileTransformer. The JVM will pass all loaded classes to this transformer, including the ReplicationWorker, that we can intercept like this:

public byte[] transform(ClassLoader l, String name, Class c, ProtectionDomain d, byte[] b) {
  // The class to be injected
  if(name.equals("io/airbyte/workers/general/DefaultReplicationWorker")) {
    // [...]

Example 4-4. Code 0.2. Intercept a class using JVM-Agent Premain

As shown in the Code 0.2., the transform method is also provided the bytecode for the class as a byte[], that we can transform with the airbyte-agent code before returning it. This can be done with the ASM library discussed in chapter 3, using four classes:

ClassReader

Reads the bytecode of the class into the structure that can be visited by a ClassVisitor

ClassWriter

Writes a provided structure from a ClassReader and writes it as bytecode (byte[]).

ClassVisitor

Wraps a ClassWriter which will delegate to the ClassVisitor’s visiting methods (e.g., visitMethod, visitCode) the information from the ClassReader so that it can transform them before being written.

MethodVisitor

Instantiated by a ClassVisitor’s visitMethod, it has its own set of visiting methods to transform the contents of methods.

The ClassVisitor is the most important piece of code to be analyzed to understand how to turn the airbyte process into a data observable one. Despite the fact that the instrumentation is relatively simple and straightforward, its code is quite long, therefore let me focus on the key aspects.

The instrumentation implements a ClassVisitor for the ReplicationWorker that intercepts its run method by overriding the visitMethod. When the run method is intercepted, it uses ASM to include a call to the AgentFactory (from the airbyte-agent project) and instantiate a new agent for the replication work.

To do so, the instrumentation defines an instance of a MethodVisitor which is intercepting the three main steps of the replication process, for each message:

Its read from the source: by intercepting a call to the function attemptRead on the Source object and calling the agent’s handleMessageRead to generate observations about the source.

Its mapping to the destination format: by intercepting a call to mapMessage and calling the agent’s handleMessageMapped methods to maintain the information about the lineage.

Its write into the destination: by intercepting a call to accept and calling the agent’s handleMessageCopied method to complete the lineage and generate observations about the destination.

With this in place, the replication process is fully observable, and data metrics are accumulated to the source and destination from each message read, mapped, and written.

Using bytecode injection helped us add data observability capabilities to existing Airbyte servers. However, it would have been preferred that Airbyte itself exposes its data observations, to allow a data observability platform to consolidate it in the global picture. This is what we discuss next.

Transformation

While the ingestion stage is the very line of defense against incoming data and its inherent responsibility is quite limited to moving data from A (e.g., the ODS) to B (e.g., the Data Lake), the complexity doesn’t come from “what” is done with the data, but the lack of control on the incoming data.

The transformation stage, however, is dedicated to manipulate the ingested data into as many forms as needed for downstream usages such as analytics or machine learning, or more generally data products. Therefore, what is important to create at this stage is visibility about how the transformations are behaving, including their interactions, along pipelines. Because, assuming that the data is somewhat under control at the ingestion stage, then, issues happening in the transformation stage are more likely associated with inappropriate transformations that could be due to misunderstanding of the business needs, time-sensitive expectations about the data, and more.

Consequently, in this section, I will focus on how to create data observations about the manipulations of the data and which observations are interesting to generate to handle issues more easily. For this, I will use two different technologies, Apache Spark, then dbt.

Recipes

Apache Spark

Transforming data at scale historically was a challenge for data engineers until Apache Hadoop popularized Map Reduce. However, Hadoop was not straightforward to set up and use, for example, after the installation of several projects, to test your first project, you would have had to create a maven project, create a few Java classes with your business logic decomposed into Map-phase and Reduce-phase, and finally combine those together. It was not easy enough for Big Data to become mainstream (i.e., usable by many more people) as the entry barrier was too high. Eventually, this is what Apache Spark was capable of doing as a consequence of two things:

Using Scala as the programming language, leveraging the functional approach pre-Java8 and the Scala interactive shell

Introducing a new paradigm to implement MR jobs called Resilient Distributed Dataset (RDD), the functional alternative to Map-Reduce classes implemented at the Collection level.

As Spark became the de facto standard to manipulate data, it embraced the DataFrame API, which follows the relational principles and therefore offering also SQL-like capabilities. This introduction could expand even further the community of Spark users to the Python and SQL community, so far limited to the Scala one. Hence, Spark is one of the technologies that enabled the explosion of the data economy since the mid-2010s.

In Fig 4., I am showing a quick analysis on Google search terms for terms such as Big Data and Spark, we see the Big Data crosses the line of Hadoop which starts declining when Spark gets traction. Then, what’s interesting to see is the close relationship existing between (py)Spark and pandas, in Python, discussed in Chapter 3.

Figure 4-5. Google Search Trends for Big Data, (py)Spark, Hadoop, and pandas¹¹

We’ll see in this section how we can leverage one of the core aspects of Spark to accurately generate many data observations through automation. This approach uses directed acyclic graphs, or DAGs, which results from using RDDs to compose computations in a functional (descriptive way).

The DAG is not a unique Spark concept per se but has been further popularized by Spark. For example, DAG has been embraced by the Google ML library tensorflow¹² as well as by high-level tools like open-source orchestrators (e.g., prefect, airflow) which will be covered later in this chapter.

DAG is key to support data observability, in Spark but similarly to any DAG based frameworks, because it is providing the support to generate important observations on how data are composed as it is natively available in its structure. Each vertex in the graph represents a function applied on the incoming input to generate outputs that will be, in turn, consumed by other vertices. Hence, by making each vertex data observable, the whole graph becomes one, because only data is moving from one vertex another. In other words, each vertex represents a micro-application, a micro-transformation on data in flight (not stored on physical storage for example), see Fig 5.. It is worth noting that a DAG is the theoretical representation of a programming paradigm, called dataflow programming¹³, dedicated to solving computing issues coming with sequential programming that requires the (data) state to be managed by the developer in a distributed environment. Spark is one of the libraries that follows this paradigm.

Figure 4-6. A database sorting operation implemented in Prograph¹⁴ in the 80-90s

Because Spark is deployed at scale in many organizations, some efforts have been made to extend its capabilities with data observability, such as the Spline project that I introduce next.

Spline

Spline¹⁵ is an external component that generates observations about data applications, data source metadata (location, schema), lineage, and execution information.

Adding Spline to a Spark job is a matter of adding the Spline Agent to the Spark classpath and installing the Spline Server (and the associated ArangoDB database) to aggregate the observations that will be presented in the Spline UI as shown in the picture Fig 6.

Figure 4-7. Spline UI showing the lineage of a Databricks Spark application

The Spline Spark agent can be initialized in a codeless fashion by adding the following configuration to the job:

--conf "spark.sql.queryExecutionListeners=za.co.absa.spline.harvester.listener.SplineQueryExecutionListener" \
  --conf "spark.spline.lineageDispatcher.http.producer.url=http://localhost:9090/producer"

Example 4-6. Code 1. Spline agent codeless configuration

From the configuration, we can see that this agent will use the events generated by Spark to create visibility about some situations, such as writing to a data store.

For this, Spline uses a component that Spark has already exposed, the QueryExecutionListener, which defines this method: onSuccess(funcName: String, qe: QueryExecution, durationNs: Long)¹⁶. In fact, Spark triggers many SparkListenerEvents¹⁷, these can be used to generate observations such as the data lineage and data metrics, the onSuccess function, however, intercepts mainly the events related to the job after successful completion. Therefore, when a job is successfully executed, Spline can extract data observations from the QueryExecution object that is provided to the function.

The QueryExecution is key in this scenario because it contains a reference to the LogicalPlan that represents the DAG of the executed job. Hence, by traversing the graph of computation encoded in the LogicalPlan, Spline can extract references to the data sources used by the job, as well as their schema and associated lineage.

At first glance, this is all fantastic, as it seems Spark combined with Spline implements a lot of what we have described in the first chapter. Nevertheless, one notable downside is that Spline’s architecture is fundamentally Spark-centric and highly coupled to its own server and UI. What we want is to allow the data observations to be used across technologies, in order to provide a holistic view of the data system, and therefore accelerate tedious operational processes such as root cause analysis.

So what we need to do is to intercept the observations generated by Spline and turn them into the core mode (at least), and allow them to be made available to a data observability platform. To accomplish this, we can use the CompositeLineageDispatcher provided by Spline, this allows us to dispatch the observations to the original Spline Server, but can also be adapted to use alternative endpoints, such as files or another HTTP server.

Several patterns can be implemented to achieve this conversion and redispatching, like sending the observation to the data observability platform, which turns them into the core model (cf. Chapter 2) or implementing a dedicated LineageDispatcher¹⁸ that conversion into the model itself.

The latter option allows the observations to be manipulated and linked following the core model directly in the agent instead of arriving disconnected to another service that will have to reconsolidate the graph, potentially dealing with the ordering of the observations by arrival time. – As we are talking about HTTP communication.

Here is a skeleton of what such a custom dispatcher could look like (in Scala):

class FODOLineageDispatcher(conf: Configuration) extends LineageDispatcher with Logging {
  type DataObservationBase = Any
  type Observations = List[DataObservationBase]
  // Create client to publish data observations using configuration
  val client:Any = ???
  // Navigate the plan to convert it into the core model
  def convertIntoDataObservations(plan:ExecutionPlan):Observations = ??? // use client to publish
  // Send observations to a platform, a file, or anything collecting them
  // Use client to publish
  def reportDataObservations(obs: Observations):Unit = ???
  override def send(plan: ExecutionPlan): Unit =
    reportDataObservations(convertIntoDataObservations(plan))

  override def send(event: ExecutionEvent): Unit = ()
}

Example 4-8. Code 2. Spline custom dispatcher

To use this dispatcher, you will need to build it in a jar, put it on the classpath of your Spark application, and configure Spline to use it, as shown in the code below:

spline.lineageDispatcher.cust.className fodo.ch4.transformation.FODOLineageDispatcher
spline.lineageDispatcher cust

Example 4-10. Code 3. Configure the custom dispatcher

This way, you will skip the Spline server (installation and usage), but use your own data observations collection strategy. Nevertheless, if you want to keep both options, there is also the possibility of using the composite dispatcher.

Using the strategy presented so far, we can leverage the information gathered by Spline in the ExecutionPlan to generate the observations as presented in chapter 2. Here is the information available in its JSON representation:

“name”: the name of the application that ran the job (e.g., Databricks Shell)

“operations”: contains the information about the job’s structure, including:

“write”: information about the data source created or updated by the job. This contains the location of the data source (e.g., file path) and its format (e.g., parquet). It also embeds an “extra” field that can be used to attach additional information (see next section)

“reads”: information about the data sources used by the job to generate the data written (see above). This is an array of entries having the same information as in “write” above for each data source used.

“other”: information about the operation performed (e.g., joins, subquery).

“expressions”: contains information about the “functions” and “constants” used in the job.

“attributes”: the entries in “operations” and “expressions” are referring to columns or fields used by the job to generate the output. Those are detailed in this array, which contains an entry for each with information about the name and a reference to its type (see below).

“extraInfo”: additional information that can be extended also, typically contains:

“appName”: which can be changed to match the application name, instead of the job name introduced earlier.

“dataTypes”: which provides information about the data types for the “attributes” (see above).

We can see that Spline generates a lot of information that will help us generate a big part of the core model as Spline was primarily created to support compliance requirements (e.g., GDPR, BCBS239), and so its capabilities are well adapted for those use cases. In the figure below, you can see what part of the core model you can cover from Spline:

Figure 4-8. Spline’s core model coverage, and its extensions.

From Fig 7., we can see that a big part of the model can be extracted from Spline information, it is represented by the bold lines. Here is why and how:

Application: using “name” or “appName”.

Data Source: from “write” and “reads”.

Server: from “write” and “reads” we can extract where the data resides. But we don’t have information about the server executing the application.

Schema: from “attributes” as referred from “operations” and “expressions”, combined with “dataTypes”.

Lineage: by interpreting the “operations” and “expressions”.

Lineage Execution: the job completed with the plan, or the ExecutionEvent can be used otherwise.

Next, Fig 2. shows dashed lines which represent the information that can be generated from the application runtime during the interception of the plan:

Application Execution: this can be extracted from the runtime (e.g., System.currentTimeMillis to retrieve when the application was running).

Server: we can use local code to extract the information about the server (or cluster) running the spark job.

User: the information about who is running the application can be fetched from the runtime (e.g., $USER). However, it is trickier to know who developed it (see below).

The last part of the model covered by a Spline-based agent is shown in dotted lines and must be retrieved by integrating with the environment that built the job, such as the SBT project definition for example. The idea is to generate information such as Application Repository (e.g., git URL), Application Version (e.g., git tag, or version defined in SBT), and User (e.g., git committer) at the build time and made available at runtime, for example in a manifest file. Note that, this can be added also as a continuous integration step.

As stated above, some capabilities of data observability are missing in Spline, namely the Data Metrics, hence let’s see how we can take Spline to the next level in the next section.

Going beyond the compliance use case

At the time of writing, Spline is focused on collecting data observations from the Static Space, such as DataSource, Schema, and Lineage. The other kind of observations from the Dynamic Space, that Spline collects is pipeline metrics, which are part of Application Observability (in this case, a subset called Pipeline Observability). However, we miss an important part of the Dynamic Space of the core model, the Data Metrics to control the behavior of the data during its usage or creation.

In this section, I will provide the key elements that will allow you to extend Spline with this capability. Because the full implementation can get pretty complex, I will present a high-level picture of the code. However, the book repository contains a good starting point to work on the agent you can add to your projects (or use an existing implementation such as the one provided by Kensu at https://docs.kensu.io/pyspark).

What we want to do is to create additional visibility about the data going in and out of a job, not where they are, but what they look like from the content perspective. So that, if something goes wrong, we have many observations at hand to explore the issues, or if we want the detect something, it can be done during the execution of the job, which is the last chance before falling in the garbage-in-garbage-out scenario (again), or to fail the job even if the “write” has happened, but wrong data was written. So we want data metrics.

To do that with Spline, we can use an extension mechanism called “Filter”, in fact, a “Post Processing Filter”, which gives us hooks to update some part of the ExecutionPlan used in the previous section to generate the model. These hooks are around five cases:

Data read.

Data written.

Data operation executed.

Execution Plan available.

Execution Event (more general).

The cases we are interested in the most are when the data is read and written¹⁹ as they give us the opportunity to generate observations about the data involved in the job.

To implement that, we have to implement a post-processing filter by extending the AbstractPostProcessingFilter²⁰ (default empty implementation), adding it to our agent, and then configuring Spline to use it (see below). The skeleton of such filter is available in the repository provided with the book, here are some hints about a couple of parts:

Generate metrics: to generate metrics on the data read or written, we can leverage Spark itself, to do this we can browse the LogicalPlan provided by the hook, which can be turned into a DataFrame with a trick that is hidden in the Spark code base: Dataset.ofRows(spark, plan). Using the returned DataFrame we can easily generate many observations by design (automatically) such as the amount of data (count), distinct values, null values, or even statistics (e.g., min, std, or skewness).

Make metrics available: because metrics are not defined in the Spline model, we have to use the extraInfo to save the generated Data Metrics for each data source. This is not very convenient, as we lose the types or the structure of the object to give us, but a convention can be used easily and shared with the dispatcher (introduced in the previous section) that will encode them in the model.

Note

For the sake of simplicity, the example in the repository only relies on the ExecutionPlan, not on the reads, and write data, but the same approach can be applied by decoupling the work across the different events instead of only when the plan is available.

For Spline to apply this filter, it simply needs to be in the same jar as the dispatcher, which is on the classpath, and those configurations to be added to the job:

spline.lineageDispatcher.cust.className fodo.ch4.transformation.FODOLineageDispatcher
spline.postProcessingFilter fodo-filter
spline.postProcessingFilter.fodo-filter.className …FODOPostProcessingFilter

The configurations are defining which filter needs to be applied (fodo-filter) and which class is implementing it (fodo.ch4.transformation.FODOPostProcessingFilter).

For completeness, it is also worth noting that Spark has, since its version 3, started to introduce “observations” as this book defines them. In fact, there is now a “observe” function provided directly by the DataFrame interface that allows additional metrics to be computed alongside the Spark job. Therefore, the engineer developing the job can “instrument” it directly with them, and they will be available in the filter, so that they can be added to the extraInfo before saving them contextually with the other observations. This is a nice way to introduce “custom metrics” as opposed to default metrics which are generated automatically by the agent. Such custom metrics are generally extremely important as they are based on domain knowledge instead of being generic, and can be quite complex (e.g., multiplying fields, or even involving several data sources).

In this section, we have seen how an existing agent can be extended to generate additional information about a technology such as Spark. Over the years, either such agent will proliferate for most technologies, and so it is worth keeping an eye on them (such as those provided by Kensu). Also, as discussed before, it is likely that the data technologies will embrace them directly with requiring external agents, as Spark started to do with the “observe” method. Next, let’s take a look at another data transformation tool that can be extended from its python core, this time, to generate data observations–namely, dbt.

dbt

dbt is a SQL-first transformation workflow easing best practices such as modularity, testing, and documentation. dbt has gained popularity over the years by leveraging SQL to simplify the work to build data models, without necessarily requiring coding skills (e.g., python, or Spark).

Given that dbt is responsible for transforming data sources, it must also be data observable, which some aspects are already available out-of-the-box. In this section, we will see what is available, then show how a system such as dbt can be extended with additional data observability capabilities.

When defining a model in dbt, the engineer builds a SQL query that will rely on either seed or other models, in their simplest form a seed is a file that will be loaded into a model, a model is a relational database table. Along the lifecycle of a dbt project, the system will generate observations about the job, which are available in the below artifacts as JSON files generated by dbt commands (e.g., build, compile, run, test)²¹:

manifest²²: contains many information such as models and lineage.

catalog²³: contains the metadata of the data (e.g., schema, location).

run results²⁴: contains the results of commands, including the tests.

Another artifact is the sources that contains a specific type of check performed by DBT itself: the freshness test. This test provides metrics (and associated check) about the data at the moment of its usage complying with the principles of synchronization of data observability introduced in chapter 3, and the concept of expectations in chapter 2.

The good news is that dbt already embraces data observability as most of the observations of the core model is produced. Nevertheless, collecting them requires extra effort to combine the information coming from all these files, which may not be produced in certain circumstances (e.g., failures).

Therefore an agent for dbt that produces the observations in the core model along the process could be done by adding an external component that would collect the information from those files when they are ready. This can be relatively easily achieved, however it would probably have to run outside of the “box”, which is contrary to the notion of “observability” as defined in chapter 1.

Note

Such as the external listener²⁵ provided by the open source data catalog, DataHub, which connects to the dbt HTTP endpoint or s3 to retrieve, parse, and process those files to create data catalog entries.

Moreover, as in the Spark case, the Data Metrics wouldn’t be available in the JSON file, only perhaps what is available by the tests, but those are generally only available as part of the CI/CD, which means that the test command is not run in production, but serves as a gate to validate if a model can go into production, meaning that, the metrics (and checks) associated with the tests won’t be available when the dbt project is run.

In the next section, I will propose another approach that consists of extending the dbt core module (the central engine of dbt), with additional capabilities which can leverage the information used to produce the artifacts, but also add more capabilities, such as the generation of Data Metrics for the seed and models.

dbt-core agent

In this section, I will provide you guidance on how to extend an existing python application such as dbt-core, to introduce directly in the engine the data observable ability. This agent will use a technique introduced in chapter 3, the AST based transformation of the code at runtime.

Note

The term dbt-core refers to the dbt framework that can be cloned from GitHub or cloned from python repositories, this avoids the confusion with dbt-cloud which is the SaaS version of the framework.

To use this strategy, we need to dig a bit deeper into the dbt core code available on GitHub²⁶ to analyze where we can ingest, as we did for the Airbyte agent, the needed functionality that will intercept or generate the observations.

First of all, when dbt runs it will run a “task”. A task is implemented as a class called ModelRunner. This class is responsible for running what was developed by the engineer, and therefore what we want to make observable by default.

A ModelRunner can represent mainly²⁷ a seed task or a model itself, and an instance of that class contains all the observations, or will give us the opportunity to generate what is not provided by default. This happens in its execute²⁸ method defined as such:

def execute(self, model, manifest):

Example 4-13. Code 3. dbt “ModelRunner.execute” method signature

As we can see, this method has the knowledge of the model to be executed but also the manifest which gives us access to the artifacts introduced above, like this:

context = generate_runtime_model_context(model, self.config, manifest)

Example 4-15. Code 4. Generate dbt context (artifacts)

This will give us the information about the job (name, sources, etc). Therefore, this is a good place to initiate the collection of observations. Because, it is not really an option to change the code of the installed dbt²⁹, I will explain how to inject it similarly to Airbyte. Let’s see how we can achieve this, then I will address the next part of the generation of the observations related to the data involved in the task.

The approach is to inject additional python code “at-runtime”, by intercepting when the ModelRunner.execute method is called and when the context is built. To do this, we’ll use a python capability called “site”³⁰, which is allows us to define a “Site-specific configuration hook” by running a specific file named “sitecustomize.py”³¹.

When this file is made available on the “site-packages” path that will be used by python to load modules, the python runtime ensures that it will always be executed before anything else. Therefore, this script can be used to run additional code that will be injected into the dbt runtime before the “ModelRunner.execute” is loaded and executed.

Thus, the sitecustomize.py script will use another module called ast which gives us the capability to parse python code into its AST and modify it. This is how we’ll inject our agent code to generate the observations. In the following, I will provide additional details how this can be achieved, however, you can refer to the book’s repository to see the full code.

There are two steps to follow to inject code:

Intercept when a module and its functions are loaded, and parse it.

Modify the parser AST of the module to inject the code that will generate the data observations.

The first step is achieved by using the importlib package which is nothing but the package that implements import. This package provides two classes³²:

importlib.abc.MetaPathFinder: a class with a find_spec method to find a module to be imported. This method returns a importlib.machinery.ModuleSpec that contains the import-related information used to load a module. MetaPathFinder instances are found on the sys.meta_path.

importlib.abc.Loader: this abstract class is used to (re)load and execute a provided module (found by the Finder above). It comes also with the exec_module method which executes the loaded module.

Here is how to combine these two classes, starting with the ModelRunner:

class DBTAgentPathFinder(MetaPathFinder):
  def find_spec(self, fullname, path, target=None):
    # [...]
    if fullname == "dbt.task.run":
      return ModuleSpec(fullname, DBTAgentLoader(fullname, p))
    # [...]

class DBTAgentLoader(Loader):
  def __init__(self, fullname, path) -> None:
    # [...]
    source = Path(path, module_file).read_text()
    tree = parse(source)
    # TODO new_tree = ???
    self.code = compile(new_tree, 'exec')

  def exec_module(self, module):
    exec(self.code, module.__dict__)

sys.meta_path.insert(0, DBTAgentPathFinder())

Example 4-17. Code 5. Intercepting a module load with importlib in sitecustomize.py

The key elements to notice in Code 5. are:

fullname == “dbt.task.run”: intercepts the file run.py containing the ModelRunner.

ModuleSpec(fullname, DBTAgentLoader(fullname, p)): returns a DBTAgentLoader for this intercepted module.

DBTAgentLoader.__init__: reads the module (.py) file and parses into its AST tree, using ast.parse. This tree is what needs to be modified in the second step explained below.

self.code = compile(new_tree, 'exec'): compiles the new tree that will be executed.

exec(self.code, module.__dict__): implements the exec_module method of Loader to load the self.code, that is, the code that was modified using ast.

sys.meta_path.insert(0, DBTAgentPathFinder()): ensures that the new finder we have defined is located on the meta_path where finders are searched for by python.

The second step is to modify the tree parsed, to achieve this we’ll use the mechanisms coming with ast called NodeTransformer. A NodeTransformer is a visitor with corresponding visit_XXX methods that can be overridden to intercept and modify the visited node XXX. In our case, the two types of node that are interesting are:

ClassDef: the definition of a class (e.g., ModelRunner).

FunctionDef: the definition of a function (e.g., execute).

Modify the code is in reality much simpler than we could anticipate, here is what it looks like to inject the data observability agent in the execute function’s code after the context has been initialized (as said above):

class DBTTransformer(NodeTransformer):
  CURRENT_ModelRunner = False

  def visit_ClassDef(self, node):
    if node.name == "ModelRunner":
      DBTKensuTransformer.CURRENT_ModelRunner = True
    # [...]

  def visit_FunctionDef(self, node):
    # [...]
    if DBTTransformer.CURRENT_ModelRunner:
      if node.name == "execute":
        new_node.body.insert(1,
          ImportFrom(
            module='fodo_dbt_agent,
            names=[
                alias(name='dbt_agent', asname=None)
            ],
            level=0,
          )
        )
        new_node.body.insert(2,
          Assign(
              targets=[Name(id='__fodo_agent', ctx=Store())],
              value=Call(
                  func=Name(id='dbt_agent', ctx=Load()),
                  args=[
                      Name(id='context', ctx=Load()),
                      Name(id='model', ctx=Load()),
                  ],
                  keywords=[],
              ),
              type_comment=None,
          )
        )
      # [...]

Example 4-19. Code 6. Define a NodeTransformer to modify ModelRunner.execute

In Code 6, we start injecting the data observability code, starting with the initialization of the agent that uses the context and the model variables to create observations about the application, the data sources, and even the lineage defined by this dbt model.

To do so, here are the key steps:

if node.name == "ModelRunner": because the tree represents the whole run.py in this case, there may be several classes defined into it, and we’re only interested in ModelRunner. This code allows the transformation to only be active for this class.

if node.name == "execute": when visiting the ModelRunner class, we are interested in the execute method only, so we can extend its code. This is the starting point.

new_node.body.insert(1,: this is where we start inserting code into the current function’s parsed tree – we add a new node after the very first one, that, in the case of execute is initializing the context which we need to create observations.

ImportFrom: builds an import AST node, with the module name fodo_dbt_agent that needs to be imported as it will contain our data observability agent code.

Assign: a variable assignment node, following the import, that will create a variable __fodo_agent, to be initialized with the data observability agent (see below).

Call: an AST node representing a function call, in this case, it will call the dbt_agent function with the context and the model as parameters. Those variables are provided by the ModelRunner.execute local variables.

At this point, we have all the mechanics to intercept and change any aspect of a python based system such as dbt. What remains to be done is to create the agent itself, the module fodo_dbt_agent referred to in the list above. This code is provided in the book’s repository, and is quite long, as we must generate a lot of observations. However, I’ll drive you through the important parts which are related to what is missing from dbt supported observations, that the data metrics, as pretty much the whole core model can be generated from the context and model.

In order to generate data metrics observations for the data sources involved in a seed or model task, we must first understand how they are used. dbt works on top of a templating system called Jinja. dbt provides jinja templates to allow the engineer to manipulate data sources, via adapters. An adapter³³ is a contract that must be implemented to support a new data source (e.g., Snowflake, Delta Tables, MySQL), so it can be implemented independently of the dbt core by the community. The adapter instances are created by dbt core and made available to the jinja execution context.

The instantiation of adapters and the execution of the jinja task defined by the engineer is executed in the ModelRunner._build_run_model_result, however this method is only calling the jinja engine. Most of the work to access the data sources, read them, and process them, is done by the adapters.

What we are looking for is a handle to the data sources used in a task, so we must also intercept the work done by the adapters to generate the data observations.

Before giving you the solution on how to do this, it is worth pausing and notice again that the responsibility to generate data observations must be given to the teams who are creating the tools used by engineers, natively. Because each data source will have its own complexity and behavior (hence the adapters), it is more complex for a generic data observability agent to handle them all. Of course, the agent will come with abstraction such as adapters to allow its own extensivity. Though, it would have been more convenient if the dbt adapters would have already provided this capability.

In order words, the dbt adapter contracts should evolve to also expose data observability capabilities, such as data metrics. this is already the case for the metadata.

To close this section on dbt agent, here are two hints to generate observations about the seed and the model run by dbt.

The seed task is a specific jinja template that reads a file (e.g., CSV) into an agate table. Where agate is a python package to perform data manipulation, and for this also provides a SQL-like API on top of files loaded as Table. dbt will load the seed file into an agate table using the ProviderContext.load_agate_table method, where the ProviderContext is of course what feeds the jinja with the context, and load_agate_table the method that reads the file into the table.

Therefore, the way we can intercept and compute observations of the loaded files can be to add another interceptor for the above method in the sitecustomize.py.

For the model task, however, this is where adapters come into play, and we must extend our coverage to all of them to be fully data observable. This can be achieved if we intercept a method defined at the adapter level, such as BaseAdapter.execute. But we also have the opportunity, for SQL adapter, to intercept SQLAdapter.add_query. Nevertheless, all adapters have also to implement a connection which provides, of course, a handle on the data source. Therefore, we can simply intercept the SQLConnectionManager.add_query and use the provided connection to execute additional queries on the data source, such as computing the number of nulls, the distribution of all columns, etc. For an example of such implementation, please refer to the kensu_postgres.py file in the book’s repository.

In this section, we saw that data transformation systems have been partially implementing data observability, and probably will continue to align with what we have covered in this book so far. However, because we need it now, you have in hand the strategies that can be used to augment those capabilities now.

Let’s now move to the next phase of the data engineering lifecycle, where the data is served to the next users and see what could be done in this phase to support data observability.

Serving

At this stage, the data is ingested and transformed, yet it has to be made available and, most probably, in a continuous manner for consumers described in the next section. So in this section, I will address first how the data can be made available without breaking the data observability chain in a data warehouse such as BigQuery. Then, I will address the second point, which is how to ensure the data is continuously available using an orchestrator which is responsible to run the pipeline according to the consumers expectations (e.g., freshness).

The Serving stage is the last frontier where data engineers have influence, and therefore, the last line of defense to ensure data observability. Consequently, it is also the first line of defense for the next group in the lifecycle, namely the analytical group.

This shared frontier means that making data production and consumption observable, and thus transparent and controllable, constitutes a communication medium between two groups of people working hand in hand towards the same goals with different practices and, probably, tools. Therefore, making the serving stage data observable is four times as valuable as elsewhere:

For the data engineers about their own work.

For the consumers to be aware of produced data behavior

Fort the data engineers to understand better how data has changed and impacted their consumers.

For the consumers to provide explanations about the expected behavior.

In other words, a step towards the definition, update, and respect of SLO and SLAs with mutual acknowledgement of each other’s constraints and capabilities.

Recipes

As served data are traditionally exposed with queryable systems for example relational databases with SQL, those systems typically hide the complexity of the data transformation in their engine–and this process is not yet data observable (REF to chapter 3).

A solution could be to create agents to extend the capabilities of such databases to become data observable, by generating from the engine all lineages of the tables used in the queries and their destination, and even more interestingly generating metrics of those tables along the process.

Although this work would generate a ton of fun, it won’t be enough, unless the protocol between the application and the database is also made data observable,, we need to have information also about the applications working with this data, and this kind of information is not made available to the database. Again, this is an idea to explore using practices such as distributed tracing for example, but, the database would need to support that. We’re not there yet.

An alternative that will always work though is to consider that the SQL queries executed by databases are always provided by an application, see Fig. 6½, which must be, anyway, data observable. Therefore, an agent in this application would also have access to the SQLs thrown at the databases, and so, use them to generate the data observations that should have been generated by the databases.

Figure 4-9. SQLs are just character strings sent to an API by applications, always.

We will look at two types of applications for one type of database in the following recipes. The applications are:

Python applications creating the SQLs, and

Orchestrators capable of sending SQLs to databases (in a no or less code manner).

The chosen database is Google BigQuery.

Before jumping into the recipes, caution must be made on the cons of this practice that will generate the data observations on behalf of the database. As we need to generate metrics about the data (in and out), the applications will have to execute additional queries to the databases, which is going to increase the load of the database (and cost) and also introduce a slight latency between the time when the data is used in the query, and when it is “observed”³⁴.

Nevertheless, this trade-off is acceptable, considering that blindness would have a cost of another magnitude on the business (remember the 1-10-100 rule).

BigQuery in Python

In this section, I’ll describe how an application using the BigQuery python library provided by Google (REF) used in a python script can be extended to become data observable. This will serve as an example, as most python clients for other databases could be extended using the same strategies provided in this section. Also, the overall strategy to intercept the SQLs at the client level can easily be extended to other languages (e.g., JAVA JDBC driver).

The Google python library for BigQuery provides Client and QueryJob classes to define how to connect to the server and execute SQLs. There are two types of jobs that can be executed:

One-way: those jobs aren’t meant to retrieve data returned to the python script, but to update the data in BigQuery. Examples are: creating a new dataset, updating or deleting rows in a table, etc.

Roundtrip: those jobs are typically queries that the BigQuery servers will execute to prepare a dataset to be serialized back to the python’s script, which can in turn manipulate them (for example using pandas).

It is important to understand both types because, in the first case, for One-Way jobs, the lineage connecting inputs and output are fully included. However, for Roundtrip ones, they are initiating only the lineage as there is no output yet defined in them, the outputs of the lineage will depend on how the data will be used by the script afterwards.

To describe how the agent can be structured, we will separate what it will do, and how to do it in Python using delegation and monkey-patching.

Figure 4-10. Python script with BigQuery Client sending two types of Job

In Fig. 8, we see how the two types of jobs interact with Google BigQuery through its API to send SQLs. In gray are highlighted the components that should be made data observable and allow the python application to be data observable natively. Because they are not, I’m going to introduce the agent in the python application which will take care of it as discussed above, see Fig 9.

The agent, which will be described later in this section, is loaded into the python application and acts as an interceptor of the jobs sent to the BigQuery API. With this information, the agent can extract the lineage using an appropriate parser³⁵ or simply using the plan of the query³⁶ –which in fact is the closest to data observability functionality, as it returns the operations performed in a query and one metric, the number of data written (in the One-Way scenario). Over time, the plans will eventually evolve to return more data observations to complete the picture that must be made observable by the application, meanwhile, let’s fix it.

When the lineage includes the table references used in the process, this provides the opportunity to retrieve the metadata and compute metrics on them, using dedicated SQL queries executed against them by the agent.

Those two steps are shown in Fig. 9 as a label to the arrow going out of the agent and representing its work.

Figure 4-11. Fig.9 - Python script with Data Observability agent for BigQuery Jobs

You have now an idea of what the agent should be doing to make the python application using BigQuery data observable, the next step in this recipe is to provide strategies on how to implement it. As said above, this example based on BigQuery is not specifically dedicated to this database, but can be extended with abstraction to support other databases such as PostgreSQL, Oracle, My SQL or any relational database.

Early in this section,we have seen two classes, Client and QueryJob, we are going to revisit to make our python script data observable by implementing the extra steps we just covered. To support this work, here are some examples³⁷ of how BigQuery is used:

from google.cloud import bigquery
# Connect the client.
client = bigquery.Client()
# 1. Perform a query.
QUERY = (
    'SELECT name FROM `bigquery-public-data.usa_names.usa_1910_2013` '
    'WHERE state = "TX" '
    'LIMIT 100')
query_job = client.query(QUERY)  # API request
rows = query_job.result()  # Waits for query to finish
# Waits for query to finish
# And create a DataFrame with the data returned
df = query_job.to_dataframe()
# Store the data locally (this is bad)
df.to_csv("/tmp/iwillforgetaboutthis/important_data.csv")
# 2. Update a table.
table_id = "<destination id :: TODO>"
job_config = bigquery.QueryJobConfig(destination=table_id)
sql = """
    SELECT corpus
    FROM `bigquery-public-data.samples.shakespeare`
    GROUP BY corpus;
"""
# Start the query, passing in the extra configuration.
query_job = client.query(sql, job_config=job_config)  # Make an API request.
query_job.result()  # Wait for the job to complete.

Example 4-21. Code 7. BigQuery python scripts – examples

In Code 7., you can see that the client is the Client instance created first to connect with the underlying credentials the developer had to provide. Then this client is used to execute (query) a QueryJob which can be created from a simple QUERY string or a QueryJobConfig that contains the destination corresponding to where the result of the query must be stored, and therefore, not returned to the application runtime.

You are getting used to it, I won’t detail the agent completely as you can find it on the book’s GitHub repository, however, here is an explanation of the concepts that it forms.

First of all, we need to extend the client with tracing capabilities, as it will initiate computations with its methods query and load_from_uri for example (where the latter is loading a dataset from an URI into a table). In fact, each of those methods needs to create or retrieve an existing agent logger to track the data sources (uri, file, tables) and compute the observations.

To achieve this, we’ll use another strategy described in chapter 3, the delegation, which consists of overriding the Client class. Because we want to reduce the amount of changes for non data observable script, we will limit it to only changing the module path from google.cloud.bigquery to fodo.google.cloud.bigquery. It is very simple to execute, I simply need to create a folder structure that starts with food then recursively add google and biquery. In this last folder, I have to create the client.py where the delegation with data observability capabilities will be done on the Client class.

It is important to add __init__.py files in the folders created, but also add imports to those in the folder where delegations are done in order to load all other non-delegated classes to be also exposed in this prefixed module path. For example, in the fodo/google/bigquery/__init__.py file, because the client.py is overridden, it must contain the line shown in Code 8.

from google.cloud.bigquery import *

Example 4-23. Code 8. Load all content of a python module in a prefixed module

Now that the structure is prepared, the delegation can be done as follow:

Define the class Client by extending the google one.

Override the methods to be made observable, for example, query.

Retrieve the self instance of the delegated class (super), to allow the original method to be called without creating interferences with the delegated action (in order words, execute it as it would have been executed without our extensions).

Execute the original method to ensure the original result is computed.

From the method’s parameters, intercept the variables that can be used to compute the data observations such as the sql, the loaded data (uri), or from the QueryJobConfig.

Generate the data observations as explained before.

Return the result of the original method.

In the listing above, I return the original result as-is, however for the Roundrip case, this result can be used to retrieve the data in-memory (using pandas for example) to perform further transformations and actions. Therefore, this result must have data observability capabilities.

To explain this part, I will focus on the Client.query that returns a QueryJob and its result method that retrieves the data. So, the QueryJob instance is created by the Client.query method, hence, if we want to reuse the delegation pattern, I need to change its code using ast as for dbt or copy-pasting its code in our delegated Client³⁸.

However, there is a simpler way, using monkey-patching (introduced in chapter 3), to beef-up the method of the return QueryJob with data observability capabilities. This will consist of replacing the methods directly in the instance using the patcher function, see Code 9.

def patcher(o, f_name, f_patch):
    # 1. Keep a reference to original non-patched method
    orig_f = getattr(o, f_name)
    # 2. Define the patching function.
    def monkey(*args, **kwargs):
        # 3. Executing original non-patched code to get the result
        res = orig_f(*args, **kwargs)
        # 4. Patch the result
        res_patched = f_patch(res)
        # 5. Return the patched result
        return res_patched
    # 6. Inject the documentation of the original function (important for intellisense)
    monkey.__doc__ = orig_f.__doc__
    # 7. Reassign (patch) the original method with `monkey`
    setattr(o, f_name, monkey)
    # 8. Return the patched object
    return o

Example 4-25. Code 9. Python patcher function

The patcher function shown in Code 9. takes 3 parameters:

The object holding the method to be overwritten.

The function name to be overwritten.

The function code to overwrite with.

This function is used between the steps 4 and 5 of the delegation listing presented above, so that the returned QueryJob by query is data observable.

And finally, let’s see how patcher can be used to overwrite the to_dataframe method of QueryJob which returns the data as a pandas DataFrame. The returned DataFrame has to be data observable then too, which can be achieved with the same patterns as explained in this section–using for example the python module kensu which is provided with such data observable DataFrame wrapper.

In Code 10., we see an excerpt that implements this agent using the above techniques:

class Client(client.Client):
    def query(self, query, ...):
        gclient = super(Client, self)
        job = gclient.query(query)
        generate_data_observations(query, job)
        return patch_to_dataframe(job)
def patch_to_dataframe(job):
  def to_dataframe_with_do(df):
    # Check on the `kensu` package
  return patcher(job, "to_dataframe", to_dataframe_with_do)

Example 4-27. Code 10. Python BigQuery agent simplified

With this agent at hand, the changes required to make the whole Code 7. data observable consists in one single modification as shown in Fig. 10.

Figure 4-12. The change require in BigQuery python script to make it data observable

This seems to be quite some work to make a script data observable, however, it is setting the foundation for higher-level system to be data observable with less effort. The next section is going to demonstrate this when SQL is configured as part of an orchestrated pipeline.

Orchestrated SQL with Airflow

As seen with dbt, data tools’ trend is to simplify most of the data engineering tasks (e.g., DataOps). This includes the empowering of SQLs over coding scripts and the definition of pipelines where steps, such as executing a SQL query, are combined in a straightforward manner, using tools called orchestrators, such as Airflow.

Airflow provides a simple to use Domain Specific Language, or API, to create the DAG of data operations running as atomic tasks. Using such an orchestrator reduces considerably the boilerplate to create pipelines based composable tasks, operate them, track success and failures that can condition the remaining of the pipelines.

Such a task can be a SQL job in BigQuery that loads a dataset into a table or performs a query stored in a new table. The pipeline was defined with the following code which is a simplified form of what you can find in the book’s repository.

etl_bigquery_external_table_task = BigQueryCreateExternalTableOperator(
    task_id=f"bq_load_external_table_task",
    table_resource={
        "tableReference": {
            "projectId": PROJECT_ID,
            "datasetId": BIGQUERY_DATASET,
            "tableId": "loaded_dataset_external_table",
        },
        "externalDataConfiguration": {
            "autodetect": "True",
            "sourceFormat": "JSON",
            "sourceUris": [
                'gs://bucket/tripdata'],
        },
    },
)
CREATE_BQ_TBL_QUERY = (
    f"CREATE OR REPLACE TABLE dataset_table\
    PARTITION BY DATE(some_date) \
    AS \
    SELECT * FROM loaded_dataset_external_table;"
)
bq_create_partitioned_table_job = BigQueryInsertJobOperator(
    task_id=f"bq_create_dataset_partitioned_table_task",
    configuration={
        "query": {
            "query": CREATE_BQ_TBL_QUERY,
            "useLegacySql": False,
        }
    }
)

Example 4-29. Code 11. Airflow DAG operators for BigQuery

Code 11 shows how two BigQuery operators, namely the BigQueryCreateExternalTableOperator and BigQueryInsertJobOperator, can be combined to load a data set and create a new table from a SQL statement. The two operators define the steps that are executed in sequence to create a partitioned table as the final result served for analytical purposes (see next sections).

Each step must be made data observable, as it represents the execution of a lineage, taking some data as input to generate outputs that will be used by other steps. So that, the whole DAG is a pipeline that will be made data observable because each step will be data observable, and altogether, can be combined using the lineage, where data sources are stitching everything together.

The good thing is that each of those steps are nothing but python code executing BigQuery jobs using the Google Python library, and thus, if the agent explained in the previous section is added to the project, it is already data observable. Consequently, we don’t have much to do, because data observability came by-default, as the foundations have embraced good practices, which can be relied on by higher level layers and abstractions, such as an orchestrator–pretty much like a program using a tested library doesn’t have to test the library, but its usage, as long as the usage is appropriate, which in turn should be asserted by the library itself during its execution, leading back to the concept of expectations introduced in chapter 2.

Therefore, airflow can be made data observable by overloading its operator classes for BigQuery as presented in Fig 11. using the same strategy as per the python agent.

Figure 4-13. Using the BigQuery agent to automatically makes the Airflow operator data observable

Therefore the original airflow DAG descriptor can remain the same but only imports must be adapted to import the data observable ones from the agent library, similarly as Fig 10.

In Fig 12., I present examples of such modifications for BigQuery operators, but also for other type of operators such as Bash or Python. You’ll find the according agent code in the kensu-py library at this address: https://github.com/kensuio-oss/kensu-py/blob/5bcaa61aa25367204db6a2d09e41a615ca446a1c/kensu/airflow/providers/google/cloud/operators/bigquery.py#L24.

Figure 4-14. Changes to make a complete Airflow DAG data observable (bash, python, BigQuery)

Note

Note that Airflow has a mechanism via the classes SQLCheckOperator³⁹ and its variations extended to support BigQuery. Those components can be used to include extra steps in the Airflow DAG (which could be run at given intervals). Those could be used by data engineers to generate data observations (especially DataMetrics) and at the same time check expectations (e.g., rules). This is handy although it comes with limited capabilities that data observability offers, such as using the history of metrics to validate the current data state for example. Also it requires additional and repetitive work to be added to the DAG, including its maintenance, still disconnected (desynchronized) from the usage in the other components.

Analytics

Developing and using machine learning models is a key goal for all data organizations to make decision making processes more autonomous or to leverage data from multiple sources.

With business intelligence, business stakeholders are the most exposed and involved, expectations are high, and trust about the whole data program depends on the success or failure of business intelligence analytics results.

Analytics is where trust can be destroyed with the highest impact. As distrust spreads among stakeholders, the question of budget allocation for data activities may be placed on the table, as data teams receive budgets from the business to create data products for their domain, without trust, business stakeholders will less likely count on data teams to drive some of their goals.

Therefore, we better make sure that any data used for analytics is observable in order to allow fast reaction and prevention of issues coming from inappropriate data, data drift, behavioral changes. Analytics and data applications require data observability to expose an aspect of the information that will qualify their behavior, whilst other facets are brought by other areas, such as Application Observability.

On top of that, analytics and especially ML applications need to extend their observations to other data observability areas specific to their operations; namely, Analytics Observability. The purpose of this combination is to link the behavior of the data with the behavior of the machine learning models or business intelligence “queries” ones.

With this capability enabled in analytics applications, it is possible to detect if and when anomalies are happening in the analytics process, and rapidly identify whether the responsible is the data or the analysis (or both).

In the below sections, I will give some examples of how data observability agents can be leveraged to generate observations at the intersection of both areas, Data and Analytics–for different scenarios in machine learning and business intelligence.

Machine Learning Recipes

To illustrate how data and analytics observations can be generated altogether in a machine-learning environment, I will use a common python library called Scikit-Learn.

Python scikit-learn

Scikit-Learn is one of the first successful open source python libraries dedicated to the data science community. Because it is also in python and well coupled with pandas which we encountered already in the section about BigQuery, it will give us the opportunity to leverage mature techniques introduced before and available agents.

The typical process to train and use a model with scikit-learn is presented in Fig13.⁴⁰, with two pipelines, the training and prediction ones.

Figure 4-15. Machine learning process involving scikit-learn

In Fig. 13, the machine learning steps involving scikit-learn alone are defined marked in yellow and consist into:

Choosing and parametrizing the model: create a new instance of classifier class for example, like sklearn.svm.SVM.

Training the model: executing the fit method on the model.

Predicting: executing the predict method on data.

However, the pipeline has other steps that will play a role in the final results and must be data observable too, some of them rely on some functionality of scikit-learn, but the biggest parts will be done using dedicated libraries:

Loading (training) Data: this step can be assimilated to the ingestion phase in the data engineering lifecycle, and in the case of scikit-learn, would use a library such as pandas.

Prepare the data: the data preparation is the transformation phase in the data engineering lifecycle, and in fact, could have been handled also as a data engineering step to create a data mart. Nevertheless, it could also be simply another pandas manipulation in a python script to make sure the data is matching the model contract. Only even shaping the data (create the matrix of numbers with the right dimensions) using the numpy library. It is worth noting that the preparation of data can also involve some machine learning related tasks, such as scaling the data, which are available in scikit-learn, therefore the data observability agent has to work across technology.

Load and serve model: these crucial steps are similar to the serving phase of the data engineering lifecycle. They could be easily done using a library like joblib or pickle. Same as above, the storage and load of a model involves some functionality of scikit-learn, and thus interoperability must be also preserved here.

In Fig. 13., I have added the environments where the training and prediction pipelines would run⁴¹. Consequently, some key artifacts such as data sources, library versions, or the infrastructure may be different across the environments. But, some constraints must be respected across environments, such as:

Loading the data: the connection should be similar between each environment, or different scripts must be created (and aligned).

Prepare the data: the preparation step must be identical in production as in the lab, if the data sources are similar, otherwise the data must be prepared differently to align with the requirements of the predict method of the model.

Serve and load: the model must be stored and loaded with the same library to ensure compatibility at the binary level, but also with the rest of the pipeline.

As you can imagine, the whole system is somewhat fragile, especially if steps are decoupled from each other and assigned to several people or teams. Even a small change in data source, schema, content, parameter, versions, or anything else may generate wrong predictions.

Because most steps, such as loading or preparing the data are related to ingestion and transformation stages of the data engineering lifecycle, I will consider them as covered. So that, I can concentrate on the characteristics specific to where scikit-learn is involved, the model training and predict steps.

Without getting too deep into analytics observability (cf end of the section), what must be done for scikit-learn to be made observable is to track the data used to train a model (incl. the location, schema, metrics, etc) and link it to the fitted model stored and served for predictions. Indeed, a model stored and served is therefore considered as anything but a data source, with a schema, and even metrics. In fact, a model is data, a linear regression model for example is rebuilt by reading the fitted weights, and, thus, additional metrics could be computed on those weights, such as the number of zeros, the min and max, etc.

Associated with models, we could think of generating observations about the model behavior, or performance, this is done with estimators (e.g., r2), although we can technically do it,, I don’t consider them as data source metrics. Such information is not intrinsic to the model, they reflect. They reflect the model performance related to the data during the training phase, hence they are more specific to machine learning, and therefore should be associated with the trained model using an analytics extension to the observability core model introduced in chapter 2.

Then comes the prediction phase, we must make observable the generation of those predictions, by linking the model and the observations (the data) to them.

To accomplish this, because scikit-learn is also a python library, we can use the overload pattern used before and overload, for example, the LinearRegression class, as proposed in the following (pseudo)code excerpt:

import sklearn.linear_model as sk_lm
class LinearRegression(sk_lm.LinearRegression):
    def fit(self, X, y, sample_weight=None):
        # 1. call the `sk_lm.LinearRegression` first to compute the fitted model
        model = super(LinearRegression, self).fit(X, y, sample_weight)
        # 2. Compute for the `model`.
        # E.g., the number of `parameters`, the min and max of the returned `weight`, etc
        metrics_for(model)
        # 3. Link the `model` to the input data `X` and `y`
        lineage_for(_in=[X, y], _out=[model])
        # 4. Include analytics observations
        # E.g., fetch hyper parameters, execute estimators, ...
        # 5. wrap `model` to ensure its usages will also be observed, and return
        return do_wrap(model)
    def predict(self, X):
        # 1. Compute the predictions first
        predictions = super(LinearRegression, self).predict(X)
        # 2. Wrap the `predictions` to make it data observable
        do_wrap(predictions)
        # 3. link `predictions` to the `model` and the data `X`
        lineage_for(_in=[self, X], _out=[predictions])
        return predictions

Example 4-31. Code 11. Scikit-learn’s LinearRegression extension for Data Observability

In Code 11, you can see the key steps to follow will turn any LinearRegression model training and predictions into a fully data observable applications. Because we can rely on the other libraries which have followed (or have an agent for it) the data observability principles, the amount of work is relatively small, but also it is straightforward as it consists mainly of linking the inputs to the outputs.

With this agent added to your data science environment, the code changes in your analytics scripts are limited to the following as they enable the agent which will collect all observations defined by default (see the code repository for a fully working example):

# Prefixing the module package with our agent name "doagent" for example
from doagent.sklearn.linear_model import LinearRegression
regressor = LinearRegression()
regressor.fit(x_train, y_train)

Example 4-33. Code 11. Example of scikit-learn of a data observable LinearRegression usage

As said before, we also have the opportunity to introduce analytics observations, and by joining those capabilities from data and analytics observability, it becomes possible to detect and analyze across the analytical parts and engineer ones when a model has biases or is not anymore appropriate for the population it is supposed to predict from and about.

Business Intelligence Recipes

Business intelligence, especially considering tools such as reports and dashboards, is probably the last line of defense to solve and prevent issues from happening, as it is directly related to business decisions taken by stakeholders. Therefore, considering the skills of such persona, their power, and their availability, it is crucial to demonstrate a high level of maturity and accountability when delivering, as any issue or lack of confidence can generate a storm of consequences for the whole data organization.

In this section, I have chosen to demonstrate how data observability can be introduced in two kinds of applications that are likely to be used by decision-makers. First, a notebook technology that can be deployed in production with presentation capabilities, as this has become a practice over the years (yet controversial). For this, I’ll use Databricks which is one of the main actors who could make this practice a reality and easy to apply.

Then I’ll follow up with a more traditional data tool, a dashboarding software used to present business facts in a single view connected to data sources, generally the data warehouse or data marts. For this, I’ll use an open source product called Superset.

Databricks

Databricks is a company built around the distributed computing library Apache Spark we covered earlier in this chapter. However, over the years, the company has developed notebook and SQL capabilities (among others, such as DataOps) to support better business intelligence requirements.

In this section, I will focus on the presentation aspects, which gives the possibility to users to define analytics⁴² and present them as a sort of report. However, notebooks are also supporting the storage of data or work on machine learning models, but this part was already covered in earlier sections.

Also, Databricks can be considered as a low code platform, where queries can be added directly to notebooks and the results presented right away. Therefore, it is massively simplifying the work to make this possible. And the machinery behind the platform is obviously hidden from the outside world. Hence adding agents directly at that level would not be allowed⁴³. Nevertheless, we can still do it at a higher level where Databricks provides us access: for example the cluster running the analyses defined in the notebooks.

But, before explaining what this integration would look like, I’ll go through what we would like to make data observable in a Databricks notebook. A notebook for reporting will likely execute those tasks:

Connect to data sources.

Perform analytics.

Present the results as tables or charts.

So, there is no real output being stored, but they are visually served to the final users. Consequently, to conceptualize the lineage, we need output data sources to represent the tables and charts. This is why, for reporting notebook, we have a kind of split-brain between Data Source and Application, because such a notebook is both at the same time:

Application: because it is a script that will run to produce results.

Data source(s): each table or chart is a visual storage for the underlying data.

Moreover, we can connect the user receiving the information, as we can create a user entity for them that will be linked to the application execution entity created from the fact that the notebook needs to execute to present the information. So that we can fully observe how a report is being consumed and intercept signals from that.

The Fig. 14 describes this scenario, which will serve us to understand how the agent can be integrated and what it generates.

Figure 4-16. Databricks notebook’s usage flow

In Fig. 14, you can see that the notebook relies on the Spark cluster that it is connected to, therefore as long as this environment is data observable we’re almost the whole scenario is already covered, as it may only remain these three peculiarities coming from the Databricks notebook:

Capturing the user.

Capturing the notebook name or path to qualify the name of the application.

Capturing the Databricks visual components as data sources.

So, we’re quite lucky as Spark was covered in this chapter already, for which there is a reference implementation. So, only the three above points need to be connected to the data observations already collected by the Spark agent.

For the two first pieces of information, we have to use some internals of Databricks, the class com.databricks.dbutils_v1.DBUtilsHolder, which comes with those fields⁴⁴:

userName: the connected user name,

notebookPath: the path to the current notebook.

Then, for the visuals, there are several ways to create them, a common one is to use the display function. The display function runs Spark jobs to organize and collect the data to be presented in the frontend. Here, again, most of the work is done by the installed Spark agent as it will generate data observations about the collected data; that is, when one of the Spark functions collect, take, or others are used. Consequently, we could create a Data Source for the notebookPath and attach it to the schema of each data collected, alongside with the statistics returned. Proceeding the following, the data observations will look like this:

Figure 4-17. Excerpt of the observations generated by Databricks agent⁴⁵

Therefore, turning any Databricks notebook always to produce the observations presented in Fig 15., it is simply a matter of dropping the agents (jar) into the classpath of the Spark driver (cf. Fig. 148) using the Databricks cluster configuration page.

Superset

Superset is an open source reporting and dashboarding solution that supports many types of data sources. Superset is implemented in python and can be installed from docker, or also from scratch (the source code).

Superset offers a low-no-code interface for analysts to query available datasets using SQL directly, or via the wizards that simplify the exploration of the dataset to generate the table or chart visualizations. Actually, it is quite close to a combination of writing a query in a Databricks notebook, then using the visualization to manipulate the results further with display. The table and charts can be added to dashboards to have all important information in one glance.

It is therefore easy for the user to access and define, dashboard using the wizard, resulting in a lot of information to be observed. Superset must therefore be made data observable with a high degree of automation, as it is not conceivable that the users have to go through many steps for each information they generate so easily.

Making dashboards data observable has the potential to solve a common challenge coming with reporting tools, their management, especially their reliability of course, but also simply their reasons to be maintained–as it is easy to create a dashboard at a given point in time to analyze a situation, but without the goal to make it perpetual (e.g., tightly coupled with the time of creation), and forget to delete it. In fact, the deletion is not necessarily forgotten, it is often the case where we keep them as they are like a good starting point for a potential next analysis. However, it makes it harder then to understand for the data team what are the impacts that need to be taken care of during a change process (e.g., name change).

So that, adding the data observable capability to Superset will, among other things, make transparent how the dashboards are created, how they are run, how often, and, with data metrics in place, their reliability.

It turns out that enabling this capability is really straightforward and requires little changes, yet will cover a wide range of cases. This is because a chart in Superset is always accessing the data using SQL queries and returning the results as pandas.DataFrames. Therefore, as we have already discussed those in the Transformation phase of data engineering lifecycle, the agent can leverage the existing agents in a modular fashion then focus on the specifics of Superset.

In the figure below, you can see where an agent has been added to enable by default data observability for charts and what it would look like:

Figure 4-18. Superset chart refresh flow

In Fig. 16, you can see that most of the process touching data is actually already covered, consider for example the database to be BigQuery, we can use the agent introduced in Serving to make this part data observable. Then, the command to get the data wrap the result into a DataFrame, also covered in this chapter, however, like for Databricks, the data is not written on a storage, but transformed into JSON and returned in-memory back to the browser for presentation.

Therefore, the agent specifics for Superset is essentially to intercept in the file get_data_command.py when the SQL is executed and the result retrieved, then enrich the data observations generated by the SQL and Pandas agent with the information about the dashboard and the chart that will present the data.

This will be accomplished by intercepting the two following objects:

The QueryContext: which contains the information about the dashboard and the chart, and

The result of the execution of its method get_payload which contains the resulting DataFrame.

The agent has this information if it is embedded in the run method of the ChartDataCommand class defined in get_data_command.py. As, this class has a field _query_context and the run method looks roughly like this:

try:
    payload = self._query_context.get_payload(
        cache_query_context=cache_query_context, force_cached=force_cached
    )
except CacheLoadError as ex:
    raise ChartDataCacheLoadError(ex.message) from ex

Example 4-35. Code 12. Superset ChartDataCommand.run logic

As Code 12 shows, we have all information we need to create all data observations, and so we can simply add the following lines after payload is returned:

    from superset_agent import generate_data_observations
    generate_data_observations(self, payload)

Example 4-37. Code 13. Superset’s agent interception

To do so, we can reuse what was done for dbt and inject the AST representation of calling generate_data_observations. The generation of data observations, as said above, relies on existing data observable capabilities from underlying modules, but it does one special thing in order to represent the output Data Source, which is the chart, here is the excerpt from the code available on the book’s repository:

qc = chart_data_command._query_context
from superset.charts.dao import ChartDAO
chart_id = qc.form_data["slice_id"]
chart_slice = ChartDAO.find_by_id(chart_id)
chart_name = str(chart_slice) # returns the name of the slice
from superset.dashboards.dao import DashboardDAO
dashboard_id = qc.form_data["dashboardId"]
dashboard = DashboardDAO.find_by_id(dashboard_id)
dashboard_dashboard_title = dashboard.dashboard_title

name = f"""Superset:{dashboard_dashboard_title}:{chart_name}"""
_format = "Superset:Chart"
location = f"""/superset/explore/?form_data={{"slice_id":{chart_id}}}"""

Example 4-39. Code 14. Create Data Source entity observations for Superset Chart

In Code 14, you see that we can extract the dashboard and chart ids, which by reusing the data access objects (DAO) for both classes, gives the agent the opportunity to retrieve their name and title. So that, a Data Source can be generated with a name containing the dashboard and chart, quite similar to what was achieved for Databricks’ notebooks.

Note that, the location of the chart can be precisely generated as well, that is, generating the data observation that uniquely identifies where the chart is available and can be accessed.

Installing this agent into an existing Superset environment ensures that the link between a chart and the datasets can be observed, including when the lineages are executed and who. This is addressing the challenge mentioned above, but also because both agents for SQL and pandas are generating data metrics, the reliability of the charts is also made observable.

¹ https://learning.oreilly.com/library/view/fundamentals-of-data/9781098108298/ch01.html#the_data_engineering_lifecycle-id000035

² https://andy-petrella.medium.com/datastrophes-il-buono-il-brutto-il-cattivo-c0bad03e8dcb

³ https://research.google/pubs/pub49953/

⁴ Also used by Airbyte to perform CDC: https://docs.airbyte.com/understanding-airbyte/high-level-view

⁵ https://www.singer.io/

⁶ https://docs.airbyte.com/understanding-airbyte/high-level-view

⁷ https://docs.airbyte.com/api-documentation

⁸ https://github.com/airbytehq/airbyte-platform/blob/main/airbyte-commons-worker/src/main/java/io/airbyte/workers/internal/book_keeping/MessageTracker.java

⁹ https://github.com/Fundamentals-of-Data-Observability/oreilly-fodo-source-code/tree/main/ch4/s1-ingestion/r1-airbyte-agent

¹⁰ https://github.com/Fundamentals-of-Data-Observability/oreilly-fodo-source-code/blob/main/ch4/s1-ingestion/r1-airbyte-agent/airbyte-instrumentation/src/main/java/io/kensu/agent/airbyte/asm/instrumentation/Premain.java

¹¹ https://trends.google.com/trends/explore?date=all&q=%2Fm%2F0bs2j8q,%2Fm%2F0ndhxqz,%2Fm%2F0fdjtq,pyspark,python%20pandas

¹² https://www.tensorflow.org/guide/keras/functional

¹³ https://en.wikipedia.org/wiki/Dataflow_programming

¹⁴ https://en.wikipedia.org/wiki/Prograph

¹⁵ https://absaoss.github.io/spline/

¹⁶ https://github.com/apache/spark/blob/v3.3.0/sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala#L52

¹⁷ https://github.com/apache/spark/blob/869fc2198a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala#L36

¹⁸ https://github.com/AbsaOSS/spline-spark-agent/blob/release/0.7.11/core/src/main/scala/za/co/absa/spline/harvester/dispatcher/LineageDispatcher.scala

¹⁹ Even so, the operation executed could be extremely interesting as an extension of the core model.

²⁰ https://github.com/AbsaOSS/spline-spark-agent/blob/develop/core/src/main/scala/za/co/absa/spline/harvester/postprocessing/AbstractPostProcessingFilter.scala

²¹ https://docs.getdbt.com/reference/artifacts/dbt-artifacts

²² https://docs.getdbt.com/reference/artifacts/manifest-json

²³ https://docs.getdbt.com/reference/artifacts/catalog-json

²⁴ https://docs.getdbt.com/reference/artifacts/run-results-json

²⁵ https://github.com/datahub-project/datahub/blob/master/metadata-ingestion/src/datahub/ingestion/source/dbt.py

²⁶ https://github.com/dbt-labs/dbt-core

²⁷ It can also represent a Snapshot, which generates information about the changes over time. But I didn’t cover this part, as it is a practice to control what happens on data, not to build data.

²⁸ https://github.com/dbt-labs/dbt-core/blob/6c76137da4e72bbcad6e7789a78c31f1bf5cffac/core/dbt/task/run.py#L248

²⁹ Well, because python is not compiled… we could, but, that would be another level of nastiness

³⁰ https://docs.python.org/3/library/site.html

³¹ Or “usercustomize” at a lower level.

³² More information about these low level mechanism of python can be found in the New Import Hooks specification: https://peps.python.org/pep-0302/

³³ https://docs.getdbt.com/guides/advanced/adapter-development/1-what-are-adapters

³⁴ The quotes are there because the metrics observed here are not fitting exactly the definition in chapter 1, because additional queries are hitting the database to compute metrics. Therefore, it is intrusive to the experiment.

³⁵ For example: https://github.com/reata/sqllineage

³⁶ https://cloud.google.com/bigquery/docs/query-plan-explanation

³⁷ https://cloud.google.com/bigquery/docs/samples

³⁸ Nooooooooooooooooo

³⁹ https://airflow.apache.org/docs/apache-airflow/1.10.13/_api/airflow/operators/sql/index.html#airflow.operators.sql.SQLCheckOperator

⁴⁰ Although, this process is quite similar for any other machine learning library

⁴¹ When online training is applied, the diagram would need to be slightly more complicated, however, there is still work being prepared in the lab before moving to production. The minimal step would be, at least, to validate how models are going to be built. Even if we use an autoML, there is still some work to be validated in the lab.

⁴² but not limited to analytics, in this specific case, as this is powered by Apache Spark.

⁴³ At the time of writing, because it is likely that Databricks will make their system also data observable, given that they are also the same people who introduced observe in the Spark library.

⁴⁴ An example is available in the Kensu Spark agent https://github.com/kensuio/kensu-spark-collector/blob/3fe300470b808c8b3e0710917dfbac07cd11c1a8/src/main/scala/io/kensu/sparkcollector/environments/DatabricksEnvironmentProvider.scala

⁴⁵ And that’s why, it can’t reasonably be done manually…

Chapter 5. Roles of Data Observability in a Data Organization

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 5th 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 author at gobrien@oreilly.com.

In the previous chapter, you have learned the definition of what data observability is and how data technologies and data teams can embrace it. In this chapter, I will look at data observability in a systematic manner to analyze how it fits into the data organization, such as its data architecture and culture. Because the culture of data is a complex system itself, I will address it in two parts;

Through the lens of the roles of data engineering in a data organization.

The role of data in the evolution of our economy (and, thus, organizations).

Data Architecture

Data architecture is a critical component of a data organization. It sets the foundation of how data will deliver value for each department individually and collaboratively.

The intent of this section is not to define the ideal data architecture, if there is such a thing, but to review where data observability must be introduced to support the sustainable management of data usage in the organization.

It is worth noting that not all organizations are equal regarding data architecture; some already have an established architecture expected to evolve or be replaced partially or entirely, and others are simply starting from a white page. Therefore, I will explain why you must consider data observability from the start when building a new architecture and how, despite its central role, it can be introduced as an add-on to existing data architecture.

Where does Data Observability fit in a Data Architecture

To address where data observability fits into a data architecture, I have to split data observability into two pieces: the Data Observable System and the Data Observability Platform. This is to respect a fundamental principle of architecture, the separation of concerns.

The Data Observable System

The data observable system relates to the involvement of data observability as an element of data applications. As we recall from the first chapters, data observability is a capability that can be introduced from within data applications which can generate the whole data observability core model.

Therefore, data observability is included at first in the application layer of the data architecture, as presented in Figure 5-1.

Figure 5-1. Data Architecture including data observable applications

Two different cases must be considered in this layer: when data tools, frameworks, and libraries are partially or not (yet) data observable and when they are natively data observable.

When the components supporting the creation of data applications are not fully data observable, an additional component must be added in one or two places depending on the tools and architecture. These places are represented by the labels 1 and 2 in Figure 5-1. The label 1 on the App represents the introduction of the data observable capability I have discussed in Chapter 3. The label 2 is where tools are extended with additional artifacts (e.g., agent) to generate data observations by default for all data applications built with them, as discussed in Chapter 4.

In the future, however, all tools and libraries will be data observable by design or natively, which means all data applications will be data observable by default, so data observability will be taken for granted. Therefore, we don’t represent what is trivial to avoid bloating the architecture diagram and accelerate the definition of new architectures.

In the meantime, however, I have to consider the tools that are either legacy or too closed (black box) to give us a chance to apply the practices presented in Chapters 3 and 4. For those, I reserve a place in the data observable system, which has one foot in the data layer and another in the data observability platform. Chapter 6 covers this specific, yet temporary, case in detail.

The Data Observability Platform

In the previous section, Figure 5-1 shows that all applications are made data observable either natively or with appropriate add-ons. Therefore, data observations are generated continuously by a distributed interconnected system of data applications describing their behavior with data and conversely.

The data observations generated by each application are yet to be used to address the challenges introduced in Chapter 1 that require a system to detect data issues in a scaling environment and smoothen the process to find and resolve the root cause to anticipate the issues. Moreover, each application is most likely strongly (e.g., orchestration) or weakly (e.g., using the same data sources) dependent on others, such that there is huge potential value in the interplay of global data observations. For example, by respecting the data observability core model, it is possible to leverage the information about these dependencies (e.g., lineage) to automate the detection and resolution of data issues.

You also need to consider the value that exists in past data observations because the current status of data in its usage context can be determined more efficiently based on past statuses. A simple example is a data ingestion application. The amount of data ingested most likely follows underlying patterns (e.g., timeseries) that can be learned to determine if the amount of current data ingested is an anomaly or not–cf the section in Chapter 2 about expectations.

Consequently, data architectures must be extended with an additional central component, the Data Observability Platform. At the very minimum, this component is responsible for ingesting all data observations, aggregating them, and ideally learning from them to offer the above capabilities. The Figure 5-2 updates Figure 5-1 with this platform next to other central components.

Figure 5-2. Data Architecture including the data observability platform

The data observability platform does not operate in isolation but requires collaboration with other platforms like the orchestrator or application observability platform. This collaboration is necessary because, as outlined in Chapter 1, data observability is a dimension of an observable system. Data observability serves as a buffer that provides control and visibility between data governance and all other data management domains.

Data observability also plays an important role in the efficiency of a data (engineering) team, which I’ll discuss in a later section. First, let’s look at when a data architecture must ensure the system will be data observable and how to introduce the data observability platform.

Data architecture with data observability: when and how

In an ideal world, you would introduce data observability from the beginning. In the real world, however, delivering the platform and the first value always comes first. Monitoring and reducing the total cost of ownership (TCO) may become the next priority if time and capacity allow to “not work on direct business value generation.” .

I have seen a lot of companies willing to drive a “Proof of Concept” on data observability, but, in fact, do we really have to prove that the concept of generating data observations to increase the information about a system is good? Okay, I understand that the data observability platform, the tool I mean, is necessary to be “tested” to validate its capabilities match the expectations and needs. However, making the system data observable shouldn’t even be questioned. It must be embraced by both data architecture and the organization’s whole data community. So, I recommend starting “yesterday” with the data observable part and looking for the right data observability platform that will adapt well to your specific environment.

For now, let me cover two broad categories, the new and existing architectures.

Not all cases will fit those categories perfectly; some cases land somewhere in between. Thus, understanding both categories provides a good sense of which mix may work best for you.

The introduction of data observability in an architecture depends on the specific design and implementation of the system. In the case of a new architecture, it is advisable to select data technologies that are either already data observable or can easily be made observable. It is also recommended to introduce the data observability platform in the foundational layer of the architecture, where other solutions such as the orchestrator are already available.

On the other hand, for established architectures, the introduction of the data observability platform in the foundational layer should not be hindered as it can work independently. However, this does not automatically make the system built on the architecture data observable. To achieve data observability, it is essential to convert existing data technologies into observable ones, unless they are already observable. For inherited technologies, a dedicated chapter may provide assistance.

When adding additional technologies to the architecture, it is important to choose those that are designed to be data observable. This will enable the system to maintain a consistent level of data observability throughout its development and operation. It is crucial to prioritize data observability in the design and implementation of any architecture to ensure that data can be effectively monitored, managed, and utilized.

How data observability helps with Data Engineering Undercurrents

Now that the data architecture is offering data observability capabilities to build applications. In this section, I will focus on the data engineering undercurrents. The undercurrents are often viewed as complex and inconvenient, but they are much simpler to deploy on a foundation of data observability.

Let’s start with security.

Security

Security must be embedded in the organization’s culture, which means it cannot be complex or hard to do. Although it is hard to automate all security tasks, systems are evolving to handle many of them. In software development, for example, a package with security issues can be detected, and pull requests can be created automatically by a dedicated platform.

The hard part for the engineers is that they must consider as many disastrous scenarios as possible, anticipate all observations that will be useful to detect their occurrences and automate prevention (e.g., filtering) and propagation (e.g., circuit breakers).

However, by combining the data observable system with the data observability platform, we can make things easier by automating the tedious and time-consuming tasks and leaving only the knowledge-based ones (e.g., computing the percentage of masked data) to the engineers. These tasks are directly linked to the value they generate and maintain.

Hence, the standard information, such as the observations related to data access, is provided by design by data observable systems in place. The data observability platform does the rest to match those observations with the security observability and automation components of the platform, as depicted in Figure 5-3.

Figure 5-3. Data and Security Observability platforms work hand-in-hand.

Other observations will also support engineers taking care of security, such as from the user and application observability areas that deliver observations. To illustrate this, let me walk you through some examples.

The first use case is to detect new usages or changes in the usage as a data application suddenly fetches 50% more data than in the past, although the size of the data itself has not changed. Also, it is possible to prevent data from being exposed outside because the encryption step is not applied in the pipeline, as shown in Figure 5-4.

Figure 5-4. Detecting lack of encryption

Thus, engineers can be notified when their custom metric about the percentage of masked data is going down, which may indicate data have leaked.

A last example is to alert that data flagged as important is not backed up, as there is no “backing-up” application reading and writing it elsewhere running on it, as represented in Figure 5-5.

Figure 5-5. Ooops forgot to back-up

So, now that we can support the engineers to achieve a satisfactory level of security without dramatically increasing the intrusiveness in their daily work, let’s tackle an important and wide undercurrent–data management.

Data Management

In Chapter 1, I have proposed that the DMBOK2 data management framework can introduce data observability as a new area surrounding data governance as a control layer across all others (REF to the figure). In this section, I will take the opportunity to reinforce this message mainly through the lens of data governance as it is proposed in the Fundamentals of Data Engineering book.

Data Governance

To fully understand the impact and value of data observability in data governance, it is important to examine its intersections with other areas of data management that are governed by data governance. By analyzing these intersections, we can gain insight into how data observability plays a crucial role in completing the picture of effective data culture implementation at scale.

In other words, to fully comprehend the benefits and significance of data observability in data governance, it is essential to explore how it relates to other aspects of data management that fall under the purview of data governance. By doing so, we can gain a more comprehensive understanding of how data observability contributes to the successful implementation of a data-driven culture on a larger scale.

Discoverability

Data has to be reused across projects and domains, for this, it needs to be discovered during the analysis phase of a new project scoping, for example. This means that the characteristics of the data are exposed to an engine that will allow a human or another system to match it according to their needs. If this is not the case, therefore, all data needs to be analyzed systematically to find the most appropriate one, eventually, which would require ever-growing time and resources. In short, it will very quickly become unsustainable.

Another word for characteristics is metadata. But I think it is more intuitive to say that a user searches for data based on its characteristics than its metadata. However, metadata is the generally accepted term to describe the information presented below, so I will use the term metadata.

The metadata to be exposed can be split across the following categories, each category addressing a specific type of search and associated decisions. :

Business Metadata

Project, usage, (data) definition

Tech Metadata

Descriptive Metadata - location, lineage, schema, pipeline

Content Metadata - volumes, missing data, density

Ops Metadata

Runtime identifier, dependencies (orchestration)

Reference Metadata

Lookup data, codes, …

The metadata introduced above is pretty similar to most of the observation entities presented in Chapter 2 or can be derived from them.

Therefore, a data observable system can also be described as a discoverable system, as the observations about the involved data will be automatically generated and consolidated in the data observability platform.

This is the role then of the data observability platform to be able to cooperate with the data management component which is generally used for data discovery, the data catalog. This cooperation will enable the data catalog to always return to their users the most appropriate data based for the searches or eventually push the information to users showing interest to be notified when data respecting certain characteristics are available.

Accountability

Because they use or generate the whole or a portion of the data, engineers hold a certain accountability for these specific operations. For example, suppose some incoming data used in the process is not satisfying engineers’ assumptions. In that case, the engineers will be considered responsible for either not noticing it, or worse, having generated inappropriate data (e.g., GIGO).

So, to be confident in this new responsibility, engineers must build practices to introduce defensive approaches (introduced in Chapter 2 and described in the following chapters) to increase their chances of success without impacting their capacity to deliver. I They have to make the data systems they build and use data observable. Then, they have to couple the generated observations with:

Assumptions checking (±SLO)

Expectations handling (±SLA)

The last option, which is a fallback, a protection net if you like, is to rely on the data observability platform to create visibility about latent constraints that might be interesting for the engineers to monitor continuously.

Data Quality

The role of a data engineer is often challenging when it comes to ensuring data quality. This is because the engineer is providing a service in response to a business request for new or adapted data, which typically comes from a subject matter expert (SME). The request is often triggered by a range of factors, such as exploration of reports, analysis of the market, specific knowledge about the domain, or awareness of peculiar events that may impact the business.

However, it can be challenging for the SME to explain these factors to the engineer, who may not have the same level of domain-specific knowledge. As a result, validating the final result can be difficult for both the engineer and the SME, as the amount of results could exceed their individual areas of knowledge.

Moreover, even if the SME is able to express how the final result can be validated by outlining the constraints or expectations they have, it is not always possible to ensure exhaustiveness. Additionally, the validation process of the results may not hold up in the long run or in a changing context, which is another important consideration.

Then comes the fact that the produced data is reused, either by the original requester or other users, potentially from other domains, who find the data valuable for their own use case.

On paper it would be possible for engineers and stakeholders to define a set of constraints to validate the quality of the data. In practice, this results in an uncomfortable series of discussions where none of the parties can really take a firm decision and, thus, assume full responsibility. For the stakeholders, it is complex and risky to be assigned the responsibility to provide the “a priori” definition of what is “good data” now and in the future. Engineers often experience anxiety when deploying a project in production without being certain that all the transformations they have applied are producing valid data.

In reality, a continuous improvement process is essential to achieving agreement between all parties involved. Each party should have a well-defined role in this process. It is important to acknowledge that data issues are inevitable and that there is a limitless number of potential issues that could arise. By recognizing this and implementing a continuous improvement process, everyone can work together to ensure that data issues are identified and resolved in a collaborative and constructive manner.

Therefore, the time and resources involved in anticipation of data issues must be well balanced to provide a sufficient (or “good enough”) agreement of what will be validated and the iteration necessary to upgrade this validation process.

We can now revisit the roles and responsibilities to achieve an efficient collaboration between engineers and stakeholders:

Engineers validate the known constraints.

Stakeholders describe known constraints.

Engineers generate observations to the data observability platform.

Stakeholders use the data observations to maintain or create constraints–however, the platform should be smart enough to automate this.

Engineers and stakeholders should work hand-in-hand when new types of issues are discovered and need to be anticipated in the future.

This separation of concerns allows for an iterative, less time-consuming process, setting the right level of expectations and risks on both sides. In other words, this is the only way to be in a win-win situation.

Master Data Management & Data Modeling

It takes time and resources to define the entities related to an organization’s business or internal processes. Those efforts aim at formulating standards of formats, structures, and characteristics that must be respected across data teams and projects.

Data observability helps identify the cases where those definitions are not followed appropriately. To provide an example, let’s say that a data pipeline is responsible for writing date information to a data source. It is crucial that the format of the date information is consistent across all entries. In order to ensure this consistency, a data observability metric can be implemented to track the number of entries that do not match the required date format.

This metric can be checked by a rule (expectation), and the pipeline is considered to be in failure if the count of unmatched entries is not zero. By monitoring this data observability metric, data engineers can proactively identify and resolve any issues related to inconsistent date formats, ensuring high data quality and minimizing the risk of downstream issues.

Again, data observability acts as an important control layer to ensure that practices and conventions are respected, sustained, encouraged and rewarded in the organization.

Data Integration & Interoperability

With the explosion of specialized SaaS applications (REF), each having its specificities and associated personas, it is not uncommon for organizations to have several tools with a relatively significant overlap. Plus, there is an established awareness that combining data across domains is expected to yield surprising results, such as unknown niches or underused opportunities.

Consequently, most tools offer capabilities to share their data, integrate external data, or both. This can be more easily achieved using either programming interfaces (e.g., python, ETL) or higher-level tools (e.g., a low code, no code ELT such as Matillion).

Hence, the number of connections between the tools themselves or via the data platform is increasing as fast as the complexity of setting them up is decreasing. This is where data observability brings value to controlling the network of interconnections and ensuring the prevention of issue propagation.

Data Lifecycle Management

This is a vast topic. I’m limiting myself to only the deletion and archiving of data, for example to cope with security or privacy requirements. In those cases, data observability provides an embedded capability to expose if the delete operations are executed and their effects (how much data is deleted) or the archives created (how much, often, and when).

The other way around is also a valid usage of data observability as it provides the opportunity to expose data (or a subset) rarely used that can be deleted or archived.

Ethics & Privacy

With the growing awareness of what data can achieve (good or bad), an organization also needs to have a defensive approach against inappropriate usage or sharing of the data. This is something that must be implemented as part of the organization’s data culture because most problems are unintended and happen by mistake. Data observability provides, first and foremost, visibility, which in the case of ethics and privacy is key as the organization must be as transparent as possible.

Data observability provides great support for ethics and privacy, as new usages and shares of data are logged, including with whom the data was shared. Note that data observability is not a complete security mechanism. In fact, if the associated checks can be executed by the application responsible to share the data (e.g., using computational sharing policies), then sharing can be avoided, however, it is not data observability, but genuine applied security. Nevertheless, data observability allows inappropriate usages of data to be exposed, so prevention actions can be taken to avoid the same situation in the future.

Another case is an automated decision process that may be biased because the data used to define the process was biased. This happens for different reasons, the most common being that the data was skewed, which means that some categories were overrepresented compared to others. This situation can also be supported by data observability, for example, by controlling the variations of the skewness metric or simply the histogram of a categorical field.

DataOps

DataOps has not necessarily been a well-defined perimeter, however, in Fundamentals of Software Engineering, Joe and Matt did a good job describing the areas that will likely be constant across definitions or understanding: automation, observability, and incident response.

I am connecting these areas with a set of practices and tools designed to achieve multiple goals for data engineers. These include:

Ensuring fast and efficient deployment to production, while maintaining reliability and accuracy of the outcome.

Facilitating ease of debugging in case of issues or errors.

Providing a spectrum of automated issue detection to catch and resolve problems as quickly as possible.

By implementing these practices and tools, data engineers can work with greater efficiency, confidence, and peace of mind. Let’s see where data observability fits in this picture, starting with automation.

Automation

Scaling is all about scaling data ambitions. And, to scale smoothly and sustainably:

Complexity must decrease – to find more talents.

Frictions must be reduced – to go faster in production.

Repetitive tasks must be automated – to gain time.

Therefore, all activities related to creating and maintaining trust in this highly automated system must be automated themselves. Otherwise, they become the bottleneck and will simply be skipped when teams are under time pressure. That is why data applications must be made data observable by design or extended that way, as proposed in Chapter 4.

Observability

This section pretty much closes the loop, as data observability is an area of observability. Nevertheless, as mentioned above, observability must be a best practice that can be automated as much as possible, ideally by design across all steps and tools of the data transformation process.

Observability also brings value to issue management, as discussed below, and is responsible for generating the information needed to optimize the whole process. As introduced in Chapter 2, data observations are the data that bring visibility about the data usage. Therefore, as introduced in some theories of optimization (e.g., operational research), they can be used to optimize in relevant dimensions, such as time, resources, or outcomes.

Without data observations that would be assimilated into monitoring, a system cannot be self-improving or have a deterministic improvement process. I have addressed this topic in another book, “What is Data Governance”, in which I cover the aspects of maturity. In the third section, I highlight the 5th level of maturity¹ as

Optimizing (Level 5)

The data governance program focuses on continuous improvement, monitoring the quality and performance of processes and their role in meeting business objectives. The organization efficiently and consistently improves and adapts its practices to strategic changes.

Incident Response

The last part of DataOps is related to issues management; in fact, if it is clear to everyone in a data organization that resolving issues is a “big thing”, meaning it is extremely expensive, it is also the one that is the least tracked and measured.

After seeing this in hundreds of scenarios and reporting ten times more of the same facts, my explanation is that when something goes wrong with data, without data observability and associated practices described in this book, everything is done in a rush in a vain effort to recover trust as fast as possible. So nothing is apparent, but the persons who have been involved on either side of this situation feel the ineffectiveness of the whole process.

Rephrasing, my introduction line: it feels like issue management is inefficient, and lots of money is wasted.

Hence, the role of data observability is to provide visibility about the issue management, from issue detection, resolution, and post-mortem, which constitute the appropriate process for incident response.

Based on the data observations, which include user involved and code change, organizational improvement or prevention mechanism can be drawn from their correlation with the number of tickets resolved, their frequency, the data involved, the system involved, the duration of the resolution process, etc. Consequently, you can associate cost estimation to TTD or TTR and so estimate and even predict (for budgeting) the total cost of ownership (TCO) of the system.

Orchestration

It is often crucial to determine the order of execution of applications in a pipeline, as they have to follow the underlying dependencies at the data level. For example, a batch application reading a CSV must start before the CSV has been updated with the latest data, especially if freshness is important.

When the pipeline is simple, with a limited number of contributors (ideally 1), and only one orchestration tool, managing those weak dependencies is doable. But if any of those parameters changes, then any application producing data can see its behavior changed without knowing if another application relies on it, because:

It is not clear whether the next application in the orchestrated pipeline is using the data

It is not known if there is another application from a pipeline orchestrated in another tool

It is not clear what are the requirements for the data from the next application (schema changed, …)

etc.

Note

An orchestration solution is creating a DAG of applications. It is not an ETL, so some of the examples above don’t apply (in the case where there is only one orchestrator).

It turns out that with data observable pipelines, a solution comes by default to those cases, as described in Figure 5-9. This is because data observable pipelines are generating, the data observations necessary to detect issues either individually (rules checked in the applications) or by the data observability platform (cf. expectation validator in Figure 5-6).

Then the pipelines are automatically breaking (cf. circuit breaking) themselves as some of their steps are self-declared as failed, taking this decision based on the detected issues returned by, for example, a health status service (Figure 5-6).

With this flexibility enabled by data observability, the interconnections existing across pipelines–potentially orchestrated by several different orchestrations–can be managed externally, without the orchestrations themselves making necessary strong coordination, or even adding extra burden on the data engineers or data ops to create additional “data quality” scripts or applications surrounding all steps along all pipelines. For example, adding such checking script before and after three applications running in two different orchestrators (or DAGs) involving five data sources, we’re talking about twenty additional additional scripts (or queries) to added to the whole pipeline–which is now four to ten times² bigger than it should. .

Figure 5-6. Weakly connected data observable pipelines orchestrated by potentially different orchestrators and the Data Observability Platform

Software Engineering

The last undercurrent for data engineering (obviously not limited to that role), and most underestimated, is the panel of practices that were built and improved over the last decades, especially since computer science became more mainstream.

It is true that data is not directly mappable with software, or more precisely, softwares as there are many subcategories to be considered with their own peculiarities that segregate the applicability of this or that practice–embedded softwares, web applications, or, simply, websites for example have lots of dedicated best practices.

Consequently, it is not completely absurd to consider data as part of softwares, as data can’t really exist without processing³. Hence many software practices apply to data engineering. You’ll find several covered in Fundamentals of Data Engineering. I would add data observability, especially making applications data observable, as an additional practice that must be encoded in all data practitioners’ DNA, like it is for software engineers to log, test, and check the validity of a received message (e.g., encoding, formatting, etc) in their production code, for example.

Now that we have covered architectural and practical concepts, it makes sense to cover organizational concepts where data observability plays a strong role. For this, I will use a paradigm in which I strongly believe – Data Mesh.

Support for Data Mesh’s Data as Products

I’ll start this last section with a disclaimer: my intent here is not to explain what Data Mesh is–I’ll leave this to its creator, Zhamak Dehghani, or other experts investing a large stake of their time thinking about it. My goal is to give you some advice on why data observability is undoubtedly part of the practices and technological components of Data Mesh, at least of its implementation in an organization. If you want or need an explanation about Data Mesh, there are two things I suggest you to do, now, which are:

Read the genesis on Martin Fowler’s blog.

Read Zhamak’s book Data Mesh.

Of the many concepts supporting Data Mesh, I will focus on Data as a Product, which is, in a way, consisting in the delivery and what could be considered as an asset for the organization (e.g., it can be monetized).

In Data Mesh, data as a product is a principle supporting the implementation of data products, artifacts that can be combined via input and output APIs, plus an additional API for its exposure outside of the pure data flow, the control API. The book represents this as a hexagon shown in Figure 5-7. You can see that the APIs are annotated with standardized and domain agnostic, mostly for the same reasons as I introduced in Chapter 2: a core model for data observations.

Figure 5-7. A Data Product with its input, output, and control APIs

Putting aside the important concept of domains in Data Mesh, we can picture as in Figure 5-8 data as products connecting their input and output APIs to form a graph (a mesh).

Figure 5-8. Data as products in a mesh

You can see the mesh is connected with APIs. In fact, the line with arrows represents data-in-flight, as data is exchanged via input and output APIs and controlled by the control API. However, to explain data observability in such a mesh, I use a dual graph⁴. The dual graph, in this case, is when we consider the input and output APIs connections as the nodes, and the links represent the control APIs alone⁵. As this can be tricky to represent, Figure 5-9 shows the dual graph for the mesh in Figure 5-8.

Figure 5-9. Dual data mesh of Figure 5-8

In fact, this dual graph could be called the “Data Usage Mesh.” We can see data products DP1 and DP2 are present twice, it doesn’t mean those products are duplicated, but controlled information of a data product is highly dependent on its usage (the way the output API is used).

If a node is not fulfilling expectations, then the next edge can be disconnected to avoid propagation, which is the circuit breaker in place creating a (potentially isolated) subgraph linked to the found issue. In the examples above, let DP2 wrong only in the context of DP4, propagating the issue in the Data Mesh graph flag all downstream data products impacted DP3, DP4, DP6, DP7, instead of DP4 alone. By adding the Data Usage Mesh in the decision-making process, only DP2, DP4, and DP7 are impacted, and thus DP3 and DP6 are kept safe.

This shows the importance of contextual management of the expectations and, what is nice with Data Mesh, is it’s structured around APIs. This fact is clearly exposed - versus the general-purpose quality of a data source (e.g., file, table, …) accessed randomly without controlled APIs (e.g., JDBC, filesystem access, …).

That said, I have still not covered how data observability supports creating and managing data as products in a data mesh. For that, I have to open the Data Product box and look where Data Observability can be added (see Figure 5-10).

Figure 5-10. Data as a Product and Data Observability

It turns out that there are two levels of data observability, in a way, involved here.

The first level is external to the data product, to expose via the control API visibility and trustability for other data products built on it. However, it is likely coarse-grained because the intent at this level is not to explain all the nitty gritty details of how the data product works–which is likely to change over time anyway, without necessarily impacting the consumers–but to validate the data coming out the Output API.

At this level, the visibility of the connection (upstream data lineages) with the Input API is important to demonstrate and analyze within the boundaries of the data product (its internals) if a detected issue at the Output API has been either propagated from the Input API, or it is the result of a misuse of the data (within the pipeline).

However a data product is atomic in a data mesh meaning its internals can be extremely complex and not exposed .

Therefore, when an issue is due to the behavior of the data product, the second level of data observability, local to the data product level, is bringing the needed visibility to address it.

This fine-grain data observability provides the necessary mechanisms to operate at the data product level efficiently and can be, in fact, the source of the aggregated view delivered at the higher level–the data product level.

Because the first level of data observability is somewhat a coarse-grained (an aggregation) of the finer-grained information available in the second level, I have only connected the data observability platform to the latter to maximize the amount of visibility without redundancy.

Mixing both levels allows the definition of the SLO at the data product level, which are first linked to the SLA associated with what is exposed to the Control API (bottom-up). And secondly, the SLO can be adapted by leveraging the continuous validations happening in the data observability platform at the Input API of the dependent Data Products, which could also be exposed in the Control API, to provide this information to the upstream data products.

Conclusion

To sum up, this chapter has introduced data observability as a crucial element in building a robust data ecosystem. By implementing it at the foundation layer of the data platform, organizations can foster a strong data culture and efficiently manage their data. Data observability also aligns with the DataOps vision and supports all its pillars. Additionally, it bridges the gap in data organization to ensure the implementation of data strategies at scale and is integrated with related paradigms like Data Mesh. However, it is essential to note that the data observability platform cannot function independently and requires data products and pipelines to be data observable. The upcoming chapters will explore how to achieve this goal.

¹ Petrella, Andy: What is Data Governance, Chapter 3; 2021, O’Reilly.

² That’s insane, isn’t it?

³ Unless we consider our brains as such as we extend our cognitive power to the computers :).

⁴ https://en.wikipedia.org/wiki/Dual_graph

⁵ Hence we’ll not consider the “data” itself anymore, as it is abstracted out anyway.