4dc905c277b04146a5348e699b66b1ae

Unified Lambda Architecture (Realtime DB + DW + OLAP)

Mon, 19 Jun 2023 04:01:26 GMT

Database types

Realtime DB

The database should be able to scale and keep up with the huge amounts of data that are coming in from streaming services like Kafka, IoT and so on. The SLA for latencies should be in milliseconds to very low seconds.

The users also should be able to query the real time data and get millisecond or sub-second response times.

Data Warehouse (Analytics)

A data warehouse is specially designed for data analytics, which involves reading large amounts of data to understand relationships and trends across the data. The data is generally stored in denormalized form using Star or Snowflake schema.

Data warehouse is used in a little broader scope, I would say we are trying to address Data Marts here which is a subset of the data warehouse and addresses a particular segment rather than addressing the whole enterprise.

In this use case, the users not only query the real time data but also do some analytics, machine learning and reporting.

OLAP

OLAP is a kind of data structure where the data is stored in multi-dimensional cubes. The values (or measures) are stored at the intersection of the coordinates of all the dimensions.

Blue → 2017 Apr + “Cost of Goods Sold” + Accessories

Yellow → Clothing + “Gross Revenue” + 2016 Jan

If we want 2015 Units sold for Accessories → sum up corresponding 5 coordinates. Each Coordinate is the sum of all the underlying transactions.

Problem Statement

Any proposed solution should be able to address the following 4 main areas:

● It should be able to consume the data from streaming systems like Kafka in real time with millisecond or very low second latencies. It should also process the data in batches.

● Users should be able to query data with millisecond to very low second latencies.

● The database should be able to scale up to 100’s of concurrent queries.

● The users should be able to do analytics on the source data.

Why not databases like RedShift?

● Can not stream huge amounts of data.

● Does not give millisecond response times.

● Does not provide high concurrency.

● Managed service might cost more.

Why not NoSQL (e.g.. Cassandra, Dynamo, HBase, MongoDB/Couchbase)

● NOSQL databases like Cassandra and HBase scales with Real-time streaming. Mongo/Couchbase will not scale for real time streaming of that volume.

● Provides Millisecond response times for querying data on partition keys only.

● Provides High Concurrency.

● Does not support analytics.

● Does not support OLAP.

Unified Lambda Architecture

This blog proposes Unified Analytical solution (Realtime DW + OLAP) that addresses the following:

● Real Time data streaming

● Real Time Analytics

● High Concurrency

● OLAP Analytics

● High Availability

● Cost effective.

The proposed solution will solve the current use cases and provide a very good enterprise solution and add a lot of value to the data like providing more real-time reporting, analytics, training models on real time data and also on large volume of data.

The above diagram explains how the proposed solution will work. It uses “unified lambda architecture” which provides Batch, Stream, and Speed layers together. The data ingested into detailed tables and Rollups/Cubes will be synced behind the scenes based on the design and instructions provided.

When users try to access data, the query optimizer redirects to underlying data (can be detailed, aggregated and both) based on the query need. End users need not know the underlying mechanism and they just simply use SQL to access the data.

Technologies

● Apache Druid

● Apache Starrocks

● Apache Kylin

The following table compares the features for each technology mentioned above:

Cube Vs Materializers

An OLAP cube is a data structure that overcomes the limitations of relational databases by providing rapid analysis of data. Cubes can display and sum large amounts of data while also providing users with searchable access to any data points. This way, the data can be rolled up, sliced, and diced as needed to handle the widest variety of questions that are relevant to a user's area of interest.

Processing of an OLAP cube occurs when all the aggregations for the cube are calculated and the cube is loaded with these aggregations and data. Dimension and fact tables are read, and the data is calculated and loaded into the cube. When we design an OLAP cube, processing must be carefully considered because of the potentially significant effect that processing might have in a production environment where millions of records may exist. A full process is very compute intensive and takes a very long time based on the amount of data and computation we are using.

Adding more dimensions adds more complexity to cube processing. When real-time data comes in, it might take some time to add it to all partitions.

The amount of cube processing can be reduced by optimizing it based on query usage and/or space-based optimization.

On the other hand, Materializers are a small subset of the Cube which will reduce time and complexity of cube processing.

Infrastructure and Platform Architecture – Modern data lakes and databases

Sun, 18 Jun 2023 02:38:31 GMT

This blog puts together Infrastructure and platform architecture for modern data lake. The following are taken into consideration while designing the architecture:

Should be portable to any cloud and on-prem with minimal changes.

Most of the technologies and processing will happen on Kubernetes so that it can be run on any Kubernetes cluster on any cloud or on-prem.

All the technologies and processes use auto scaling features so that it will allocate and use resources minimally possible at any given time without compromising the end results.

It will take advantage of spot instances and cost-effective features and technologies wherever possible to minimize the cost.

It will use open-source technologies to save licensing costs.

It will auto provision most of the technologies like Argo workflows, Spark, Jupyterhub (Dev environment for ML) and so on, which will minimize the use of the provider specific managed services. This will not only save money but also can be portable to any cloud or multi-cloud including on-prem.

Concept

The entire Infrastructure and Platform for modern data lakes and data platform consists of 3 main Parts at very higher level:

Code Repository
Compute
Object store

The main concept behind this design is “Work anywhere at any scale” with low cost and more efficiently. This design should work on any cloud like AWS, Azure or GCP and on on-premises. The entire infrastructure is reproducible on any cloud or on-premises platform and make it work with some minimal modifications to code.

Below is the design diagram on how different parts interact with each other. The only pre-requisite to implement this is Kubernetes cluster and Object store.

Object store

Object store is the critical part of the design and holds all the persistent data including entire data lake, Deep storage for databases (metadata, catalogs and online data) and stores logs or any other information that need to be persisted. Different policies will be created to archive and purge data based on the retention polices and compliance.

Code Repository

Entire Infrastructure, software and data pipelines will be created using a code repository which will play very key role in this project. Entire “compute” can be created using code repository in conjunction with devOps toolset.

Compute

All the Compute will be created on Kubernetes using code repository and CI/CD pipelines. It can work on any cloud or on-premises if Kubernetes environment is available.

We are planning to run most of the processing on spot instances as much as possible if we implement this on cloud, which will significantly reduce the cost. The ideal target for the entire Compute is 30% (or less) on regular instances and 70% (or more) on spot instances.

Technical Architecture

The following is the proposed infrastructure and platform technical architecture, which should satisfy all the Data Lake project technical needs. Most of the technologies will be built as self-serve capabilities using automation tools so that the developers, Data Scientists and ML Engineers can use them as needed.

Data Processing and ML Training

Most of the work is done during this stage of the project and most of the cost incurred for the project happens here.

Argo workflow and Argo Events

Argo is an open-source Workflow Engine for orchestrating tasks on Kubernetes. Argo allows us to create and run advanced workflows entirely on your Kubernetes cluster. Argo Workflows is built on top of Kubernetes, and each task is run as a separate Kubernetes pod. Many reputable organizations in the industry use Argo Workflows for ML (Machine Learning), ETL (Extract, Transform, Load), Data Processing, and CI/CD Pipelines.

Argo is basically a Kubernetes extension and hence, it is installed using Kubernetes. Argo Workflows allows organizations to define their tasks as DAGs using YAML. Argo comes with a native Workflow Archive for auditing, Cron Workflows for scheduled workflows, and a fully featured REST API. Argo comes with a list of killer features that set it apart from similar products.

Note: the above diagram is courtesy of Argo documentation.

Argo Events is an event-driven workflow automation framework for Kubernetes which helps us to trigger K8s objects, Argo Workflows, Serverless workloads, etc. on events from a variety of sources like webhooks, S3, schedules, messaging queues, gcp pubsub, sns, sqs, etc.

Note: the above diagram is courtesy of Argo documentation.

Key Features and advantages of Argo

Open-Source: Argo is a fully open-source project at the Cloud Native Computing Foundation (CNCF). It is available free of cost for everyone to use.

Native Integrations: Argo comes with native artifact support to download, transport, and upload files during runtime. It supports any S3 compatible Artifact Repository such as AWS, GCS, HTTP, Git, Raw, and Minio. Also, it’s easy to integrate with Spark using SPOK (Spark on Kubernetes). It can be used to run Kubeflow ML pipelines.

Scalability: Argo Workflows has robust retry mechanisms for high reliability and is highly scalable. It can manage thousands of pods and workflows in parallel.

Customizability: Argo is highly customizable, and it supports templating and composability to create and reuse workflows.

Powerful User Interface: Argo comes with a fully featured User Interface (UI) that is easy to use. Argo Workflows v3.0 UI also supports Argo Events and is more robust and reliable. It has embeddable widgets and a new workflow log viewer.

Cost savings: We can run a lot of processing on Spot instances where permitted, which saves lot of cost.

Spark (SPOK)

It is recommended to use Spark for most of the processing wherever possible including data ingestion, transformation and for ML training. Spark is a reliable and proven distributed processing engine and can process the ETL pipelines very faster. This is heavily used in the industry and standard for any big data and data lake project. We can also leverage on the spot instances for spark processing that reduces the overall cost of the project.

The following is an example of how we can effectively run Spark on Spot node pools. As all the processing happens on executor nodes. Running executors on spot node pools tremendously saves cost. Spark also can reprocess any failed tasks automatically which makes it a reliable candidate to run workloads on spot instances.New Paragraph

ArgoCD

ArgoCD allows us to use custom or vendor provided images to process the incoming data and any other workloads. We can also run this on Spot instance node pool with retry’s. Some enhancements may be required to run them on Spot instance node pools.

Hive

Hive can be used to access data stored on any file system using SQL. It can be used for metadata cataloging as well. Hive also comes with ACID transactional features, which can be used for delta lake use cases if any.

Delta lake or Hive Transactional

Delta lake allows row level manipulation of data and can-do time travel on data. Hive transactional, Databricks and Iceberg are some of the technologies that will facilitate delta lake on Big Data.

Streaming (Kafka)

Kafka or similar technologies can be used where we need to process non file based incoming data in real-time.

JupyterHub

JupyterHub brings the power of notebooks to groups of users. It gives users access to computational environments and resources without burdening the users with installation and maintenance tasks. Users - including engineers and data scientists - can get their work done in their own workspaces on shared resources which can be managed efficiently by system administrators.

JupyterHub runs on Kubernetes makes it possible to serve a pre-configured data science environment to any user.

Persistent data storage

Object Store (S3, Azure Blob, GCS) plays a very critical role in storing all the data, which is a key part of this project. The following are some of the ways to store and process the data from Object store cost effectively:

❖ Compress data effectively. This will reduce storage costs, IO cost and reduce processing times.

❖ Retain only required data. Have an Archival process in place.

❖ Partition data effectively. This will reduce the IO costs when processing the data.

❖ Reduce accessing raw data as much as possible by incorporating different presentation layers with minimal data.

❖ Utilize cold storage wherever possible.

Virtualization layer

As accessing data from Object store is cumbersome, it's a good idea to build virtualization layer on top of it. This will bring the following advantages to raw data:

❖ Data can be accessed using SQL statements.

❖ This will serve as a metadata catalog on top of the Object store.

❖ Can use partitioning and other optimization techniques to reduce IO when consuming data from the Object store.

Hive, Athena, Presto and Redshift Spectrum are some of the technologies that can be used for this purpose.

Database (Speed/Consumption data layer)

The OLAP and/or NoSQL database may be required to satisfy the end user data consumption needs. Databases like Starrocks, Druid or Cassandra may be used for this purpose. The selection of the database is based on following requirements:

❖ The volume of the data stored in the database.

❖ Amount of the data ingested by time interval.

❖ How the data is ingested is like real time streaming and/or batch processing.

❖ Response times needed while accessing data by end users.

❖ QPS/TPS

❖ Data access patterns.

Data facilitation and secure data access

This stage is lightweight and facilitates access to the data, which is stored in a persistent data store like database. This will also provide authentication and authorization mechanisms to access the data.

Serverless Cloud functions and/or microservice architecture may be used to facilitate REST APIs.

Orchestration and DevOps

Kubernetes, Docker, and Container services

Kubernetes, Docker, and Container services play key roles in this project. These technologies facilitate most of the components mentioned above. The technologies like Spark will be built and readily available as containers for developers to develop workflows, pipelines and so on using them.

Kubernetes cluster will have different node pools (or groups) to run different workloads. Most of the processing pipelines and model training can run on node pools(groups) with spot instances.

CI/CD

Jenkins (CI) and Argo CD (CD) will be used for continuous integration and continuous deployment into Kubernetes. Terraform may be used for non-Kubernetes related cloud deployments. Some additional tools like Chef or Ansible may be leveraged when the need arises for those technologies.

Git is used as a code repository, which will be integrated with CI/CD pipelines.New Paragraph

Compute for Big Data

Sat, 17 Jun 2023 21:23:01 GMT

Spark-On-Kubernetes is growing in adoption across the ML Platform and Data engineering. The goal of this blog is to create a multi-tenant Jupyter notebook server with built-in interactive Spark sessions support with Spark executors distributed as Kubernetes pods.

Problem Statement

Some of the disadvantages of using Hadoop (Big Data) clusters like Cloudera and EMR:

Requires designing and build clusters which takes a lot of time and effort.
Maintenance and support.
Shared environment.
Expensive as there are a lot of overheads like master nodes and so on.
Not very flexible as different teams need different libraries.
Different cloud technologies and on-premises come with different sets of big data implementations.
Cannot be used for a large pool of users.

Proposed solution

The proposed solution contains 2 parts, which will work together to provide a complete solution. This will be implemented on Kubernetes so that it can work on any cloud or on-premises in the same fashion.

I. Multi-tenant Jupyterhub

JupyterHub allows users to interact with a computing environment through a webpage. As most devices have access to a web browser, JupyterHub makes it easy to provide and standardize the computing environment of a group of people (e.g., for a class of data scientists or an analytics team).

This project will help us to set up our own JupyterHub on a cloud and leverage the cloud's scalable nature to support large groups of users. Thanks to Kubernetes, we are not tied to a specific cloud provider.

