ETL Offloading

In a typical data warehouse environment, IO- and CPU-intensive data transformation and loading is executed on the data warehouse (Figure 8-1). Over time, as more data is ingested and stored in the data warehouse, its ETL window will start eating into the business hours, causing critical reports and dashboards to be unavailable to business users. An obvious workaround is to upgrade the data warehouse. This is a short-term workaround. Over time, you’ll have the same performance and scalability issue as you ingest more data into your data warehouse.

The correct solution is to augment your data warehouse with a data lake via ETL offloading. As you can see in Figure 8-2, most of the heavy data transformation is executed on the big data platform that, thanks to highly distributed data processing frameworks such as Spark and MapReduce, can churn through massive amounts of data in relatively short amounts of time. The data lake can act as a staging area or an operational data store. Processed and aggregated data is then loaded into the data warehouse, which is now free to be used for data analytics.

Analytics Offloading and Active Archiving

You can also use your data lake to offload some of your expensive reports and ad hoc queries. By offloading some of your analytics, you free up your data warehouse for more important tasks. Analytics offloading also allows your data warehouse to handle more data and concurrent users while keeping a small data warehouse footprint – saving your organization millions in data warehouse upgrades.

If you already use your big data platform for ETL offloading, then you already have a copy of your data stored in the big data platform. You can also help make your data warehouse more efficient by moving or archiving older data to the big data platform. Once you’ve got the data you need in the big data platform, you can just direct your BI or data visualization tool such as Tableau or MS Power BI to Impala via JDBC (Figure 8-3). In some cases, BI tools has a semantic layer that provides a way to connect, read, and cache data from different data sources. It hides the details of the data source from users, providing a unified view of your data.

Data Consolidation

One of Hadoop’s best features is its ability to store massive amounts of data. By combining storage from commodity hardware and providing distributed storage engines such as HDFS, HBase, and Kudu, Hadoop made storage and processing of large data sets not just possible but also practical. Unlike the data warehouse, which can only store structured data, data lakes can store and process structured and unstructured data sets, making it a true repository of corporate data.

Consolidating your data in a centralized data lake delivers several benefits. First, it enhances data analysis by making it easier to join and correlate data. Second, because your data is in one central location, it makes it easier for users to run SQL joins across various data sets providing, for example, a 360-degree view of your customer. Third, feature engineering – choosing and processing attributes and creating feature sets for your predictive models is also easier. Fourth, data governance and master data management becomes straightforward once data is consolidated in one central location, providing you with one golden copy of the data instead of having multiple copies of the same piece of data spread across multiple data silos.

Perhaps the most tangible benefit from consolidating your data is the potentially massive reduction in licensing, operational, and infrastructure costs. It is fairly common for large organizations to accumulate hundreds or even thousands of databases scattered across its enterprise. By consolidating these databases into a centralized data lake (Figure 8-4), you can decommission the old servers hosting these databases and stop paying for the database software license.

Dimensional Modeling

Dimensional modeling is a way of logically modeling database tables for ease of use and fast SQL query performance. Dimensional models serve as the logical basis for most business intelligence and OLAP systems in the market today. In fact, most data warehouse platforms are designed and optimized with dimensional modeling in mind. Dimensional modeling has several concepts:

Star Schema

There are two ways to physically implement a dimensional model: by using a star or snowflake schema. A star schema contains a fact table in the middle, surrounded by dimension tables. The fact and dimension tables are connected by the tables’ foreign and primary keys. The ER diagram in Figure 8-5 is an example of a star schema.

Snowflake Schema

A snowflake schema is very similar to a star schema, with a central fact table but with dimension tables that are further normalized into subdimension tables. Ralph Kimball, the data warehousing expert, generally recommends using a star schema. He recommends using a snowflake schema in cases where some measures refer to a host of attributes not shared by other measures. ^v The ER diagram shown in Figure 8-6 is an example of a snowflake schema.

Big Data Warehousing with Impala and Kudu

For years, Hive was used to implement rudimentary data warehouses in Hadoop. It was painfully slow and unwieldy, but it was way easier to use than MapReduce and it got the job done so users tolerated it.

Years later Impala and other open source MPP engines appeared. Used together with columnar file formats such as Parquet, it paved the way for Hadoop in the Enterprise, enabling BI and OLAP tools to connect to Hadoop clusters and get fast performance. ETL and dimensional modeling with Impala and Parquet were still cumbersome, but the benefits outweighed the costs, so users endured using it.

Impala and Kudu just made it a whole lot easier to store, process, and analyze relational data on Cloudera Enterprise. While it still doesn’t have all of the advanced features of a commercial data warehouse platform, its capabilities are now close enough to a relational database.

Let’s work on an example by implementing a simple dimensional star schema model in Kudu. I’ll replicate a subset of the AdventureWorksDW database from SQL Server to Kudu and run some SQL queries against it using Impala. We’ll copy three dimension tables and one fact table. We’ll also copy a subset of the columns. The schema should look like Figure 8-7.

