Randal Scott King

Resources for Data Scientists

Blog

Data Warehousing with Hadoop

hadoop data warehouse

 

Almost from the moment Hadoop was first introduced, organizations have sought to replace their expensive data warehousing systems with it. Hadoop’s distributed nature and the fact that it uses commodity hardware make it cheap, massively scalable, and highly available. However, data warehousing with Hadoop is often ill-advised and the projects have ended badly. HDFS, the Hadoop filesystem, was not designed for the speed and updateability requirements of data warehouses.

The Problem of Immutability

The main problem behind replacing an existing data warehouse with Hadoop is a seemingly innocent concept called immutability. Data in HDFS cannot be changed, they can only be overwritten or appended. This makes Change Data Capture (CDC) and other data warehousing concepts difficult to implement. When Hadoop is used for archival or analytical purposes this is not usually an issue. However, data warehouses alter existing data tables anywhere from daily to multiple times a second.

Workarounds using existing Hadoop tools were invented to address the issue, but they are cumbersome, slow, and unreliable. Many who have attempted these workarounds have concluded that the efforts required are simply not worth the investment.

Previous Attempts

To make a Hadoop cluster based on HDFS and Hive function as a data warehouse, two main approaches evolved: partitioning based on time and/or the “Four-Step Method.”

Time-based partitioning separates the data and stores each month, week, or other time unit of data as a separate file. If changes are detected in a record for the month of July in the originating table, for example, the entire month is overwritten in HDFS. While somewhat effective at keeping data current, this method is needlessly cumbersome. Why overwrite an entire month or week just to update a single record? This is a waste of resources.

The Four-Step Method takes a different, much more complex approach. New data from the originating system are used to create a temporary table, which is then joined to the existing table in a view, which is then used to overwrite the existing data in HDFS at the end of the current time period.

Confused? You’re not alone.

Overcoming Immutability

MapR realized early in the game that immutability would be an issue and created a filesystem that was true read-write. Despite the proprietary nature of MapR-FS it uses the same API as HDFS, which all but negates compatibility issues.

Cloudera recognized the issues that the immutability of data in HDFS caused for their customers, and set to work to develop an alternative that would store data in such a way that would allow updates and changes in an expedient fashion. The result is Kudu, an efficient, updateable file store that acts much like a relational database in everyday use.

Commands like INSERT, UPDATE, and DELETE that are unavailable in HDFS and Hive (or didn’t work like their counterparts in relational databases), now function as expected and scale to previously unimagined sizes of data. Kudu even ups the ease of use by including support for an UPSERT command which chooses whether a record should be updated or inserted based on the primary key. Which brings us to the next point…

Keys to the Kingdom

Another notable difference between Hadoop and relational databases has been the lack of keys in Hive. Keys are essential to the success of Change Data Capture (CDC), the process by which data in the data warehouse are continually updated. Kudu’s support of and requirement for a primary key in each table mean that there is now a mechanism in Hadoop for preventing duplicate data during updates. This was a glaring omission before.

Caveats

It can’t be all sunshine and roses, can it? Since Kudu is a relatively new technology there are limitations that must be considered when using it for data warehousing, some of which include the following:

  • Kudu requires primary keys but doesn’t support foreign keys, so linking between tables in the way a relational database would is not yet possible.
  • As of version 1.3, Kudu doesn’t support creating tables from files or the LOAD DATA INFILE command. The fastest way to bulk load data is to bring them into HDFS, create a Hive table over them, and then create the Kudu table through a CREATE AS SELECT statement. Another alternative is to interface directly with the API, bypassing ODBC or JDBC drivers. The API works with C++, Java, and now even includes bindings for Python.
  • The number of supported data types for columns in Kudu is relatively small and does not include either DATE or TIMESTAMP.
  • Supporting a Hadoop data warehouse can be more challenging than with traditional solutions. Hadoop is an ecosystem of components all working together.

Now, Why Would I Do This?

Now that immutability isn’t the limiting issue it was, the question remains: why use a Hadoop data warehouse? This usually boils down to one of two things: price and/or scale. Some switchers cite excessive licensing costs for traditional data warehouses and limitations of the hardware used in those solutions. Hadoop also easily scales past petabytes of data and does so inexpensively.