II. Spark on Kubernetes (SPOK)

Users can spin their own spark resources by creating sparkSession. Users can request several executors, cores per executor, memory per executor and driver memory along with other options. The Spark environment will be ready within a few seconds. Dynamic allocation will be used if none of those options are chosen.

All the computes will be terminated if they’re idle for 30 minutes (or can be set by the user). The code will be saved to persistent storage and available when the user logs-in next time.

Data Flow Diagram

Implementation

Interface

WebURL: https://Jupyterhub.*****.com

Notebook server configuration

In this section, the user selects the image and configs like pySpark, python with pandas and so on.

Data lake design patterns on Azure (Microsoft) cloud

Sat, 09 May 2020 19:00:43 GMT

Build scalable data lakes on Microsoft cloud (Azure)

Unlike the traditional data warehousing, complex data lake often involves combination of multiple technologies. It is very important to understand those technologies and also learn how to integrate them effectively. This blog walks through different patterns for successful implementation any data lake on Azure cloud platform.

Pattern I: Full Data lake stack

Pattern II: Unified Batch and Streaming model

Pattern III: Lambda streaming architecture

When we are building any scalable and high performing data lakes on cloud or on-premise, there are two broader groups of toolset and processes play critical role. One kind of toolset involves in building data pipelines and storing the data. The another set of toolset or processes does not involve directly in the data lake design and development but plays very critical role in the success of any data lake implementation like data governance and data operations.

The above diagrams show how different Azure managed services can be used and integrated to make it full blown and scalable data lake. You may add and remove certain tools based on the use cases, but the data lake implementation mainly moves around these concepts.

Here is the brief description about each component in the above diagrams.

1. Data Sources

The data can come from multiple desperate data sources and data lake should be able to handle all the incoming data. The following are the some of the sources:

• OLTP systems like Oracle, SQL Server, MySQL or any RDBMS.

• Various File formats like CSV, JSON, AVRO, XML, Binary and so on.

• Text based and IOT Streaming data

2. Data Ingestion

Collecting and processing the incoming data from various data sources is the critical part of any successful data lake implementation. This is actually most time consuming and resource intensive step.

Azure has various highly scalable managed services to develop and implement very complicated data pipelines of any scale.

Azure Data Factory

Azure Data Factory is a Microsoft fully managed service where you can build unified batch and streaming data pipelines. It also provides horizontal scaling and tightly integrated with other Big Data components like SQL DataWarehouse, Azure CosmosDB, Azure Table Storage and Event Hub.

https://azure.microsoft.com/en-us/services/data-factory/

Azure HDInsight

Azure HDInsight is a managed Azure service for Hadoop/Spark echo system. You can use HDInsight for various purposes:

• To build data pipelines using spark, especially when you have lot of code written in Spark when migrating from the on-premise.

• To do Lift and Shift existing Hadoop environment from onsite to cloud.

• If you want to use Hive and HBase databases part of your use cases.

• To build Machine learning and AI pipelines using Spark.

HDInsight clusters can be built on on-demand and also can be auto scaled depending on the need. You can also use preemptive VMs where you don’t need production scale SLAs, which costs lot less compare to using regular instances.

https://azure.microsoft.com/en-us/services/hdinsight/

3. Raw data layer

Object storage is central to any data lake implementation. Azure BLOB and Azure Data Lake Storage serves as raw layer. You can build highly scalable and highly available data lake raw layer using Azure BLOB which also provides very high SLAs.

Azure Storage supports three types of BLOBs, Block, Append and Page BLOBs.

https://docs.microsoft.com/en-us/azure/storage/blobs/

4. Consumption layer

All the items mentioned before are internal to data lake and will not be exposed for external user. Consumption layer is where you store curated and processed data for end user consumption. The end user applications can be reports, web applications, data extracts or APIs.

The following is some of the criteria while choosing database for the consumption layer:

• Volume of the data.

• Kind of the data retrieval patterns like whether applications use analytical type of queries like using aggregations and computations or retrieves just based on some filtering.

• How the data ingestion happens whether it’s in large batches or high throughput writes (IOT or Streaming) and so on.

• Whether the data is structured, semi-structured, quasi-structured or unstructured.

• SLAs.

Azure SQL Data Warehouse (Synapse Analytics)

Azure SQL Data Warehouse is managed analytical service that brings together enterprise data warehouse and Big Data analytics. SQL Data Warehouse is an important part of the Azure data lake stack. It is a data warehouse database, serverless managed service and can scale on petabytes of data. SQL Data Warehouse is very good for analytical queries where you use lot of aggregations and computations. It is renamed as Synapse Analytics.

https://azure.microsoft.com/en-us/services/synapse-analytics/

Azure CosmosDB

Azure Cosmos DB is a managed NoSQL database available on Azure cloud which provides low latency, high availability and scalability. This will allow to migrate MongoDB, Cassandra and other NoSQL workloads to the cloud.

https://azure.microsoft.com/en-us/services/cosmos-db/

Azure Table Storage

Azure Table Storage is a wide column NoSQL database scales over petabytes of data and provides semi structured schema model. It is also Azure fully managed service.

https://azure.microsoft.com/en-us/services/storage/tables/

5. Machine Learning and Data Science

Machine Learning and Data science teams are biggest consumers of the data lake data. They use this data to train their models, forecast and use the trained models to apply for future data variables.

Azure Data Explorer

Azure Data Explorer is a fully managed Azure data analytics service for real-time analysis on large volumes of data streaming from applications, websites, IoT devices, and more. Ask questions and iteratively explore data on the fly to improve products, enhance customer experiences, monitor devices, and boost operations. Quickly identify patterns, anomalies, and trends in your data. Explore new questions and get answers in minutes. Run as many queries as you need, thanks to the optimized cost structure.

https://azure.microsoft.com/en-us/services/data-explorer/

Azure HDInsight

Please see above

Azure Machine Learning

Azure Machine Learning service is a huge combination of ML/AI library, where you can build and deploy Machine Learning models quickly and easily. It also has very good support for open-source frameworks and languages including MLflow, Kubeflow, ONNX, PyTorch, TensorFlow, Python, and R

https://azure.microsoft.com/en-us/services/machine-learning/

a. Data Governance

Data Governance on cloud is a vast subject. It involves lot of things like security and IAM, Data cataloging, data discovery, data Lineage and auditing.

Azure Data Catalog

Azure Data Catalog is a fully managed metadata management service which can be fully integrated with other components like Azure BLOB, SQL Data Warehouse and Azure Table Storage. You can quickly discover, understand and manage the data stored in your data lake.

You can view my blog for detailed information on data catalog.

https://www.unifieddatascience.com/data-cataloging-metadata-on-cloud

Azure Search

Azure Search is a kind of enterprise search tool that will allow you quickly, easily, and securely find information.

https://azure.microsoft.com/en-us/services/search/

Azure Active Directory

Please refer to my data governance blog for more details.

https://www.unifieddatascience.com/data-governance-for-data-lakes-on-cloud

Azure Key Vault

Azure Key Vault is a hosted KMS that lets us manage encryption keys in the cloud. You can use Azure Key Vault to encrypt keys and secrets that use keys stored in hardware security modules (HSM).

https://azure.microsoft.com/en-us/services/key-vault/

b. Data Operations

Operations, Monitoring and Support is key part of any data lake implementations. Azure provides various tools to accomplish this.

Azure Monitor

Azure Cloud Platform offers Azure Monitor, a comprehensive set of services for collecting data on the state of applications and infrastructure. Specifically, it supports three ways of collecting and receiving information.

https://azure.microsoft.com/en-us/services/monitor/

Please refer to my blog cloud operations for full details.

https://www.unifieddatascience.com/cloud-operation-and-monitoring-on-gcp

Data lake design patterns on AWS (Amazon) cloud

Fri, 08 May 2020 03:17:38 GMT

Build scalable data lakes on Amazon cloud (AWS)

Pattern I: Full Data lake stack

Pattern II: Unified Batch and Streaming model

Pattern III: Lambda streaming architecture

The above diagrams show how different Amazon managed services can be used and integrated to make it full blown and scalable data lake. You may add and remove certain tools based on the use cases, but the data lake implementation mainly moves around these concepts.

Here is the brief description about each component in the above diagrams.

1. Data Sources

The data can come from multiple desperate data sources and data lake should be able to handle all the incoming data. The following are the some of the sources:

• OLTP systems like Oracle, SQL Server, MySQL or any RDBMS.

• Various File formats like CSV, JSON, AVRO, XML, Binary and so on.

• Text based and IOT Streaming data

2. Data Ingestion

AWS has various highly scalable managed services to develop and implement very complicated data pipelines of any scale.

AWS Data Pipeline

AWS Data Pipeline is Amazon fully managed service where you can build unified batch and streaming data pipelines. It also provides horizontal scaling and tightly integrated with other Big Data components like Amazon Redshift, Amazon Dynamo, Amazon S3 and Amazon EMR

https://aws.amazon.com/datapipeline/

AWS Glue

AWS Glue is a fully managed ETL service which enables engineers to build the data pipelines for analytics very fast using its management console. You can build data pipelines using its graphical user interface (GUI) with few clicks. It automatically discovers the data and also catalog the data using AWS Glue catalog service.

https://aws.amazon.com/glue/

AWS EMR

AWS EMR is a managed amazon cloud service for Hadoop/Spark echo system. You can use AWS EMR for various purposes:

• To build data pipelines using spark, especially when you have lot of code written in Spark when migrating from the on-premise.

• To do Lift and Shift existing Hadoop environment from onsite to cloud.

• If you want to use Hive and HBase databases part of your use cases.

• To build Machine learning and AI pipelines using Spark.

AWS EMR clusters can be built on on-demand and also can be auto scaled depending on the need. You can also use spot instances where you don’t need production scale SLAs, which costs lot less compare to using regular instances.

https://aws.amazon.com/emr/

3. Raw data layer

Object storage is central to any data lake implementation. AWS S3 serves as raw layer. You can build highly scalable and highly available data lake raw layer using AWS S3 which also provides very high SLAs.

It also comes with various storage classes like S3 Standard, S3 Intelligent-Tiering, S3 Standard-IA, S3 One Zone-IA and S3 Glacier, which are used for various use cases and to meet different SLAs.

https://aws.amazon.com/s3/storage-classes/

4. Consumption layer

The following is some of the criteria while choosing database for the consumption layer:

• Volume of the data.

• Kind of the data retrieval patterns like whether applications use analytical type of queries like using aggregations and computations or retrieves just based on some filtering.

• How the data ingestion happens whether it’s in large batches or high throughput writes (IOT or Streaming) and so on.

• Whether the data is structured, semi-structured, quasi-structured or unstructured.

• SLAs.

AWS Dynamo

Amazon Dynamo is a distributed wide column NoSQL database can be used by application where it needs consistent and millisecond latency at any scale. It is fully managed and can be used for document and wide column data models. It also supports flexible schema and can be used for web, ecommerce, streaming, gaming and IOT use cases.

https://aws.amazon.com/dynamodb/

Amazon Redshift

Amazon Redshift is a fast, fully managed analytical data warehouse database service scales over petabytes of data. Amazon Redshift provides a standard SQL interface that lets organizations use existing business intelligence and reporting tools. Amazon Redshift is a columnar database and distributed over multiple nodes allows to process requests parallel on multiple nodes.

https://aws.amazon.com/redshift/

Amazon DocumentDB

Amazon DocumentDB is a fully managed document-oriented database service which supports JSON data workloads. It is MongoDB compatible. Its fast, high available and scales over huge amounts of data.

https://aws.amazon.com/documentdb/

5. Machine Learning and Data Science

Machine Learning and Data science teams are biggest consumers of the data lake data. They use this data to train their models, forecast and use the trained models to apply for future data variables.

Amazon Glue

Please refer above

Amazon EMR

Please refer above

Amazon SageMaker

Amazon SageMaker can be used to quickly build, train and deploy machine learning models at scale; or build custom models with support for all the popular open-source frameworks.

https://aws.amazon.com/sagemaker/

Amazon ML and AI

Amazon has huge set of robust and scalable Artificial Intelligence and Machine Learning tools. It also provides pre-trained AI services for computer vision, language, recommendations, and forecasting.

https://aws.amazon.com/machine-learning/

a. Data Governance

Data Governance on cloud is a vast subject. It involves lot of things like security and IAM, Data cataloging, data discovery, data Lineage and auditing.

Amazon Glue Catalog

Amazon Glue Catalog is a fully managed metadata management service which can be fully integrated with other components like Data Pipelines, Amazon S3 and so on. You can quickly discover, understand and manage the data stored in your data lake.

You can view my blog for detailed information on data catalog.

https://www.unifieddatascience.com/data-cataloging-metadata-on-cloud

Amazon CloudSearch

Cloud search is a kind of enterprise search tool that will allow you quickly, easily, and securely find information.

https://aws.amazon.com/cloudsearch/

AWS IAM

Please refer to my data governance blog for more details.

https://www.unifieddatascience.com/data-governance-for-data-lakes-on-cloud

AWS KMS

AWS KMS is a hosted KMS that lets us manage encryption keys in the cloud. We can create/generate, rotate, use, and destroy AES256 encryption keys just like we would in our on-premises environments. We can also use the cloud KMS REST API to encrypt and decrypt data.

https://aws.amazon.com/kms/

b. Data Operations

Operations, Monitoring and Support is key part of any data lake implementations. AWS provides various tools to accomplish this.

AWS CloudTrail

AWS offers CloudTrail, a comprehensive set of services for collecting data on the state of applications and infrastructure.

Please refer to my blog cloud operations for full details.

https://www.unifieddatascience.com/cloud-operation-and-monitoring-on-gcp

AWS CloudWatch

AWS CloudWatch Logs maintains three audit logs for each AWS project, folder, and organization: Admin Activity, Data Access, and System Event. AWS write audit log entries to these logs to help us answer the questions of "who did what, where, and when?" within your AWS Cloud resources.