I’ll transfer the following columns: ProductKey, OrderDateKey, CustomerKey, OrderQuantity, UnitPrice, SalesAmount, TaxAmt, TotalProductCost, Freight, OrderDate, SalesOrderNumber, and SalesOrderLineNumber from the FactInternetSales table as shown in Figure 8-8.

I’ll copy the following columns: CustomerKey, FirstName, LastName, BirthDate, YearlyIncome, TotalChildren, EnglishEducation, EnglishOccupation, HouseOwnerFlag, and NumberofCarsOwned from the DimCustomer table as shown in Figure 8-9.

We’ll pick four columns from DimProduct: ProductKey, EnglishProductName, EnglishDescription, and Color from the DimProduct table as shown in Figure 8-10.

We’ll pick a few columns from the DimDate table: DateKey, FullDateAlternateKey, DayNumberOfWeek, EnlighsDayNameOfWeek, DayNumberOfMonth, DayNumberOfYear, WeekNumberOfYear, EnglishMonthName, MonthNumberOfYear, CalendarQuarter, CalendarYear, FiscalQuarter, and FiscalYear as shown in Figure 8-11.

I need to copy the tables from SQL Server to Impala. To do that, I manually generate a CSV file for each table. You can use any of the ETL tools we covered in Chapter 7. In this case, I used SQL Server’s data export and import tool to generate the CSV files. Don’t forget to exclude the column headers. I then move the files to HDFS and create an external table on top of each file. Listing 8-1 shows the steps.

Now that the data is accessible from Impala, We can create the Kudu tables and populate it with the data from the external tables. The steps are shown in Listing 8-2.

There are a few important things to discuss here. Notice that I used the Impala built-in function uuid() to generate a unique ID for the fact and dimension tables. Impala and Kudu do not include an auto-increment feature, which is actually considered an anti-pattern on distributed environments. With Kudu, Spanner, and similar distributed systems, range partitioning on a column whose values are monotonically increasing will direct all inserts to be written to a single tablet at a time, causing the tablet to grow much larger than other tablets and limiting the scalability of inserts. This is also known as hot spotting. Hot spotting also happens in HBase when inserting rows with monotonically increasing row keys.

While using UUIDs as a primary key on a database running on a single SMP server is a disaster, in distributed MPP environments, hash partitioning using a unique key that is not derived from the business or data contained in the table will spread the workload across the nodes in your cluster. Hash partitioning protects from data and workload skew and increases scalability of data ingestion by enabling inserts to go to multiple tablets in parallel. ^viii

Some database systems such as Microsoft SQL Server have a native UUID data type that provide a more efficient storage for UUIDs. Kudu doesn’t have a UUID data type, therefore you have to use the STRING data type to store the UUID. The 36-character UUID does take up some space; however benchmarks indicate that having a 36-character primary key in Kudu doesn’t affect performance for most queries, although your experience may vary. If you are concerned about using UUIDs, then another option is to hash the UUID to generate a non-sequential BIGINT value to serve as your primary key. You can use Impala’s built-in hashing function fnv_hash() (which is not cryptographically secure, which means collision may occur when generating unique ID’s) or a more cryptographically secure hashing algorithm. Just make sure you test the performance of your insert statements. Calling built-in functions every insert statement may potentially cause performance bottlenecks. Of course, there are other ways to generate a non-sequential unique key.

As you can see, the fact table is hash and range partitioned. By combining the two partitioning strategies, we gain the benefit of both, while reducing the limitations of each. ^ix By using range partitioning on the date column, your query can take advantage of partition pruning when scanning by date ranges. In addition, new partitions can be added as new data is added to the table (which will result in 16 additional tablets). By using hash partitioning on the ID column (which is a non-sequential UUID), inserts are spread evenly across multiple tablets up to the number of hash partitions (which in our case is 16), increasing throughput and performance of data ingestion. Comparing this strategy with the current AdventureWorksDW schema, you will find a couple of differences. First off, FactInternetSales’ primary key is on SalesOrderNumber and SalesOrderLineNumber. This is adequate for an SQL Server database running on one server. However, this breaks several of the rules we’ve just discussed about partitioning and choosing primary key values on distributed systems. SalesOrderNumber is monotonically increasing, and the majority of SalesOrderLineNumber’s value is 1. If we use these two columns as our primary key, all inserts will be directed to one partition, causing data and workload skew and slowing down data ingestion.

Dimensions tables are usually small and mostly read-only. Slowly changing dimensions doesn’t usually require the kind of performance consideration as fact tables so we’ll just hash partition the dimension tables using a unique UUID as the primary key. Larger dimension tables may require a different partitioning strategy.

While Kudu requires all tables to have primary keys, it doesn’t have foreign key constraints. Therefore, data inconsistencies are possible in a Kudu data warehouse. However, it is common practice for data warehouses to have foreign key constraints disabled due to performance and manageability reasons. Listing 8-3 shows the structure of the Kudu tables we just created.

We can now run Impala queries against our data warehouse in Kudu.

For example, here’s a query to get a top 20 list of everyone who earns more than $100,000 and are also home owners.

To get a list of everyone who bought a “Sport-100 Helmet” and also owns more than two cars.