Infer Schema when Importing to PostgreSQL or MySQL

infer schema
Postgres and MySQL are great Open Source databases in use by organizations of all sizes. Both of them are flexible and somewhat scalable. Both of them have nice GUI front-ends available to make them even easier to use.

Sequel Pro for MySQL is a handy GUI frontend that runs on Macs, and one of the things I like most is that it will infer the schema of a CSV on import. This saves a lot of time since it’s not necessary to create the table and define the columns ahead of time. Unfortunately Sequel Pro can also be a bit buggy.

pgAdmin is a similar tool designed for Postgres, and is more stable than Sequel Pro in daily use. But like a great many other database tools, pgAdmin won’t infer schema (and if I’m wrong about that, please comment below). So if you need to import a CSV with a lot of columns and you’d rather not define them all yourself, what can you do?

Back to the Command Line

One of the most useful means to deal with the problem is to use Pandas. Pandas dataframes will infer schema when you read in a CSV, and the “to_sql” function will load dataframes into a few different databases. Here’s an example of Pandas reading a CSV located at /Users/me/Desktop/mydata.csv and then loading it into a Postgres table named yourdata in a database called hisdb.

First we have to import Pandas, and “create_engine” from sqlalchemy:

from sqlalchemy import create_engine
import pandas as pd

Now we use create_engine to create a connection to the Postgres database:

engine = create_engine(‘postgresql://user:password@localhost/hisdb’)

Then we can load our CSV into a Pandas datframe, and use the “to_sql” to load that dataframe into a table called “yourdata” over the engine connection:

df = pd.read_csv(‘/Users/me/Desktop/mydata.csv’)
df.to_sql(‘yourdata’, engine)

Done and done. On occasion it will be necessary to change the data type of a column because Pandas guessed wrong, but that’s a minor inconvenience when compared to manually defining dozens of columns ahead of time.

Using SQL Commands in Spark with SparkSQL

SparkSQL

Spark has become a standard for performing analysis on huge amounts of data due to its distributed nature. SparkSQL evolved as a necessary component of Spark due to the need for working with structured data.

There are many times when there is a need to query data in Spark with SQL commands. Doing so isn’t complicated at all, and in fact you can use the same SQL syntax you would use in a relational database as long as you execute a few Spark-specific commands first. Here’s what it looks like:







First we’ll need to create our Spark Session and application:

In [16]:
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("example").getOrCreate()

Next we can load a csv. In the code below we’ll create a dataframe named “df”, use the Databricks CSV loader, specify that the data has a header and uses a comma for a delimiter, and give the path to the file. I like to use this version of the read command because it allows the user to specify the delimiter being used (not all delimited files use commas).

In [17]:
df = spark.read.format("com.databricks.spark.csv").option("header", "true").option("delimiter",",").load("/Users/randalking/Desktop/Mac/Superstore.csv")
In [18]:
df.show(10)

+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+
|Row ID|Order ID|Order Date|Order Priority|Order Quantity|    Sales|Discount|     Ship Mode| Profit|Unit Price|Shipping Cost|     Customer Name|Province| Region|Customer Segment|Product Category|Product Sub-Category|        Product Name|Product Container|Product Base Margin|Ship Date|
+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+
|     1|       3|  10/13/10|           Low|             6|   261.54|    0.04|   Regular Air|-213.25|     38.94|           35|Muhammed MacIntyre| Nunavut|Nunavut|  Small Business| Office Supplies|Storage & Organiz...|Eldon Base for st...|        Large Box|                0.8| 10/20/10|
|    49|     293|   10/1/12|          High|            49| 10123.02|    0.07|Delivery Truck| 457.81|    208.16|        68.02|      Barry French| Nunavut|Nunavut|        Consumer| Office Supplies|          Appliances|"1.7 Cubic Foot C...|       Jumbo Drum|               0.58|  10/2/12|
|    50|     293|   10/1/12|          High|            27|   244.57|    0.01|   Regular Air|  46.71|      8.69|         2.99|      Barry French| Nunavut|Nunavut|        Consumer| Office Supplies|Binders and Binde...|Cardinal Slant-D®...|        Small Box|               0.39|  10/3/12|
|    80|     483|   7/10/11|          High|            30|4965.7595|    0.08|   Regular Air|1198.97|    195.99|         3.99|     Clay Rozendal| Nunavut|Nunavut|       Corporate|      Technology|Telephones and Co...|                R380|        Small Box|               0.58|  7/12/11|
|    85|     515|   8/28/10| Not Specified|            19|   394.27|    0.08|   Regular Air|  30.94|     21.78|         5.94|    Carlos Soltero| Nunavut|Nunavut|        Consumer| Office Supplies|          Appliances|Holmes HEPA Air P...|       Medium Box|                0.5|  8/30/10|
|    86|     515|   8/28/10| Not Specified|            21|   146.69|    0.05|   Regular Air|   4.43|      6.64|         4.95|    Carlos Soltero| Nunavut|Nunavut|        Consumer|       Furniture|  Office Furnishings|G.E. Longer-Life ...|       Small Pack|               0.37|  8/30/10|
|    97|     613|   6/17/11|          High|            12|    93.54|    0.03|   Regular Air| -54.04|       7.3|         7.72|      Carl Jackson| Nunavut|Nunavut|       Corporate| Office Supplies|Binders and Binde...|Angle-D Binders w...|        Small Box|               0.38|  6/17/11|
|    98|     613|   6/17/11|          High|            22|   905.08|    0.09|   Regular Air|  127.7|     42.76|         6.22|      Carl Jackson| Nunavut|Nunavut|       Corporate| Office Supplies|Storage & Organiz...|SAFCO Mobile Desk...|        Small Box|               null|  6/18/11|
|   103|     643|   3/24/11|          High|            21|  2781.82|    0.07|   Express Air|-695.26|    138.14|           35|    Monica Federle| Nunavut|Nunavut|       Corporate| Office Supplies|Storage & Organiz...|SAFCO Commercial ...|        Large Box|               null|  3/25/11|
|   107|     678|   2/26/10|           Low|            44|   228.41|    0.07|   Regular Air|-226.36|      4.98|         8.33|   Dorothy Badders| Nunavut|Nunavut|     Home Office| Office Supplies|               Paper|           Xerox 198|        Small Box|               0.38|  2/26/10|
+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+
only showing top 10 rows

In order to use SQL commands in SparkSQL we’ll need to register the data as a table:

In [19]:
df.registerTempTable("Superstore")

Finally, we can use SQL commands by invoking the spark.sql function and our command in quotes as an argument:

In [20]:
result = spark.sql("SELECT * FROM Superstore LIMIT 5").show()

+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+
|Row ID|Order ID|Order Date|Order Priority|Order Quantity|    Sales|Discount|     Ship Mode| Profit|Unit Price|Shipping Cost|     Customer Name|Province| Region|Customer Segment|Product Category|Product Sub-Category|        Product Name|Product Container|Product Base Margin|Ship Date|
+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+
|     1|       3|  10/13/10|           Low|             6|   261.54|    0.04|   Regular Air|-213.25|     38.94|           35|Muhammed MacIntyre| Nunavut|Nunavut|  Small Business| Office Supplies|Storage & Organiz...|Eldon Base for st...|        Large Box|                0.8| 10/20/10|
|    49|     293|   10/1/12|          High|            49| 10123.02|    0.07|Delivery Truck| 457.81|    208.16|        68.02|      Barry French| Nunavut|Nunavut|        Consumer| Office Supplies|          Appliances|"1.7 Cubic Foot C...|       Jumbo Drum|               0.58|  10/2/12|
|    50|     293|   10/1/12|          High|            27|   244.57|    0.01|   Regular Air|  46.71|      8.69|         2.99|      Barry French| Nunavut|Nunavut|        Consumer| Office Supplies|Binders and Binde...|Cardinal Slant-D®...|        Small Box|               0.39|  10/3/12|
|    80|     483|   7/10/11|          High|            30|4965.7595|    0.08|   Regular Air|1198.97|    195.99|         3.99|     Clay Rozendal| Nunavut|Nunavut|       Corporate|      Technology|Telephones and Co...|                R380|        Small Box|               0.58|  7/12/11|
|    85|     515|   8/28/10| Not Specified|            19|   394.27|    0.08|   Regular Air|  30.94|     21.78|         5.94|    Carlos Soltero| Nunavut|Nunavut|        Consumer| Office Supplies|          Appliances|Holmes HEPA Air P...|       Medium Box|                0.5|  8/30/10|
+------+--------+----------+--------------+--------------+---------+--------+--------------+-------+----------+-------------+------------------+--------+-------+----------------+----------------+--------------------+--------------------+-----------------+-------------------+---------+

 

 

Importing Word into RapidMiner

word into rapidminer

 

On a project for a recent client I needed to apply some common Natural Language Processing (NLP) techniques to surveys they had gathered, but one of the requirements for the project was that the source document had to remain in Word’s .docx format and couldn’t be exported to .txt. RapidMiner was the tool of choice for this engagement since it is graphical in nature and has a very usable library for text analysis, but what it doesn’t have is an operator that specifically imports .docx files. So I needed to import Word into Rapidminer.

Microsoft Word files are basically zip files that contain an XML representation of the actual document. It stands to reason that if you can unzip the wrapper and get to the XML inside, you have a good chance of being able to read the document and do whatever you need in terms of analysis. RapidMiner has an operator for executing custom Python scripts (if you download the Python extension), so I chose to start there and see if it could handle those tasks.

First we’ll need to download the Python extension, which you can do by going to Extensions–>Marketplace in the menu at the top of the page. It’s one of the most popular downloads, so just go to “Top Downloads,” select it from the list, and click “Install Packages” at the bottom of the window. You’ll need to restart RapidMiner afterwards for the extension’s operators to become available.

rapidminer python extension

To use a custom Python script, search for the “Execute Python” operator and drag it onto the workflow. Double-click and you’ll see the usual parameter editing box on the top right of the screen, which should contain a button labeled “Edit Text.” This is where we’ll enter the code.

rapidminer python operator

I try not to reinvent the wheel when coding, so I Googled the problem to see if someone had tackled it before me and someone definitely had. The code I used is below:

If you want to download it straight from Etienne’s blog, just follow this link:

http://etienned.github.io/posts/extract-text-from-word-docx-simply

The initial workflow looked like this:

rapidminer python workflow

After using Etienne’s code to unwrap the .docx file, it was easily readable by the “Read Document” operator. After that I transformed all words to lowercase, tokenized them, removed stop words, then converted the resulting word list to data and loaded it into a database for analysis. Simple.

Using Seahorse for Spark on a Cloudera HA Cluster

I’m loving Seahorse, a GUI frontend for Spark by deepsense.io. The interface is simple, elegant, and beautiful, and has the potential to significantly speed up development on a machine learning workflow by its drag-and-drop nature. Thus far I haven’t run into any major bugs that affect the results so naturally that shoots it near the top of my list for “useful things.”

 

 

That said, the lab cluster here at Brilliant Data is a highly-available Cloudera CDH installation, and at first getting Seahorse to work was a frustrating experience. Turns out this is due to a bug in how Cloudera handles configuration files in HA environments (see this link for details).

The fix is simple enough, but not all that intuitive.

When you first install Seahorse the instructions ask you to place copies of your Hadoop configuration files (yarn-site.xml, core-site.xml, etc) into a directory called “MyYARN” in your Seahorse home folder. Many people would think to copy those from the cluster using scp or something similar, but this can backfire if you pull them from the wrong directory (and yes, there is a wrong directory in a Cloudera HA cluster.)

First, open Cloudera Manager and go to download page for the version of the config files that Cloudera uses by clicking on the menu next to the cluster name and selecting “View Client Configuration URLs.”

select_config

 

That will pop-up a list of the options. We want the YARN config files so select those, which by default will download the zip file to the Downloads folder on the client machine.

 

select_yarn

Use your favorite method to transfer the zip file to the machine running Seahorse, unzip the file, and transfer only the .xml files to /<seahorse_home>/data/MyYARN. Restart the Docker container for Seahorse and everything should work fine now.

Have fun with your new GUI for Spark.