Please refer to my blog for more details.

https://www.unifieddatascience.com/cloud-operation-and-monitoring-on-gcp

Data lake design patterns on google (GCP) cloud

Thu, 07 May 2020 15:39:15 GMT

Build scalable data lakes on cloud (GCP)

Pattern I: Full Data lake stack

Pattern II: Unified Batch and Streaming model

Pattern III: Lambda streaming architecture

The above diagrams show how different google managed services can be used and integrated to make it full blown and scalable data lake. You may add and remove certain tools based on the use cases, but the data lake implementation mainly moves around these concepts.

Here is the brief description about each component in the above diagrams.

1. Data Sources

The data can come from multiple desperate data sources and data lake should be able to handle all the incoming data. The following are the some of the sources:

• OLTP systems like Oracle, SQL Server, MySQL or any RDBMS.

• Various File formats like CSV, JSON, AVRO, XML, Binary and so on.

• Text based and IOT Streaming data

2. Data Ingestion

GCP has various highly scalable managed services to develop and implement very complicated data pipelines of any scale.

Cloud Dataflow

Cloud Dataflow is a google fully managed service where you can build unified batch and streaming data pipelines. It is built on open source Apache beam project. It also provides horizontal scaling and tightly integrated with other Big Data components like Big Query, Pub/Sub and Stack driver.

https://cloud.google.com/dataflow

https://beam.apache.org/

Cloud Dataproc

Cloud Dataproc is a managed google service for Hadoop/Spark echo system. You can use Dataproc for various purposes:

• To build data pipelines using spark, especially when you have lot of code written in Spark when migrating from the on-premise.

• To do Lift and Shift existing Hadoop environment from onsite to cloud.

• If you want to use Hive and HBase databases part of your use cases.

• To build Machine learning and AI pipelines using Spark.

Dataproc clusters can be built on on-demand and also can be auto scaled depending on the need. You can also use preemptive VMs where you don’t need production scale SLAs, which costs lot less compare to using regular instances.

https://cloud.google.com/dataproc

Cloud Composer

Cloud Composer is a fully managed workflow engine built on top of open source Apache Airflow. You can call dataflow and Dataproc spark jobs from Composer and create a controlled workflow and schedule them to run.

https://cloud.google.com/composer

Cloud Data Fusion

Cloud Data Fusion is a new entry to GCP ETL/ELT stack. It is a fully managed and cloud native data pipeline tool. It allows developers build high performing data pipelines graphically. It also has huge set of open source library of preconfigured connectors that you can connect to most of the data sources. It’s kind of code free data pipeline model where you save lot of time on developing complex transformations and data ingestion.

https://cloud.google.com/data-fusion

3. Raw data layer

Object storage is central to any data lake implementation. Cloud storage serves as raw layer. You can build highly scalable and highly available data lake raw layer using Cloud storage which also provides very high SLAs.

There are also different classes of Object storage like multi-regional, regional, coldline and nearline which can be used for different purposes based on requirements.

https://cloud.google.com/storage

4. Consumption layer

The following is some of the criteria while choosing database for the consumption layer:

• Volume of the data.

• Kind of the data retrieval patterns like whether applications use analytical type of queries like using aggregations and computations or retrieves just based on some filtering.

• How the data ingestion happens whether it’s in large batches or high throughput writes (IOT or Streaming) and so on.

• Whether the data is structured, semi-structured, quasi-structured or unstructured.

• SLAs.

Cloud BigTable

BigTable is a petabyte scale NoSQL managed services database which is good for both analytical and operational workloads. It is very good for high throughput read/writes and low latency (sub 10ms) database needs. Some of the use cases for BigTable are streaming, time series data, ad tech, fintech and so on. BigTable provides eventual consistency and atomic transactions. It is little expensive compared to BigQuery and Datastore.

https://cloud.google.com/bigtable

BigQuery

BigQuery is an important part of the google data lake stack. It is a data warehouse database, serverless managed service and can scale on petabytes of data. BigQuery is very good for analytical queries where you use lot of aggregations and computations.

BigQuery also comes with ML, GIS and BI engine features.

https://cloud.google.com/bigquery

Cloud Datastore (Firestore)

Cloud Datastore is a highly scalable NoSQL database, ACID compliant and also allows indexing. It is also schemaless document store. Useful where you need NoSQL document DB use cases and transactions.

https://cloud.google.com/datastore

5. Machine Learning and Data Science

Machine Learning and Data science teams are biggest consumers of the data lake data. They use this data to train their models, forecast and use the trained models to apply for future data variables.

Cloud Dataprep

Cloud Dataprep is a managed service that will help to clean, explore and prepare data, which can be used for data analytics and machine learning. It also comes with visual graphical user interface.

Cloud Dataproc

The Spark ML which is part of the Cloud Dataproc can be used for building machine learning pipelines and to build and train data models.

Cloud AutoML

Cloud AutoML provides various machine learning tools which allows to build, and train various high-quality machine learning data models required by business. Some of the AutoML tools include vision, video intelligence, natural language, translation and tables.

https://cloud.google.com/automl

AI Hub

Google provides various plug and play AI components into AI Hub.

https://cloud.google.com/products/ai

a. Data Governance

Data Governance on cloud is a vast subject. It involves lot of things like security and IAM, Data cataloging, data discovery, data Lineage and auditing.

Data Catalog

Data Catalog is a fully managed metadata management service which can be fully integrated with other components like Cloud storage, Big Query and Pub/Sub. You can quickly discover, understand and manage the data stored in your data lake.

You can view my blog for detailed information on data catalog.

https://www.unifieddatascience.com/data-cataloging-metadata-on-cloud

Cloud Search

Cloud search is a kind of enterprise search tool that will allow you quickly, easily, and securely find information.

https://cloud.google.com/products/search

Cloud IAM

Please refer to my data governance blog for more details.

https://www.unifieddatascience.com/data-governance-for-data-lakes-on-cloud

Cloud KMS

Cloud KMS is a hosted KMS that lets us manage encryption keys in the cloud. We can create/generate, rotate, use, and destroy AES256 encryption keys just like we would in our on-premises environments. We can also use the cloud KMS REST API to encrypt and decrypt data.

https://cloud.google.com/kms

b. Data Operations

Operations, Monitoring and Support is key part of any data lake implementations. Google provides various tools to accomplish this.

Stack driver

Google Cloud Platform offers Stackdriver, a comprehensive set of services for collecting data on the state of applications and infrastructure. Specifically, it supports three ways of collecting and receiving information.

Please refer to my blog cloud operations for full details.

https://www.unifieddatascience.com/cloud-operation-and-monitoring-on-gcp

Audit Logging

Cloud Audit Logs maintains three audit logs for each Google Cloud project, folder, and organization: Admin Activity, Data Access, and System Event. Google Cloud services write audit log entries to these logs to help us answer the questions of "who did what, where, and when?" within your Google Cloud resources.

Please refer to my blog for more details.

https://www.unifieddatascience.com/cloud-operation-and-monitoring-on-gcp

Disaster Recovery and Business Continuity Plan on google cloud

Thu, 23 Apr 2020 04:13:44 GMT

How to achieve DR and BCP on GCP and other clouds?

1 Introduction

The disaster recovery and business continuity plan on-premise is a very tedious and expensive task and sometimes takes years to fully implement based on its complexity. It involves having geographically separated secondary data centers, networking between data centers, tools and processes, and developing and implementing DR solutions. And also required a thorough business continuity plan to execute in case of failures.

Thanks to the emerging big data technologies and cloud, you can achieve all of them together if the toolset is properly designed and the new concepts are fully utilized.

For example, managed instance groups which club together groups of VM instances as a single unit provides high availability and autoscaling together and can solve DR and BCP requirements.

Certain managed services like BigQuery and Pub/Sub satisfies DR and BCP requirements without any extra effort. Certain resources like Cloud storage require minimum effort to implement. Cloud SQL and VM Instances require a lot more effort to make it compliance with DR and BCP.

The DR and BCP can be implemented for certain resources like dataflow and dataproc through continuous delivery (CD) processes.

This blog outlines different strategies to fully implement DR and BCP across the toolset and resources you are currently using and probably will use in near future on GCP.

2 Key terminology

Here are some key concepts that will go hand-in-hand with disaster recovery and business continuity plans.

2.1 Availability

Availability is a measure of the time that services are functioning correctly and accessible to users. Availability requirements are typically stated in terms of percent of time a service should be up and running, such as 99.99 percent.

2.2 Reliability

Reliability is a closely related concept to availability. Reliability is a measure of the probability that a service will continue to function under some load for a period of time. The level of reliability that a service can achieve is highly dependent on the availability of systems upon which it depends.

2.3 Scalability

Scalability is the ability of a service to adapt its infrastructure to the load on the system. When load decreases, some resources may be shut down. When load increases, resources can be added. Autoscalers and instance groups are often used to ensure scalability when using Compute Engine.

2.4 Durability

Durability is used to measure the likelihood that a stored object will be retrievable in the future. Cloud Storage has 99.999999999 percent (eleven 9s) durability guarantees, which means it is extremely unlikely that you will lose an object stored in Cloud Storage. Because of the math, as the number of objects increases, the likelihood that one of them is lost will increase.

Even Though all four concepts mentioned above go together in most cases, this blog mainly focuses on Availability and Durability.

Here are some more concepts of interest that may help understanding DR and BCP.

2.5 Load Balancing

Google Cloud offers server-side load balancing so you can distribute incoming traffic across multiple virtual machine (VM) instances. Load balancing provides the following benefits:

● Scale the application.

● Support heavy traffic

● Detect and automatically remove unhealthy VM instances using health checks. Instances that become healthy again are automatically re-added.

● Route traffic to the closest virtual machine

Google Cloud load balancing is a managed service, which means its components are redundant and highly available. If a load balancing component fails, it is restarted or replaced automatically and immediately.

2.6 Autoscaling

Compute Engine offers autoscaling to automatically add or remove VM instances from an instance group based on increases or decreases in load. Autoscaling lets our apps gracefully handle increases in traffic, and it reduces cost when the need for resources is lower. After you define the autoscaling policy, the autoscaler performs automatic scaling based on the measured load.

3 Locations, Regions and Zones

Google Cloud services are available in locations across North America, South America, Europe, Asia, and Australia. These locations are divided into regions and zones. You can choose where to locate our applications to meet the latency, availability, and durability requirements.

3.1 Locations

GCP resources are hosted in multiple locations worldwide. These locations are composed of regions and zones.

3.2 Regions

Regions are independent geographic areas that consist of zones. Locations within regions tend to have round-trip network latencies of under <1ms on the 95th percentile.

3.3 Zones

A zone is a deployment area for Google Cloud resources within a region. Zones should be considered a single failure domain within a region. To deploy fault-tolerant applications with high availability and help protect against unexpected failures, deploy the applications across multiple zones in a region.

To protect against the loss of an entire region due to natural disaster, you need to deploy all critical applications and mainly databases into multiple regions and where you can not deploy applications into multiple regions, you need to have a plan on how you can bring those applications on different regions.

As of March 2020, the GCP has following regions and zones within the USA.

● Region: us-central1

○ Zone: us-central1-a

○ Zone: us-central1-b

○ Zone: us-central1-c

○ Zone: us-central1-f

Location: Council Bluffs, Iowa, USA

● Region: us-west1

○ Zone: us-west1-a

○ Zone: us-west1-b

○ Zone: us-west1-c

Location: The Dalles, Oregon, USA

● Region: us-west3

○ Zone: us-west3-a

○ Zone: us-west3-b

○ Zone: us-west3-c

Location: Salt Lake City, Utah, USA

● Region: us-west2

○ Zone: us-west2-a

○ Zone: us-west2-b

○ Zone: us-west2-c

Location: Los Angeles, California, USA

● Region: us-east4

○ Zone: us-east4-a

○ Zone: us-east4-b

○ Zone: us-east4-c

Location: Ashburn, Northern Virginia, USA

● Region: us-east1

○ Zone: us-east1-b

○ Zone: us-east1-c

○ Zone: us-east1-d

Location: Council Bluffs, Iowa, USA

As per GCP, the following is the availability of a resource when implemented within a single zone, multiple zones (Regional) and in multiple regions (global).

Percentage Uptime Geography

99.9% Single Zone

99.99% Single region

99.999% Globally

The above availability can be translated into time as below:

Percent Uptime Downtime/Day Downtime/Week Downtime/Month

99.00 14.4 minutes 1.68 hours 7.31 hours

99.90 1.44 minutes 10.08 minutes 43.83 minutes

99.99 8.64 seconds 1.01 minutes 4.38 minutes

99.999 864 milliseconds 6.05 seconds 26.3 seconds

99.9999 86.4 milliseconds 604.8 milliseconds 2.63 seconds

4 GCP Resource types and disaster recovery

This section will explain how DR and BCP can be implemented with various GCP resource types.

4.1 Compute Engine Instances

A single VM instance can only be created in a single zone. High availability in Compute Engine is ensured by several different mechanisms and practices.

When creating VM instances, one of the following methods should be used to protect it from DR scenarios if that VM is critical for business continuity. The method you chose is based on many factors including criticality of the resource to business and SLAs involved in bringing back that resource in case of failures.

4.1.1 Hardware Redundancy and Live Migration

At the physical hardware level, the large number of physical servers in the GCP provide redundancy for hardware failures. If a physical server fails, others are available to replace it. Google also provides live migration, which moves VMs to other physical servers when there is a problem with a physical server or scheduled maintenance has to occur. Live migration is also used when network or power systems are down, security patches need to be applied, or configurations need to be modified.

There are certain limitations with Live migration. Currently, Live migration is not available for preemptible VMs, however, but preemptible VMs are not designed to be highly available. VMs with GPUs attached are not available to live migrate. Constraints on live migration may change in the future.

4.1.2 Managed Instance Groups

High availability also comes from the use of redundant VMs. Managed instance groups are the best way to create a cluster of VMs, all running the same services in the same configuration. A managed instance group uses an instance template to specify the configuration of each VM in the group. Instance templates specify machine type, boot disk image, and other VM configuration details.

If a VM in the instance group fails, another one will be created using the instance template.

Managed instance groups (MIGs) provide other features that help improve availability. A VM may be operating correctly, but the application running on the VM may not be functioning as expected. Instance groups can detect this using an application-specific health High Availability 53 check. If an application fails the health check, the managed instance group will create a new instance. This feature is known as auto-healing.

Managed instance groups use load balancing to distribute workload across instances. If an instance is not available, traffic will be routed to other servers in the instance group. Instance groups can be configured as regional instance groups. This distributes instances across multiple zones. If there is a failure in a zone, the application can continue to run in the other zones.

4.1.3 Multiple Regions and Global Load Balancing

Beyond the regional instance group level, you can further ensure high availability by running the application in multiple regions and using a global load balancer to distribute workload.

This would have the added advantage of allowing users to connect to an application instance in the closest region, which could reduce latency.

You would have the option of using the HTTP(S), SSL Proxy, or TCP Proxy load balancers for global load balancing.

Always use globally persisted storage or a database like Big Query to store critical data or information so that when you redeploy the VM Instance in different regions, you can attach the storage back. This will protect from loss of the data.

4.2 Kubernetes Engine

Kubernetes Engine is a managed Kubernetes service that is used for container orchestration. Kubernetes is designed to provide highly available containerized services.

VMs in a GKE Kubernetes cluster are members of a managed instance group, and so they have all of the high availability features described previously.

Kubernetes continually monitors the state of containers and pods. Pods are the smallest unit of deployment in Kubernetes; they usually have one container, but in some cases a pod may have two or more tightly coupled containers. If pods are not functioning correctly, they will be shut down and replaced. Kubernetes collects statistics, such as the number of desired pods and the number of available pods, which can be reported to Stackdriver.

By default, Kubernetes Engine creates a cluster in a single zone. To improve availability, you can create a regional cluster in GKE, the managed service that distributes the underlying VMs across multiple zones within a region. GKE replicates masters and nodes across zones. This provides continued availability in the event of a zone failure. The redundant masters allow the cluster is a managed Kubernetes service that is used for container orchestration. By default, Kubernetes Engine creates a cluster in a single zone.

If you are using any storage for stateful applications, always use globally persisted storage or a database like Big Query to store stateful information so that when you redeploy the Kubernetes cluster in different regions and can attach the storage back. This will protect from loss of the data.

4.3 App Engine and Cloud Functions

App Engine and Cloud Functions are fully managed compute services. Users of these services are not responsible for maintaining the availability of the computing resources. The Google Cloud Platform ensures the high availability of these services. Of course, App Engine and Cloud Functions applications and functions may fail and leave the application unavailable.

The DR and BCP can be maintained using CICD workflows for App Engine and Cloud functions. As they are not associated with data, these can be redeployed in just a matter of minutes.

4.4 Big Query and other distributed databases

BigQuery is by default deployed on USA multi-regional, so this will meet our DR and BCP requirements without any extra effort.

Other distributed databases like Spanner, BigTable and DataStore are globally distributed and no additional effort is required if you use these resources.

4.5 Cloud SQL

CloudSQL is a managed service for MySql and PostgreSql. CloudSQL supports Zonal and Multi-Zonal (Regional) implementations.

CloudSQL does not support multi-Regional at this point. This can be achieved building CloudSQL instances in multiple regions and by setting up replication. This can be also achieved by backups and log shipping onto global storage. Based on SLAs and criticality, one of the methods can be implemented if you need multi-regional redundancy.

4.6 Cloud Storage and Persistent disks

Persistent disks are faster compared to cloud storage. Cloud storage is good for Object storage needs like data lake. Persistent disks support block storage and even SSDs.

Both Cloud storage and Persistent disks support Zonal, Regional and Global high availability.

If the data or information stored is very critical and needs to be retained, use multi-regional. Use multi-regional for all production and critical data needs.

Use regional for non prod and non critical needs.

Use nearline or coldline for backups, archive data storage needs which are not accessed very frequently.

4.7 Pub/Sub

Pub/Sub is managed google service for streaming and available as a global service. No extra effort needed for DR and BCP requirements.

4.8 Dataflow

Dataflow resources can be created as Zonal or Regional. Dataflow pipelines are currently deployed through CICD pipelines. In case of regional failures, the resources can be redeployed using continuous deployment(CD) pipelines onto different regions.

It's important that you will have proper CI/CD pipelines developed and automated for all the dataflow jobs.

4.9 Dataproc

Dataproc resources currently support only Zonal. Like Dataflow, you need to have continuous deployment(CD) pipelines to deploy any dataproc resources in case of unavailability of a particular zone/region.

Use multi-regional persistent disks or cloud storage if you need to retain any data.

4.10 Composer

Composer supports only Zonal currently. DR and BCP requirements can be met using proper fully tested CICD pipelines.

5 GCP DR and BCP matrix

The following table explains how each resource can be made to support DR and BCP requirements when required.

6 Considerations and Assumptions

Even though this blog touches some network aspects, it doesn't fully cover all network requirements like having redundant networks between on-premise and GCP, having multiple interconnects if you are using vendor provided interconnects and so on.

This blog also doesn’t fully cover OLTP database needs and other applications beyond the scope of the data lake as I don't have much insight into those projects.

This blog focuses on GCP regions as different geographically separated data centers and assumes GCP is sufficient for our DR and BCP needs. It does not cover the entire GCP shutdown for whatever the reasons, that's where the multi-cloud approach comes into picture.

7 Pricing and other costs

As far as high availability is concerned, global or multi-regional provide far superior coverage compared to Regional and Zonal. Regional provides much high availability compared to zonal.

But from a cost perspective, global or multi-regional is far more expensive than regional and zonal. Regional is more expensive than zonal. Sometimes, costs might be more than doubled.

For example, to make VM instances redundant, you need to use 2 instances at least and you need a managed instance group(MIG) service and also there will be network changes to keep both instances in sync. The costs will be more than double in this case.

But for Kubernetes, it only adds a master server to make it more redundant in case of regional implementation, the cost will go up marginally but a lot less than two times.

But in case of managed services like BigQuery, Pub/Sub, App Engine, it will not incur any additional costs.

For Cloud storage the costs are different, if you store data in multi-regional, regional, coldline and nearline tiers. Multi-regional is more expensive and coldline and nearline are cheaper. Sometimes, the cost difference is huge.

When you design and build resources, you have to keep pricing, SLAs and DR requirements in mind. Some solutions may satisfy the needs and are also cheaper.

Besides pricing and other costs, there are also latencies that need to be kept in mind. Intra-zonal and intra-regional trips add a few milliseconds latencies to the application response times. Different layers of applications like front-end, back-end and database need to be aligned properly for better performance and to reduce network costs.

Cloud Operations and Monitoring on GCP

Thu, 23 Apr 2020 00:28:19 GMT

How to build operations and support on google cloud?

1 Introduction

There are 2 terms that mainly go with any system; Reliability and Availability. They both are closely related. Reliability is how the system going to provide the service continuously. Reliability is a probability, specifically, the probability that a system will be able to process some specified workload for some period of time. Availability is a measure of the percentage of time that a system is available and able to perform well under specified workload. The difference between reliability and availability is important to keep in mind when thinking about metrics and service-level agreements (SLAs).

This blog will focus on reliability from a reliability engineering perspective. That is, how do we design, monitor, and maintain services so that they are reliable and available.

2 Improving Reliability with Stackdriver

2.1 Monitoring

This service is used to help understand performance and utilization of applications and resources. Monitoring collects the basic data about a wide array of resource characteristics and their performance over time.

2.2 Logging

Logging collects messages about specific events related to application and resource operations. Logging to Stackdriver is automatic for most of the GCP resources like Big query, dataflow and so on, but requires installing its agent and setup on some of the resources like the compute engine.

2.3 Alerting

This service is used to notify responsible parties about issues with applications or infrastructure that need attention. Together, these three components of Stackdriver provide a high degree of observability into cloud systems.

Alerting evaluates metric data to determine when some predefined conditions are met, such as high CPU utilization for an extended period of time.

3 Stackdriver Logging

Stackdriver Logging is a centralized log management service. Logs are collections of messages that describe events in a system. Unlike metrics that are collected at regular intervals, log messages are written only when a particular type of event occurs. Some examples of log messages are:

● New account is created on VM and root privileges granted on the computer engine.

● Big Query connection failed because of authentication issues.

● Dataflow job started.

● Dataflow job succeeded/failed.

Stackdriver provides the ability to store, search, analyze, and monitor log messages from a variety of applications and cloud resources. An important feature of Stackdriver Logging is that it can store logs from virtually any application or resource, including GCP resources, other cloud resources, or applications running on premises. The Stackdriver Logging API accepts log messages from any source.

Logs can be easily exported to BigQuery for more structured SQL-based analysis and conducting trend analyses for logs extending past 30 days.

Applications and resources can create a large volume of log data. Stackdriver Logging will retain log messages for 30 days. If we would like to keep logs for longer, then we need to export them to Cloud Storage or BigQuery.

Logs can also be streamed to Cloud Pub/Sub if you would like to use third-party tools to perform near real-time operations on the log data.

3.1 Types of logging

● System Events

● Application events

● Service events

● Security Audit events

● Resource related events

3.2 Kinds of messages

● Performance and uptime

● Informational

● Debugging related

● Failures and Errors

● Access related information and exceptions

4 Monitoring with Stackdriver

Monitoring is the practice of collecting measurements of key aspects of infrastructure and applications. Examples include average CPU utilization over the past minute, the number of bytes written to a network interface, and the maximum memory utilization over the past hour. These measurements, which are known as metrics, are made repeatedly over time and constitute a time series of measurements.

4.1 Metrics

Metrics have a particular pattern that includes some kind of property of an entity, a time range, and a numeric value.

GCP has defined metrics for a wide range of entities, including the following:

■ GCP services, such as BigQuery, Cloud Storage, and Compute Engine.

■ Operating system and application metrics which are collected by Stackdriver agents that run on VMs.

■ External metrics including metrics defined in Prometheus, a popular open source monitoring tool.

In addition to the metric name, value, and time range, metrics can have labels associated with them. This is useful when querying or filtering resources that you are interested in monitoring.

4.2 Time Series

A time series is a set of metrics recorded with a time stamp. Often, metrics are collected at a specific interval, such as every second or every minute. A time series is associated with a monitored entity.

4.3 Dashboards

Dashboards are visual displays of time series.

Dashboards are customized by users to show data that helps monitor and meet service level objectives or diagnose problems with a particular service. Dashboards are especially useful for determining correlated failures.

Stackdriver provides a predefined set of os dashboards for the services being used. Users can also create a custom dashboard and add charts created with the Metrics Explorer to it.

5 Alerting with Stackdriver

Alerting is the process of monitoring metrics and sending notifications when some custom-defined conditions are met. The goal of alerting is to notify someone when there is an incident or condition that cannot be automatically remediated and that puts service level objectives at risk.

5.1 Policies, Conditions, and Notifications

Alerting policies are sets of conditions, notification specifications, and selection criteria for determining resources to monitor.

Conditions are rules that determine when a resource is in an unhealthy state. Alerting users can determine what constitutes an unhealthy state for their resources. Determining what is healthy and what is unhealthy is not always well defined. For example, we may want to run our instances with the highest average CPU utilization possible, but we also want to have some spare capacity for short-term spikes in workload. (For sustained increases in workload, autoscaling should be used to add instances to handle the persistent extra load.) Finding the optimal threshold may take some experimentation. If the threshold is too high, we may find that performance degrades, and we are not notified. On the other hand, if the threshold is too low, we may receive notifications for incidents that do not warrant our intervention. These are known as false alerts. It is important to keep false alerts to a minimum. Otherwise, engineers may suffer “alert fatigue,” in which case there are so many unnecessary alerts that engineers are less likely to pay attention to them.

5.2 Notification methods

The following notification methods are readily available with GCP:

● Mobile Devices (via Cloud Mobile App)

● PagerDuty Services/Sync

● Slack (individual or channel)

● Webhooks (https)

● Email (groups/individual)

● SMS

6 Data lake monitoring

6.1 Proactive monitoring

It is important that systems are being monitored proactively and identify or forecast the issues before they actually surface. Here are some tools that help in proactive monitoring:

● Dashboards

Some of the dashboards that might require are:

○ Daily, weekly, monthly time series Big Query performance dashboard.

○ Failures over time for Big Query, Dataflow and Dataproc.

○ Dataflow performance metrics over time by job, group and project.

○ Dataproc performance metrics over time job, group and project.

● Metrics Explorer

You can set up and view on-demand custom metrics visualizations. Metrics can also be subjected to custom filtering, grouping and aggregation options.

● Uptime checks

We can check service or application uptimes over time intervals.

● Baseline metrics

Collect baseline metrics on resources we need to monitor and compare it with current metrics. Adjust baseline metrics from time to time.

● Alerts

Setup some alerts to help monitoring GCP resources proactively. Example of setting up alerts on performance of BigQuery queries. These are just kind of informational and warning kind of stuff.

It is also a good idea to create baseline metrics (and update them time-to-time) and compare current metrics with baseline metrics gives some idea how systems are performing currently and overtime.

6.2 Reactive monitoring

Reactive monitoring is about responding to the incidents and failures that have already occurred. The alerts and notifications for certain performance conditions and failures play an important role in improving system uptime.

6.2.1 Groups

Groups are a combination of resources that can be monitored as a unit. For example, all dataflow jobs that belong to Athena project and based on some criticality.

Groups simplify setting up alerts on various resources.

6.2.2 Big Query

Since BigQuery is a service, the service and infrastructure level issues are internal to GCP. We will not be notified over those issues. There may be outage alerts if any GCP services are down and not available for a particular period of time. We are mainly concerned about the below issues with BigQuery.

6.2.3 Data Flow

Dataflow jobs can be monitored using its own UI for troubleshooting individual jobs. Dataflow also adds messages to stackdriver like when job status, when started, failed and so on. This document only covers alerting from stackdriver.

We can also setup notifications, custom messages from within dataflow jobs which are not covered here.

There are other metrics like vCpus used, number of nodes and so on which are helpful for debugging but no alerting is required on those.

6.2.4 Dataproc

There are a bunch of metrics available for monitoring dataproc. Some of them are HDFS Capacity, running jobs, job duration, connections, CPU Load, Operations running, Memory Usage, Errors, Failed Jobs and Failed Requests.

6.2.5 Composer

Stackdriver provides metrics for monitoring Composer under two Resource Types, Cloud Composer Environment and Cloud Composer Workflows. Some metrics do overlap between the two resource types. The main metrics to monitor Composer include Tasks, Tasks Duration, Workflow duration and Runs. There are also some metrics to understand the overall health of the environments.

6.2.6 GCS Buckets

There are quite few metrics available with GCP storage. Some of them are Object count, bytes received, requests and so on.

We don't need to set up any direct alerts on the GCP buckets. But when there are any errors with GCP buckets, it will be alerted indirectly like dataflow and so on.

6.2.7 Compute Engine

Here are some sample alerts you can setup on compute engine instance.

6.2.8 Quotas and Limits

The following metrics are available under resource type “Consumer Quota” for monitoring “Quotas and Limits” for Google cloud resources like BigQuery, Cloud storage and so on.

Sample screen on how to set up alerts on Quotas and Limits:

6.3 Troubleshooting (Log Search)

It's used to debug and troubleshoot issues from time to time. For example, Tableau group report issue connecting to Big Query. It could be an issue with Big Query itself or network issue or security issue.

7 Logging Pipeline messages

We can use the Dataflow SDK's built-in logging infrastructure to log information during pipeline's execution. The following url has more detailed information and coding samples.

https://cloud.google.com/dataflow/docs/guides/logging

8 Additional Monitoring Tools

The following are some of the tools available on GCP for troubleshooting applications and services deeper. At this time, these tools may not be relevant to our data lake monitoring.

8.1 Stackdriver debugger

The Stackdriver debugger allows us to inspect and analyze the state of the particular applications in real time without having to stop the application. With the debugger, we will be able to capture the call stack and variables without slowing down the application; this is particularly helpful if we want to debug or trace the code and understand its behavior. Stackdriver does this by capturing the application state by means of debug snapshots, which add less than 10 ms to the latency:

8.2 Stackdriver profiler

The Stackdriver profiler helps gather CPU usage and memory allocation information from the applications. This is different than Stackdriver monitoring, because with the Stackdriver profiler, we can tie the CPU usage and memory allocation attributes back to the application's source code. This helps us identify parts of your application that consume the most resources and also allows us to check the performance of your code.

8.3 Stackdriver Trace

Stackdriver Trace is a distributed tracing system that collects latency data from applications, Google App Engine, and HTTP(s) load balancers, and displays it in near-real time in the GCP console. For applications to be able to use this feature, they need to have the appropriate code in place with the Stackdriver Trace SDKs, but this is not needed if our application is being deployed in Google App Engine. Stackdriver Trace greatly enhances the understanding of how the application behaves. Things like time to handle requests and complete RPC calls can be viewed in real time using Stackdriver Trace.

9 Cloud Audit Logs

9.1 Admin Activity audit logs

Admin Activity audit logs contain log entries for API calls or other administrative actions that modify the configuration or metadata of resources. For example, these logs record when users create VM instances or change Cloud Identity and Access Management permissions.

9.2 Data Access audit logs

Data Access audit logs contain API calls that read the configuration or metadata of resources, as well as user-driven API calls that create, modify, or read user-provided resource data.

This feature is disabled by default and must be enabled to be utilized. Due to the volume of logs that can be generated, all publicly accessible objects in the project are exempt from this audit logging.

9.3 System Event audit logs

System Event audit logs contain log entries for Google Cloud administrative actions that modify the configuration of resources. System Event audit logs are generated by Google systems; they are not driven by direct user action.

9.4 Access Transparency logs

Viewing query text, table names, dataset names, and dataset access control lists might not generate Access Transparency logs; this access pathway gives read-only access. Viewing query results and table or dataset data will still generate Access Transparency logs. In the same way, viewing message payloads will still generate Access Transparency logs for PubSub topics. Viewing topic names, subscription names, message attributes, and timestamps might not generate Access Transparency logs; this pathway gives read-only access.

Only people who are allowed to view this log information can be granted roles/logging.privateLogViewer (Private Logs Viewer).

10 Appendix A - Dashboards

10.1 Performance over time

We need to create dashboards that will show us how Big Query queries, Dataflow and Dataproc jobs are performing.

Users should be able to choose the time period and compare the metrics.

11 APPENDIX III - Notifications and Service system Integration

The following are the notification methods that are in-built and readily available in GCP.

● Mobile Devices (via Cloud Mobile App)

● PagerDuty Services/Sync

● Slack (individual or channel)

● Webhooks (https)

● Email (groups/individual)

● SMS

You can integrate the ticketing systems like ServiceNow to create alerts and support specific tickets and to monitor the resolution status. Whenever an alert happens, The GCP notification system needs to call serviceNow API and post alert specific information.

This is can be achieved using the following options available on the GCP:

Option I : using webhooks

Currently GCP allows only static webhooks with basic and token-based authentication. This is very basic and may not satisfy complex needs.

Option II: using pub/sub

This is more flexible and allows any type of notification methods like POST API calls and so on. This also allows multiple notification methods in a single call. The messages also can be logged into BigQuery for later analysis. But this requires little more effort and incur additional costs.

Data Governance for data lakes on Cloud

183:766491093 (Srinivasa Rao) — Wed, 22 Apr 2020 01:06:20 GMT

Implement data governance with cloud native tools

Data Governance on cloud is a vast subject. It involves lot of things like security and IAM, Data cataloging, data discovery, data Lineage and auditing.

Security

Covers overall security and IAM, Encryption, Data Access controls and related stuff. Please visit my blog for detailed information and implementation on cloud.

https://www.unifieddatascience.com/security-architecture-for-google-cloud-datalakes

Data Cataloging and Metadata

It revolves around various metadata including technical, business and data pipeline (ETL, dataflow) metadata. Please refer to my blog for detailed information and how to implement it on Cloud.

https://www.unifieddatascience.com/data-cataloging-metadata-on-cloud

Data Discovery

It is part of the data cataloging which explained in the last section.

Auditing

It is important to audit is consuming and accessing the data stored in the data lakes, which is another critical part of the data governance.

Data Lineage

There is no tool that can capture data lineage at various levels. Some of the Data lineage can be tracked through data cataloging and other lineage information can be tracked through few dedicated columns within actual tables. Most of the Big Data databases support complex column type, it can be tracked easily without much complexity. The following are some examples of data lineage information that can be tracked through separate columns within each table wherever required.

1. Data last updated/created (add last updated and create timestamp to each row).

2. Who updated the data (data pipeline, job name, username and so on - Use Map or Struct or JSON column type)?

3. How data was modified or added (storing update history where required - Use Map or Struct or JSON column type).

Data Quality and MDM

Master data contains all of your business master data and can be stored in a separate dataset. This data will be shared among all other projects/datasets.

This will help you to avoid duplicating master data thus reducing manageability. This will also provide a single source of truth so that different projects don't show different values for the same.

As this data is very critical, we will follow type 2 slowly changing dimensional approach which will be explained my other blog in detail.

https://www.unifieddatascience.com/data-modeling-techniques-for-modern-data-warehousing

There are lot of MDM tools available to manage master data more appropriately but for moderate use cases, you can store this using database you are using.

MDM also deals with central master data quality and how to maintain it during different life cycles of the master data.

There are several data governance tools available in the market like Allation, Collibra, Informatica, Apache Atlas, Alteryx and so on. When it comes to Cloud, my experience is it’s better to use cloud native tools mentioned above should be suffice for data lakes on cloud/

Security Architecture for google cloud datalakes

Tue, 21 Apr 2020 19:16:56 GMT

Security design for Cloud data lakes

1 Introduction

This blog explains overall security architecture on GCP briefly and puts together the data lake security design and implementation steps.

2 Layers of Security

● Infrastructure and cloud platform security

● Identity access management

● Key management service

● Cloud security scanner

● Security keys

2.1 Infrastructure and cloud platform security

Google or any other cloud provider has security built through layers, from data center security all the way to application and information management.

The lowest layer is the hardware infrastructure security, which includes physical security, hardware security, and machine boot stack security. Google designs and fully owns and manages its data centers, and access to these data centers is highly restricted.

2.2 Identity and access management (IAM)

IAM allows us to define users and roles and help control user access to GCP resources. GCP offers Cloud IAM, which allows us to grant granular access to users for specific GCP resources based on the least privilege security principle.

Cloud IAM is made up of members to whom access is granted. The following diagram shows the different kinds of member types and also roles, which are collections of permissions. When a member is authenticated and makes a request, Cloud IAM uses roles to assess whether that member is allowed to perform an operation on a resource:

The above diagram is courtesy to GCP.

2.2.1 Types of member accounts

2.2.1.1 Google Account

Any person who interacts with GCP by a Google account. This person can be a developer, an administrator, or an end user with access to GCP.

2.2.1.2 Service Account

A service account is an account that is associated with an application. Code that runs on your GCP account runs under a service account. Multiple service accounts can be created and associated with different parts of your application. Service accounts help create granular access and permissions.

2.2.1.3 Google Group

A Google group is a collection of Google accounts and service accounts. Google groups makes it easier to apply an access policy to a collection of users. Groups also make it easy to add and remove members.

2.2.1.4 G Suite Domain

Google offers organizations a G Suite Account, which offers email, calendar, docs, drive, and other enterprise services. A G Suite domain represents a virtual group of all Google accounts created in an organization's G Suite account.

2.2.1.5 Cloud Identity Domain

The primary difference between a Cloud Identity Domain and a G Suite domain is that all users in the Cloud Identity Domain do not have access to G Suite applications and features.

2.2.2 Elements of IAM

The primary elements of IAM are as follows:

● Identities and groups

● Resources

● Permissions

● Roles

● Policies

2.2.2.1 Identities and groups

Identities and groups are entities that are used to grant access permissions to users.

2.2.2.1.1 Identities

An identity is an entity that represents a person or other agent that performs actions on a GCP resource. Identities are sometimes called members. There are several kinds of identities:

● Google account

● Service account

● Cloud Identity domain

2.2.2.1.2 Groups

Groups are useful for assigning permissions to sets of users. When a user is added to a group, that user acquires the permissions granted to the group. Similarly, when a user is removed from the group, they no longer receive permissions from the group. Google Groups do not have login credentials, and therefore they cannot be used as an identity.

2.2.2.2 Roles

A role is a collection of permissions that determine what operations are allowed on any resource. Roles are granted to users, and all permissions that role contains are applied to that user. Permissions are represented in the form of <service>.<resource>.<verb>, for example, compute.instances.delete. Resource here is a GCP resource such as projects, compute engine instances, cloud storage buckets, and so on.

There are three types of roles:

● Predefined

● Primitive (viewer, writer and so on)

● Custom

2.2.2.3 Resources

Resources are entities that exist in the Google Cloud platform and can be accessed by users. Resources is a broad category that essentially includes anything that we can create in GCP. Resources include the following:

● Projects

● Virtual machines

● App Engine applications

● Cloud Storage buckets

● Pub/Sub topics

● Big Query dataset or table

Google has defined a set of permissions associated with each kind of resource. Permissions vary according to the functionality of the resource.

2.2.2.4 Permissions

A permission is a grant to perform some action on a resource. Permissions vary by the type of resource with which they are associated. Storage resources will have permissions related to creating, listing, and deleting data. For example, a user with the bigquery.datasets .create permission can create tables in BigQuery. Cloud Pub/Sub has a permission called pubsub.subscriptions.consume, which allows users to read from a Cloud Pub/Sub topic.

Here are some examples of other permissions used by Compute Engine:

● compute.instances.get

● compute.networks.use

● compute.securityPolicies.list

Here are some permissions used with Cloud Storage:

● resourcemanager.projects.get

● resourcemanager.projects.list

● storage.objects.create

2.2.2.5 IAM policy

An IAM policy is a collection of statements that define the type of access a user gets. A policy is assigned to a resource and is used to govern access to that resource. An IAM policy can be set to any level in a resource hierarchy. You can set the policy at the organization level, the folder level, the project level, or at the resource level. Policies are inherited from parents so if we set a policy at the organization level, all the projects will inherit that policy, which in turn is inherited by all resources in those projects.

In addition to granting roles to identities, we can associate a set of roles and permissions with resources by using policies. A policy is a set of statements that define a combination of users and the roles. This combination of users (or members as they are sometimes called) and a role is called a binding. Policies are specified using JSON.

● A user group can have one or more users as members.

● A user can belong to one or more user groups.

● A user can be assigned one or more roles.

● A role can be assigned to one or more users.

● A role can access one or more resources through permission control.

● One permission is granted to one or more roles.

● One resource can only have one corresponding permission for each role.

3 Teams and Responsibilities

Here are some ideas on how GCP is controlled by various teams. Based on the responsibilities, various teams are responsible for managing security around those functions and also securing GCP resources and data within their parameters.

4 Security best practices

The following are some good security practices in brief that need to be kept in mind while implementing any security at any level.

4.1 Least privilege

Least privilege is the practice of granting only the minimal set of permissions needed to perform a duty. IAM roles and permissions are fine-grained and enable the practice of least privilege.

4.2 Auditing

Auditing is basically reviewing what has happened on our system. In the case of Google Cloud, there are a number of sources of logging information that can provide background details on what events occurred on your system and who executed those actions.

4.3 Policy Management

● Set organization-level Cloud IAM policies to grant access to all projects in your organization.

● Grant roles to a Google group instead of individual users when possible. It is easier to add members to and remove members from a Google group instead of updating a Cloud IAM policy to add or remove users.

● If we need to grant multiple roles to allow a particular task, create a Google group, grant the roles to that group, and then add users to that group.

4.4 Separation of duties

Separation of duties (SoD) is the practice of limiting the responsibilities of a single individual in order to prevent the person from successfully acting alone in a way detrimental to the organization.

4.5 Defense in Depth

Defense in depth is the practice of using more than one security control to protect resources and data. For example, to prevent unauthorized access to a database, a user attempting to read the data may need to authenticate to the database and must be executing the request from an IP address that is allowed by firewall rules.

5 Encryption

5.1 Data at Rest encryption

Encryption at rest is key for securing our data. By default, Cloud Storage will always encrypt our data on the server side before it is written to disk. There are 3 options available for server-side encryption:

● Google-Managed Encryption Keys: This is where Cloud Storage will manage encryption keys on behalf of the customer, with no need for further setup. This also does key rotation and preferred way of doing it.

● Customer-Supplied Encryption Keys (CSEKs): This is where the customer creates and manages their own encryption keys.

● Customer-Managed Encryption Keys (CMEKs): This is where the customer generates and manages their encryption keys using GCP's Key Management Service (KMS).

It is recommended to use default Google-Managed Encryption on all our buckets in normal cases. Big Query encryption at rest is managed by service itself.

There is also the client-side encryption option where encryption occurs before data is sent to Cloud Storage and additional encryption takes place at the server side.

5.2 Data in Motion encryption

As we are using google proved services, how the data moved internally is being managed by google and we don't have control over it.

But we need to implement encryption in the following cases:

● Use TLS/SSL connections between front-end (like Tableau) and Big Query.

● All the files that are transferred between GCP and on-premise need to be encrypted. Use SFTP instead of FTP.

5.3 Column level encryption

Certain columns that have personal identifiable information may need to be encrypted within the Big Query tables. This is on top of data at rest encryption.

6 Data Level Security

As we are planning to use only cloud stores and Big Query for our data lake needs, this section mainly focuses around them.

6.1 Process

Initially, when a data lake project is created in each environment, we will create all required groups with required people in it along with roles assigned. After that whenever we need to make changes like creating groups, adding people to groups and granting permissions to groups, we need a serviceNow (or similar tool) request with approvals from proper authority.

All the individuals get access through proper groups rather than direct access. The following process diagram explains the process that needs to be followed to get access to the data lake resources.

6.2 Cloud store

Access to Google Cloud Storage is secured with IAM. The following are the list of predefined roles and their details:

● Storage Object Creator: Has rights to create objects but does not give permissions to view, delete, or overwrite objects

● Storage Object Viewer: Has rights to view objects and their metadata, but not the ACL, and has rights to list the objects in a bucket

● Storage Object Admin: Has full control over objects and can create, view, and delete objects.

● Storage Admin: Has full control over buckets and objects.

Cloud Storage also offers security via ACLs. Below are the permissions available and their details:

● Reader: This can be applied to a bucket or an object. It has the right to list a bucket's contents and read bucket metadata. It also has the right to download an object's data.

● Writer: This can be applied to a bucket only and has the rights to list, create, overwrite, and delete objects in a bucket.

● Owner: This can be applied to a bucket or an object. It grants reader and writer permissions to the bucket and grants reader access to an object.

6.3 Big Query

Access to BigQuery is secured through IAM. The following are the predefined roles:

● BigQuery User: This has rights to run jobs within the project. It can also create new datasets. Most individuals in an organization should be a user.

● BigQuery Job User: This has rights to run jobs within the project.

● BigQuery Read Session User: This has rights to create and read sessions within the project via the BigQuery storage API.

● BigQuery Data Viewer: This has rights to read the dataset metadata and list tables in the dataset. It can also read data and metadata from the dataset tables.

● BigQuery Metadata Viewer: This has rights to list all datasets and read metadata for all datasets in the project. It can also list all tables and views and read metadata for all tables and views in the project.

● BigQuery Data Editor: This has rights to read the dataset metadata and to list tables in the dataset. It can create, update, get, and delete the dataset tables.

● BigQuery Data Owner: This has rights to read, update, and delete the dataset. It can also create, update, get and delete the dataset tables.

● BigQuery Admin: This has rights to manage all resources within the project.

7 Data Security Implementation

The following flowchart explains how access requests can be fulfilled through an automated process.

Google Security APIs are available here:

https://cloud.google.com/security-command-center/docs/reference/rest/

Programmatically setting up access:

https://cloud.google.com/security-command-center/docs/how-to-programmatic-access

8 Logging, Monitoring and Auditing

It is very important to log all security related events like who is accessing what resources, failed attempts and so on into a central repository like google stack driver and monitor it on regular purpose.

This section will be explained in more detail in the Cloud operations blog.

9 Other Google Security services

9.1 Data Loss Protection

The data loss prevention (DLP) API provides a programmatic way to classify 90+ types of data. For example, if we have a spreadsheet with financial and/or personal information, parsing this data with the DLP API masks and hides all this sensitive information. With this API, we can easily classify and redact sensitive data.

9.2 Cloud Security Scanner

Cloud security scanner is built to identify vulnerabilities in our Google App Engine applications. The scanner crawls through our application and can attempt to try multiple user inputs in order to detect vulnerabilities. The scanner can detect a range of vulnerabilities, such as XSS, flash injection, mixed content, clear text passwords, and use of JavaScript libraries.

Cloud security scanner can only be used with the Google App Engine standard environment and compute engine.

9.3 Key Management Service (KMS)

Spark performance tuning guidelines

183:766491093 (Srinivasa Rao) — Tue, 21 Apr 2020 01:20:55 GMT

Spark tuning for high performance

1 Introduction

This document will outline various spark performance tuning guidelines and explain in detail how to configure them while running spark jobs. This process guarantees that the Spark has optimal performance and prevents resource bottlenecking. Effective changes are made to each property and settings, to ensure the correct usage of resources based on system-specific setup. Apache Spark has in-memory computation in nature. As a result resources in the cluster (CPU, memory etc.) may get bottlenecked.

2 Spark application execution modes

Spark Jobs can run in 3 modes, local, client and cluster modes:

2.1 Local Mode

Local mode runs both driver and executors on a single node. In this mode, the partitions are processed by multiple threads in parallel. The number of threads can be controlled by the user while submitting a job. This is used only for a single node cluster for learning spark purposes.

Eg:

$ spark-submit --master local <other parameters>

2.2 Client mode

In the client mode, the driver process runs on the client node (that is, the edge or gateway node) on which the job was submitted. The client node provides resources, such as memory, CPU, and disk space to the driver program, but the executors run on the cluster nodes and they are maintained by the cluster manager, such as YARN.

E1, E2 and E3 are executors.

Eg:

$ spark-submit \

--master yarn

--deploy-mode client

--<other parameters>

2.3 Cluster mode

Cluster mode is similar to client mode, except that the driver process runs on one of the cluster workers machines, and the cluster manager is responsible for both driver and executor processes. This gives an advantage of running multiple applications at the same time because cluster managers will distribute the driver load across the cluster.

This mode is the most common and recommended mode for running the Spark applications in production environments.

E1, E2 and E3 are executors.

Eg:

$ spark-submit \

--master yarn

--deploy-mode cluster

--<other parameters>

Always run production spark jobs using Yarn cluster mode.

3 Spark execution options

The following are some resource allocation parameters that can be set while submitting spark jobs from various clients like spark-submit, Talend and so on. If you don’t pass any of these parameters, spark will take it from spark-defaults.conf, which will have minimal resources allocated most of the time, that may not be suitable to all spark jobs.

These parameters need to be set at the job level while submitting a job because different jobs need different kinds and amount of resources based on its complexity and the number of parallel jobs running at that time and so on. There is no single standard which suits all the purposes. The following are some of the things that you should take into consideration while setting up these parameters.

Number of concurrent jobs run during that time (or need to run at that time).

Resources allocated to Spark Yarn queue and there by amount of resources available for Spark.

Amount of the data Spark job processes.

Whether the Spark job uses a wide transformation like reduceByKey, aggregateBy and so on.

Amount of shuffle happening.

Whether it is caching any data to memory.

Even though there are a lot more parameters that are available with spark submit, this section will explain some of them that are critical and have considerable impact on Spark applications.

3.1 Spark driver memory

Amount of memory to use for the driver process. Default is 1g. When you use accumulators, or collect a lot of data like using collect() or take(N) actions, you might need to increase it to an appropriate size.

spark.driver.memory (from configuration file) or --driver-memory (from spark-submit)

3.2 Spark driver core

Number of cores used by the driver process. Default is 1 core. It's a good idea to allocate 2 to 4 based on cluster size and complexity of the code.

Spark.driver.cores (from configuration file) --driver-cores (from spark-submit)

3.3 Spark number of executors

This is one of the critical parameters that needs attention and may need some calculations. This is the number of executors spark can initiate when submitting a spark job. This parameter is for the cluster as a whole and not per the node.

There are multiple ways you can calculate how many executors you need to set. In the Fat executors method, you set one executor per node and in the Tiny executor approach, you set one executor each CPU core. But my experience shows it is somewhere in between. Allocate a few cores 2 to 5 for each executor in general.

For example, if you have 30 nodes in the cluster (assume each node has 8 cores allocated for the yarn queue that you are using to submit spark jobs), setting 30 executors will have 1 executor per node with 8 cores for each executor. Setting 240 executors will have 8 executors per node with 1 core allocated to each executor.

Number of executors determine parallelism. Tasks run within executors. Parallelism comes with some overheads and sometimes, overheads outweigh benefits of the parallelism. Especially, your job does a lot of shuffling or uses wide transformations, it's a good idea to go with smaller partitions. It also depends on the complexity of the spark code and number of concurrent jobs running at the same time.

I would generally go with at least 2 to 5 cores for each executor based on availability and also leave 1 to 2 cores not allocated leaving for other tasks.

Below are some examples on how to calculate the number of executors:

Number of Nodes: 30

Number of cores on each node: 32

Amount of RAM: 64GB

Spark Yarn queue: 30% which means approximately 10 cores and 21GB ram available on each node. Total of 300 (30 nodes * 10) cores and 630GB (30 * 21g) of memory available.

Scenario I

Number of parallel spark jobs = 1

Number of cores per executor = 4

Number of executors = (300 -60)/4 = 60 (leaving 2 cores on each node for other purposes)

That will make 2 executors for each node.

Scenario II

Number of parallel spark jobs = 3

Number of cores per executor = 4

Number of executors for each job = ((300 -60)/4) = 60/3 = 20 (leaving 2 cores unused on each node for other purposes). This will make up to 2 executors per node but running 3 jobs instead of one job as in scenario I. Each spark job gets a total of 20 executors.

Scenario III

Number of parallel spark jobs = 3

Number of cores per executor = 3

Number of executors for each job = ((300 -30)/3) = 90/3 = 30 (leaving 1 cores unused on each node for other purposes). In this case 3 executors on each node but 3 jobs running so one executor on each node will be allocated to each job. Each spark job gets a total of 30 executors.

In the above example (Scenario II and III), assuming all 3 spark jobs require similar resources and have similar complexity. You can allocate different cores and different numbers of executors for each spark job as long as the total will be within allocated resources for that queue. In case if the total exceeds the resources available, the consecutive spark jobs will be waiting for the resources to be released.

You can use resources like Spark UI and Spark application logs to see how the resources utilized and tweak back and forth.

Spark.executor.instances (from configuration file) or –num-executors (from spark-submit)

3.4 Spark executor memory

Amount of the memory allocated for each executor process. Default is 1g. There are many things you need to consider while allocating this like how much data you are processing, any shuffling activity and so on.

For example if you have 320 GB available on cluster ( 30 nodes * 10GB) for the Yarn queue you are using to run spark jobs and you are determined to run on 30 executors based on the above calculations, you may set this parameter at 10g (appx 320/30) assuming only one spark job running at that time.

Spark.executor.memory (from configuration file) or –executor-memory (from spark-submit)

3.5 Spark executor cores

Number of cores allocated for each executor. Default is one. Please see the above examples on how to calculate this and come to the right number.

--executor-cores(from configuration file) or spark.executor.cores (from spark-submit)

eg:

spark-submit <other parameters>

--master yarn

--deploy-mode cluster

--executor-memory 10g

--executor-cores 3

--num-executors 30

--driver-memory 5g

--driver-cores 2

4 Compression

Compressing data at various levels improves overall performance. It reduces storage space, speeds up IO, reduces network costs, improves memory usage. But it comes with some computation overheads.

Spark can transparently read files that are compressed using LZ4, Snappy and GZIP codecs.

Here are some properties that can be set within spark code and also at the cluster level which will improve the performance of spark jobs.

Spark.sql.inMemoryColumnarStorage.compressed = true

Spark SQL will automatically select a compression codec for each column based on statistics of the data

Spark.io.compression.codec=snappy

Specifies the codec used to compress internal data such as RDD partitions, event log, broadcast variables and shuffle outputs.

Spark.broadcast.compress=true

Whether to compress broadcast variables before sending them. Generally a good idea.

MemoryColumnarStorage.batchSize=10000

The size of batches for columnar caching. Larger batch sizes can improve memory utilization and compression, but risk OOMs when caching data.

5 Parallelism and Partitions

Spark automatically sets the number of “map” tasks to run each input file according to its size. For distributed “reduce” operations such as groupByKey and reduceByKey, it uses RDDs number of partitions. Sometimes, you may need to use repartition or coalesce to change the number of partitions.

You can also set spark.default.parallelism parameter to change the default. In general, we recommend 2-3 tasks per CPU core in your cluster. 2 to 4 partitions per CPU core is highly recommended.

It is important to know tasks and executors are different. Tasks are the small processes run within the executor. Number of executors and number of tasks determine total parallelism.

You can also set the number of partitions within the transformations.

val rdd2 = rdd1.reduceByKey(_ + _, numPartitions = X)

For Narrow transformation like “read”, the number of partitions is equal to (size of the file / how much data you want to process in each partition).

For the “reduce” operations where there is a lot of shuffling, keep multiplying X by 1.5 times with the number of original partitions when read from file (map tasks) until performance stops improving.

6 Storage format

Each file format has pros and cons and each output type needs to support a unique set of use-cases. For each output type, we chose the file format that maximizes the pros and minimizes the cons.

The following file formats are used for each output type.

Ingested data: SequenceFiles provide efficient writes for blob data.

Intermediate data: Avro offers rich schema support and more efficient writes than Parquet, especially for blob data.

Final output: Combination of Parquet, Avro, and JSON files

Imagery metadata: Parquet is optimized for efficient queries and filtering.

Imagery: Avro is better optimized for binary data than Parquet and supports random access for efficient joins.

Aggregated metadata: JSON is efficient for small record counts distributed across a large number of files and is easier to debug than binary file formats.

7 Shuffling

Reduce, GroupBy, ReduceByKey, sortBy transformations use shuffling. Shuffling is an expensive operation. Use reduceByKey instead for groupBy or aggregateBy. Sometimes, sorting datasets will help before using Reduce operations.

Increase level of parallelism to reduce outOfMemory errors.

Using broadcasting can greatly reduce the size of each serialized task and cost of launching a job over the cluster. If a task uses any large object from the driver program inside of them like a static lookup table, consider turning it into a broadcast variable.

You can set default shuffle partitions within spark code or at configuration level in spark defaults or at cluster.

Spark.sql.shuffle.partitions=20. Configures the number of partitions to use when shuffling data for joins or aggregations

8 Broadcasting

8.1 Broadcast hints

The BROADCAST hint guides Spark to broadcast each specified table when joining them with another table or view. When Spark decides the join methods, the broadcast hash join (i.e., BHJ) is preferred, even if the statistics is above the configuration spark.sql.autoBroadcastJoinThreshold value. When both sides of a join are specified, Spark broadcasts the one having the lower statistics. Note Spark does not guarantee BHJ is always chosen, since not all cases (e.g. full outer join) support BHJ. When the broadcast nested loop join is selected, we still respect the hint.

eg.:

from pyspark.sql.functions import broadcast

broadcast(spark.table("src")).join(spark.table("records"), "key").show()

8.2 Broadcast Variables

The broadcast variables are only copied once to each executor and then tasks can use that copy for computation. This improves both run time and storage space. Broadcasting large variables and some small datasets improve overall performance of the spark code.

The broadcast variables are read-only variables. Spark also provides accumulators that are equivalent to counters in map reduce. Accumulators are the write-only variables and can be used to get statistics such as the number of bad records in the input file while running your application.

Eg:

val lookup = Map(“This” -> “frequent”, “is” -> “frequent”, “my” -> “moderate”, “file” -> “rare”)

val broadcastLookup = sc.broadcast(lookup)

val acc = sc.accumulator(v)

9 In-memory Caching

Spark SQL can cache tables in-memory and can be reused.

Spark.catalog.cacheTable(“tableName”)

Spark SQL will scan only required columns and will automatically tune compression to minimize memory usage and GC pressure. Also, please remember to uncache the table from memory once it's no longer required.

10 Data Serialization in Spark

It is the process of converting the in-memory object to another format that can be used to store in a file or send over the network. It plays a distinctive role in the performance of any distributed application. The computation gets slower due to formats that are slow to serialize or consume a large number of files. Apache Spark gives two serialization libraries:

· Java serialization

· Kryo serialization

Java serialization – Objects are serialized in Spark using an ObjectOutputStream framework, and can run with any class that implements java.io.Serializable. The performance of serialization can be controlled by extending java.io.Externalizable. It is flexible but slow and leads to large serialized formats for many classes.

Kryo serialization – To serialize objects, Spark can use the Kryo library (Version 2). Although it is more compact than Java serialization, it does not support all Serializable types. For better performance, we need to register the classes in advance. We can switch to Kyro by initializing our job with SparkConf and calling-

conf.set(“spark.serializer”, “org.apache.spark.serializer.KyroSerializer”)

We use the registerKryoClasses method, to register our own class with Kryo. In case our objects are large we need to increase spark.kryoserializer.buffer config. The value should be large so that it can hold the largest object we want to serialize.

//Scala

conf.set("spark.kryo.registrationRequired", "true") conf.registerKryoClasses(Array(ClassOf[FirstCustomClass],ClassOf[SecondCustomClass]))

11 Spark Driver

As Spark driver has the responsibility of DAG construction and task scheduling, sometimes, the driver itself becomes the bottleneck. You can configure Spark driver to enable dynamic executor allocation. This means Spark can dynamically add or remove executors on the fly. If dynamic executor allocation is enabled and your job has performed an operation such as coalesce() and does not require some executors, then the driver will free the resources. Compared to traditional resource allocation, where we have to reserve the resources. This feature brings better resource utilization and can be useful in multi-tenant environments. You can enable it by adding the following properties to your application code:

spark.dynamicAllocation.enabled = true

spark.dynamicAllocation.executorIdleTimeout = 2m

spark.dynamicAllocation.minExecutors = n (set minimum number of executors)

spark.dynamicAllocation.maxExecutors = m (set maximum number of executors)

You can also configure the memory allocated to the driver program by setting a flag for --driver-memory.

12 Other Performance tuning considerations

12.1 Caching Level

In the case of caching, we need to be careful when choosing the storage level. The default caching level is MEMORY_ONLY, which keeps the data in memory. If the RDD is big enough and can't fit in memory, then Spark fits as many partitions as it can and the remaining partitions will be recomputed. This can be avoided if your RDD is expensive to recompute. In such cases, you can use MEMORY_AND_DISK as the storage level. This will move the remaining partitions on the disk and return it to the memory when they are needed without recomputing them.

There are other storage levels available like memory_only_ser, memory_and_disk_ser, disk_only, memory_only_2 and memory_and_disk_n which are rarely used but may improve performance in certain use cases.

12.2 Enabling Tungsten

The goal of Project Tungsten is to improve the efficiency of CPU and memory for Spark applications. Three of the main optimizations are the following:

Memory management: This allows Spark to manage memory by eliminating the overhead of Java virtual machine (JVM) objects that reduces memory consumption and garbage collection (GC) overhead.

Binary processing: One of the aims of Project Tungsten is to process the data in binary format itself, instead of JVM objects, which are heavier in size. This binary format is known as the UnsafeRow format.

Code generation: Using this feature, Spark uses optimization of structured APIs to directly generate the bytecode for your code. This can bring lots of advantages when you write large queries.

Project Tungsten is also home to some other optimizations, such as sorting, join, and shuffle. You can enable or disable Tungsten by using spark.sql.tungsten.enabled configuration.

12.3 Task scheduler

It is the responsibility of the task scheduler to schedule these tasks with an executor based on the available cores and data locality. By default, a task is assigned a single core to perform its operation. You can change this behavior by configuring spark.task.cpus. Use multiple cores for this parameter.

12.4 Catalyst optimizer

If you are using structured APIs like SQL, DataFrames and Datasets, Catalyst optimizer helps to create logical query plans for each transformation and then optimizes to create optimized query plans and then it efficiently transforms optimized logical query plans to physical query plans.

12.5 Join Performance

Spark SQL provides a variety of joins including Shuffle hash join, Broadcast hash join and Cartesian join.

If the size of both the tables is large and they both are bucketed/partitioned on the same joining field, then shuffle hash join can best suit your needs. It works best when the data is evenly distributed based on the key field and there are enough unique values for that key field to achieve the necessary parallelism. If one of the tables is small in size, you can use broadcast join, which caches the small data on each machine and avoids shuffle.

12.6 Data Locality

Data locality can have a major impact on performance. It means code and data stay together. Shipping code is much easier than shipping data as data is much larger than code. Data locality is how close data is to the code processing it. There are several levels of locality based on the data’s current location. In order from closest to farthest:

· PROCESS_LOCAL data is in the same JVM as the running code. This is the best locality possible

· NODE_LOCAL data is on the same node. Examples might be in HDFS on the same node, or in another executor on the same node. This is a little slower than PROCESS_LOCAL because the data has to travel between processes

· NO_PREF data is accessed equally quickly from anywhere and has no locality preference

· RACK_LOCAL data is on the same rack of servers. Data is on a different server on the same rack so needs to be sent over the network, typically through a single switch

· ANY data is elsewhere on the network and not in the same rack

You can configure and override this behaviour using spark.locality spark configuration option.

12.7 Code generation

Spark can generate bytecode for your queries on the fly. Spark compiles each query to its Java bytecode equivalent code when spark.sql.codegen is set to true. By default, this property is set as false. In case your queries are big, you can set this property to true to improve on performance.

12.8 Speculative execution

You often find some tasks running slower than other tasks from Spark UI. When spark.speculation is enabled, Spark will identify slow running tasks and run the slow running task on other nodes to complete jobs quickly. Whichever task completes first, the other task will get killed by Spark.

#Python

conf.set("spark.speculation","true")

12.9 Disk

Even though Spark applications benefit largely from memory, the disk also plays an important role. Spark uses disk space to store the shuffle data temporarily. In Spark standalone mode and Mesos, this location can be configured in SPARK_LOCAL_DIRS variable. In YARN mode, Spark inherits YARN's local directories. Using SSDs for these directories, Spark jobs can benefit a lot.

12.10 Tuning Data Structures

The first way to reduce memory consumption is to avoid the Java features that add overhead, such as pointer-based data structures and wrapper objects. There are several ways to do this:

Design your data structures to prefer arrays of objects, and primitive types, instead of the standard Java or Scala collection classes (e.g. HashMap). The fastutil library provides convenient collection classes for primitive types that are compatible with the Java standard library.

Avoid nested structures with a lot of small objects and pointers when possible.

Consider using numeric IDs or enumeration objects instead of strings for keys.

If you have less than 32 GB of RAM, set the JVM flag -XX:+UseCompressedOops to make pointers be four bytes instead of eight. You can add these options in spark-env.sh.

12.11 Choosing the right transformations

One of the ways optimizing spark application is by writing effective code. If you choose transformations very carefully to avoid a lot of shuffling of your data, the performance improves. Lesser the shuffle, less in the execution time. For example, reduceByKey() and groupByKey() both achieve the same results but reduceByKey is more efficient than using groupByKey() because they both process data differently. groupByKey() transformation shuffles data across the nodes based on the keys and on the other hand reduceByKey() first aggregates the data locally on each node and then its transfered to other nodes based on the key. reduceByKey() shuffles less data compared to groupByKey().

In some cases, when the data is highly skewed, you might face issues with groupByKey() because all the records with the same key are sent to a single node.

Consider using filter, sortBy, orderBy before using reduce functions, which may improve performance.

13 Delta lakes

Delta lakes is a new initiative by databricks and it is an open source storage layer that brings ACID capabilities to Spark APIs.

It offers a lot of new features like ACID capabilities, data modifications like update, insert and delete at row level to Spark dataframes. This makes Spark dataframes mutable.

Amazon AWS Cloud Managed Database Services

183:766491093 (Srinivasa Rao) — Tue, 21 Apr 2020 01:01:26 GMT

Database options on Amazon AWS

Amazon S3

Amazon Simple Storage is a managed object store service provided by AWS. It is a distributed, high-scalable, high available cloud storage. It can be used in place of HDFS like your on-premise Hadoop data lakes where it becomes foundation of your data lake. It can also be used to store unstructured data, content and media, backups and archives and so on.

Cassandra Managed service

Amazon Managed Apache Cassandra Service is a scalable, highly available, and managed Apache Cassandra–compatible database service. Cassandra is very good for application which have very high throughput and supports faster reads when queries on primary or partition keys. It provides sub milliseconds response times.

Amazon ElastiCache

Amazon ElasticCache is managed service that supports Memcached and Redis implementations. This service improves performance of the web applications by allowing to store information in in-memory cache and then retrieve information from fast in-memory caches, instead of making multiple trips to slower backend databases.

This can also be used to store static content on web and also used as fast layer in lambda architecture.

Amazon DocumentDB

Amazon DocumentDB is a fully managed document-oriented database service which supports JSON data workloads. It is MongoDB compatible. Its fast, high available and scales over huge amounts of data.

Redshift

Amazon RDS

Amazon Relational database service(RDS) provides a fully managed relational databases both open source and commercial database engines. It supports MySQL, PostgreSQL, Oracle, SQL Server and Amazon Aurora. You can also bring your own license if you have one internally. Amazon RDS manages all operations, support related tasks internally.

Amazon Dynamo

Microsoft Azure Cloud managed Database services

183:766491093 (Srinivasa Rao) — Mon, 20 Apr 2020 23:22:57 GMT

Database options on Microsoft Azure

Azure BLOB store

Azure BLOB is Microsoft’s cloud managed service for object storage. It can be used to store the unstructured data and also can be used as raw data layer for modern multi-layered data lakes on Azure cloud.

Blob storage supports Azure Data Lake Storage Gen2, Microsoft's enterprise big data analytics solution for the cloud. Azure Data Lake Storage Gen2 offers a hierarchical file system as well as the advantages of Blob storage, including:

• Low-cost, tiered storage

• High availability

• Strong consistency

• Disaster recovery capabilities

Azure SQL Database

Azure SQL database is a fully managed relational database that provides the SQL Server engine compatibility. Azure SQL database is now Azure arc-enabled. You can run this service on premises on infrastructure of your choice with cloud benefits like automation, no end of support, unified management, and a cloud billing model.

Azure Database for MySQL, PostgreSQL and MariaDB

Azure also provides managed database services built on MySQL, MariaDB and PostgreSQL.

Azure Synapse Analytics (SQL Data Warehouse)

Azure SQL Data Warehouse is managed analytical service that brings together enterprise data warehouse and Big Data analytics.

Azure Cosmos DB

Azure Cache for Redis

Azure Cache for Redis is a powerful fast and scalable in-memory data store built on Redis open source.

Google (GCP) Cloud Managed Database Services

Mon, 20 Apr 2020 23:05:41 GMT

Database options on Google Cloud Platform(GCP)

This blog explains various database and data storage options available on Google cloud platform for different kinds of data storage and access needs.

Cloud Storage

Cloud Storage is fully managed and can scale dynamically. It is commonly used for object storage, video transcoding, video streaming, static web pages, and backup. It is designed to provide secure and durable storage while also offering optimal pricing and performance for our requirements through different storage classes.

Cloud Storage uses the concept of buckets. A bucket is the basic container where our data will reside and is attached to a GCP project, such as other GCP services. Each bucket name is globally unique and once created we cannot change it. There is no minimum or maximum storage size, and we only pay for what we actually use.

BigQuery

BigQuery is a fully managed, serverless analytics service. It can scale to petabytes of data and is ideal for data warehouse workloads. It is, of course, the analysis stage of our solution, and once Dataflow processes our data, BigQuery will provide the value to our business by querying large volumes of data in a very short period of time. Queries are executed in the SQL language; therefore, it will be easy to use for many. We should emphasize that BigQuery is enterprise-scale and can perform large SQL queries extremely fast—all without the need for us to provision any underlying infrastructure.

Cloud datastore provides both strong and eventual consistency and very low latency.

It's a good option for storing streaming data before we store it back to BigQuery.

Cloud SQL

Cloud SQL is a database service that makes it easy to set up, maintain, and manage our relational PostgreSQL or MySQL database on Google Cloud. Cloud SQL announced plans to release support for Microsoft SQL Server in future.

Cloud Spanner

There may be situations where we require horizontal scaling and Cloud SQL will not fit these requirements. Cloud Spanner is a cloud-native, fully managed offering that is designed specifically to combine relational database features such as support for ACID transactions and SQL queries with the horizontal scaling of a non-relational database. We should look to use Cloud Spanner when we require storage capacity requirements above 10 TB, as well as if we have requirements for high queries per second or to deliver over multiple regions. Unlike most databases, Cloud Spanner is globally distributed and provides a strongly consistent database service with high performance.

Bigtable

Bigtable is GCP's big data NoSQL database service. Bigtable is low latency and can scale to billions of rows and thousands of columns. It's also the database that powers many of Google's core services, such as Search, Analytics, Maps, and Gmail. This makes Bigtable a great choice for analytics and real-time workloads as it's designed to handle massive workloads at low latency and high throughput.

HBase is effectively an open source implementation of the Bigtable architecture and follows the same design philosophies. Bigtable stores its data in tables, which are stored in a key/value map. Each table is comprised of rows, which will describe a single entity.

Cloud Memorystore

Memorystore provides a fully managed Redis in-memory data store service. It is used to cache application data, session data, intermediator data like lookups. It provides sub-millisecond data access,

Data Cataloging(Metadata) on Cloud

Sun, 12 Apr 2020 19:47:45 GMT

Unified Metadata design on various clouds AWS, GCP and Azur e

1 Introduction

This blog covers collecting, storing and searching and discovering various metadata across different data lake work streams on various clouds.

2 Types of metadata

Technical Metadata (System generated - auto ingested)

Table names, Column names
Table descriptions, column descriptions
Data created; date modified

Business Metadata (User provided/inferred)

PII information (Table and Column level)
Table Owner
Table Source
Business logic to derive column
Retain until date
Data Quality Owner
Data Governance
Data Quality information
Environment (prod/dev/qa/temp)
Classify data

Data Pipeline Metadata (System generated, and User provided)

Data pipelines
ETL/ELT specific information
Scheduling information

Data Lineage Metadata (automated through reusable components and system level defaults - for auditing only)

Data last updated/created (add last updated and create timestamp to each row)
Who updated the data (data pipeline, job name, user name and so on - Use Map or Struct or JSON column type)?
How data was modified or added (storing update history where required - Use Map or Struct or JSON column type)

Any solution we use to tackle above metadata should cover the following:

➔ A unified view into all the data no matter where it is stored

➔ Integration with analytical tools

➔ A way to automatically build all metadata and keep it in sync with our data as it evolves

➔ Should have Data Governance in place through IAM controls

➔ Should have a proper API to integrate it with other tools like data pipelines.

➔ Should be able to search through easily

Data Catalog (GCP)

Data Catalog is a fully managed and scalable metadata management service that empowers organizations to quickly discover, manage, and understand all their data in Google Cloud. It offers a simple and easy-to-use search interface for data discovery, a flexible and powerful cataloging system for capturing both technical and business metadata, and a strong security and compliance foundation with Cloud Data Loss Prevention (DLP) and Cloud Identity and Access Management (IAM) integrations.

AWS Glue Data Catalog

AWS Glue data catalog collects metadata from different data sources like amazon RDS, S3, RedShift and Dynamo and allows users to search and discover data from AWS provided UI or through APIs.

Azure Data Catalog

Azure Data Catalog is a fully managed cloud service. It can collect metadata from different data sources and allows users to search and discover data.

Note: The above images are courteous to respective clouds and taken from their documentation.

Data modeling techniques for modern data warehousing

Sat, 11 Apr 2020 19:08:32 GMT

Big Data presentation layer (Operational)

1.Introduction

This document presents various data modeling techniques that may be followed for your presentation layer of operational data warehouse or data lake. It presents various techniques for data modeling using google Big Query and similar technologies like Redshift, Snowflake, Azure SQL data warehouse, Couchbase, MongoDB and so on. Based on use cases and usage patterns, these techniques can be adopted for different modules and projects within our data lake sphere.

The presented data model techniques are a combination of denormalized (Flattened), Dimensional and Set/Relational models that should suit all your use cases.

These data modeling techniques will be used for operational Big Query or similar datasets from where the end users access data through various tools like Tableau.

2. Types of Data

Business master data

This data contains all of your business master data and will be stored in a separate dataset. This data will be shared among all other projects/datasets.

This will help you to avoid duplicating master data thus reducing manageability. This will also provide a single source of truth so that different projects don't show different values for the same.

As this data is very critical, we will follow type 2 slowly changing dimensional approach which will be explained later in this document.

There are lot of MDM tools available to manage master data more appropriately but for moderate use cases, you can store this using database you are using.

ETL or System (meta)data

This metadata is neither business metadata or the technical metadata but may need to store some data that will help your ETL/ELT workflows and various other systems. This may help in reducing hard coding within the ETL/ELT pipelines and any other systems. The example of this data includes external APIs, FTP/SFTP site information and their access credentials and any other reference data. When that information changes, you can update corresponding tables and don’t need to modify jobs and other references where it's being used.

This data will not be used by any end users or reports.

Transactional data

This is the main project specific data which will have all the data related to various projects which includes historical data as per the business retention requirements.

Different datasets will be created for different projects. Generally, this data along with business metadata will be consumed by end user applications like Tableau reports.

Metadata and Cataloging

This is entirely different kind of metadata and to avoid confusion, this is handled separately. This data is not stored in any tables and not used by end users or any systems/applications.

All the tables need to have the following labels so that we can integrate it with metadata catalog tool, so that users can search and view the data easily.

3 Slowly changing dimensions

Handling slowly changing dimensions is the key for any successful operational data lake or data warehousing implementations. It's a little harder to implement this for traditional databases like Oracle and SQL Servers but as most of the modern databases like Big Query allow complex types, so it became a little easier to handle slowly changing dimension types in the modern Big Data type databases.

3.1 Traditional data warehouse implementations of SCDs

In the traditional Data warehouse implementations, the following are the 3 types of slowly changing dimensions:

Type 1 SCDs - Overwriting

In a Type 1 SCD the new data overwrites the existing data. Thus, the existing data is lost as it is not stored anywhere else. This is the default type of dimension we create. We do not need to specify any additional information to create a Type 1 SCD.

Type 2 SCDs - Creating another dimension record

A Type 2 SCD retains the full history of values. When the value of a chosen attribute changes, the current record is closed. A new record is created with the changed data values and this new record becomes the current record. Each record contains the effective time and expiration time to identify the time period between which the record was active.

Type 3 SCDs - Creating a current value field

A Type 3 SCD stores two versions of values for certain selected level attributes. Each record stores the previous value and the current value of the selected attribute. When the value of any of the selected attributes changes, the current value is stored as the old value and the new value becomes the current value.

3.2 Big Data Implementation of SCDs

In the modern data lake implementations, we can club together Type 2 and Type 3 into one, we can call it as “retaining historical values”.

So, there are only 2 types of SCDs in Big Data implementations

Overwriting

In this type, the new data overwrites the existing data. Thus, the existing data is lost as it is not stored anywhere else. This is the default type of dimension we create. We do not need to specify any additional information to create this type of dimension.

Retain historical values

In this type, we retain historical changes using complex data types available within Big Data Technologies. This is some kind of semi structured type and will have full control of up to what extent we can retain historical changes. JSON, MAP, STRUCT are some example types of storage options.

This will give full freedom on how much retention we can use and any associated values like when it was changed and how it was changed and so on. Also, this can be different for different dimensions.

How to handle Type 2 and Type 3 together

Type 2 and Type 3 (in the big data world, “retaining historical values”) can be handled in Big Query and similar technologies using complex data types. Wherever we need to retain historical data for any dimension, we need to include the following column to be created along with dimensional tables and be managed through data flow jobs.

SCD_History ARRAY<STRUCT<updatedon DATETIME or TIMESTAMP, columnname STRING, oldvalue STRING, newValue STRING>>

the above syntax is for Big Query. You need to find similar data type for different type of databases like Redshift, Snowflake, Couchbase and so on. For example, Cassandra you can use MAP data type, MongoDB, you can use nested JSON and so on. Where complex data type not allowed, you can use STRING or TEXT types and use JSON format.

You can use the UNNEST operator available with BigQuery SQL to flatten out the history column in case we need to report or search values within this complex column type.

This will also give users flexibility like whether to use current value or the value at that point in time.

It is important to know this method is used for slowly changing dimensions that means the changes (or updates) happen very infrequently over time.

The following are some key points to keep in mind while adding SCD specific columns:

Whether the old or historical values used by end user applications or just for auditing.
What columns we need to track for changes over time.

4. Data modelling techniques

The following are the various data modeling techniques that can be used to implement various use cases within your data lake (presentation layer not in raw layer). Which technique you use will be decided based on the requirements like how the data is accessed by end users and how the data gets into our database like Big Query.

Different use cases, projects or modules within our projects can follow one or more of the below modeling techniques to satisfy the requirements.

4.1 Denormalized or flattened models

In this modeling technique, we denormalize and flatten fact with its dimensional data. You should be careful in choosing this approach and this technique works well with very simple use cases and does not work with most of the complex data warehouse needs.

Advantages of using flattened denormalized model

● Simple to query which avoids joins.

As all the data stored in one table, it's easy to query and don’t need to join many tables.

● Perform well with latest Big Data technologies including Big Query.

Most of the Big Data technologies which are all distributed in nature prefer this kind of data model and performs very well.

Disadvantages of using flattened denormalized model

Here are some of the reasons why this technique does not work with most of the use cases:

● Where dimensional data changes

When the dimensional data changes, we need to update millions of rows within a big denormalized table. It is very time consuming; we cannot use this technique when dimensional data constantly changes.

● Where there are too many attributes

Some dimensions are simple like states which will have state code and name, but most of the dimensions will have too many attributes like Patient. Patient dimension will have too many attributes like id, name, date of birth, address, insurance and so on. It's difficult to tuck all the information into a single denormalized flattened table.

● Dimensional data is nested and has complex hierarchies.

Sometimes dimensional data is nested like Region dimension will have regions, countries within region and states within countries and cities within states. It's very difficult to maintain all that data into one single wide table.

● Where they need to keep track of changes

Most of the time we need to handle historical changes to dimension data which will be difficult to handle if we go with denormalized flattened approach.

● Users need to see dimensional values in drop-down lists

Most of the time when users use front-end applications including reporting tools like Tableau, they need to see dimensional values in the drop-downs to filter out and/or slice and dice data, which will be a huge process to get all distinct values for that dimension from millions or billions of rows from flattened tables.

4.2 Star Schema (modified)

Big Query, Snowflake and some data warehouse databases allows joins among tables, so Modified Star Schema approach with some changes may be a good data modeling approach for our data lake implementations. Where the joins not allowed, might requires multiple queries to achieve same if we use this model, for example Cassandra.

Here are some changes we can make to the traditional star schema approach.

● Club together Type 2 and Type 3 slowly changing dimensions when we need to track historical changes. This is explained above under slowly changing dimensions.

● Club together similar dimensions into one big denormalized dimension table.

Example, using state, country and region dimensions separately, using one-dimension table can be called “Geography” or “Location” which will have region, country and state information together will perform better and a good approach.

● Grouping together multiple facts into one big fact table

Similarly, we can group together multiple fact tables into single or less number of denormalized fact tables, so that users don't need to use multiple fact tables in their queries and instead use a single flattened fact table that will speed up the query performance.

The whole idea is to use denormalized/flattened schema model as much as possible which will reduce the number of tables and thus joins within queries. At the same time, the modified star schema approach which addresses shortcomings that we mentioned earlier while using fully denormalized flattened schema approach.

There are other kinds of data modeling techniques that might suit certain operational data lake needs may be considered based on the use case and requirements.

4.3 Hybrid Star/Denormalized

In this case we use denormalized schema models which tie together facts and certain dimensions together. The dimension that will never change and simple will go along with the fact table and complex dimensions and which will change constantly will be kept separate.

Sometimes we can keep part of the dimension that is used by end user applications or reports will become part of the fact table and other parts of the dimension will stay in separate tables.

4.4 Nested dimensions within fact

You can even keep the nested dimensions that will not change over time be part of the fact table using complex data types like ARRAY<STRUCT>> in case of Big Query or JSON (MongoDB) or MAP(Cassandra) columns in case of any other NoSQL databases.

4.5 Pattern (Query) specific data model

This is another kind of data modeling technique that is used when using NoSQL technologies like Cassandra, HBase, Dynamo and so on where “Joins” among tables are not allowed.

The table is designed based on the query usage patterns. This model helps to serve data very fast but complex to maintain. Also, there will be too much redundancy.