Book Report: Pandas Workout

Unlike many Data Scientists, I didn't automatically reach for Pandas when I needed to analyze data. I came upon this discipline (Data Science) as a Java Software Engineer who used Python for scripting, so I was quite comfortable operating on JSON / CSV / text files directly, loading data into relational databases and running SQL against them, and building visualizations with Matplotlib. So when Pandas first hit the scene, I thought it was a nice library, but I just didn't see the logic in spending time to learn another interface to do the same things I could do already. Of course, Pandas has matured since then (and so have I, hopefully), and when faced with a data analysis / preparation / cleanup task, I often now reach out not only for Pandas, but depending on the task, also its various incarnations such as PySpark, Dask Dataframes and RAPIDS cuDF. When I use Pandas (and its various incarnations) I often find myself depending heavily on Stack Overflow (and lately Github Copilot) for things I know can be done but not how. To some extent I blame this on never having spent the time to understand Pandas in depth. So when I was offered the chance to review Pandas Workout by Reuven Lerner, I welcomed it as a way to remedy this gap in my knowledge.

The book is about Pandas fundamentals rather than solving specific problems with Pandas. For that you will still want to look up Stack Overflow :-). In fact, in the foreword the author specifically targets my demographic (needs to look up Stack Overflow when solving problems with Pandas). But he promises that after reading the book you will understand why some solutions are better than others.

Pandas started as an open source project by Wes McKinney, and has grown somewhat organically into the top Data Science toolkit that is today. As a result, there are often multiple ways to do something in Pandas. While all these ways may produce identical results, their performance characteristics may be different, so there is usually an implicit "right" way. The book gives you the mental model to decide which among the different approaches is the "right" one.

The book is organized into the following chapters. Each chapter covers a particular aspect of Pandas usage. I have included a super-short TLDR style abstract for each chapter for your convenience.

1. Series -- Pandas Series objects are the basic building block of Pandas and represent a typed sequence of data, that are used to construct DataFrames and Indexes. Many methods on the Series object apply in a similar way to DataFrames as well. This is a foundational chapter, understanding this will help with future chapters.
2. Data Frames -- DataFrames represent tabular data as a sequence of Series, where each Series object represents a column in the table. Pandas inherits the idea of DataFrames from R, and the incarnations I listed (and a few that I didn't) use DataFrame as a basic abstraction as well. This chapter teaches you how to select from and manipulate DataFrames. Unless you've used Pandas extensively before, there is a high chance you will learn something useful new tricks here (I did, several of them).
3. Import and Export -- covers reading and writing CSV and JSON formats to and from DataFrames. Covers some simple sanity checks you can run to verify that the import or export worked correctly. I learned about the pd.read_html method here, probably not that useful, but interesting to know!
4. Indexes -- Indexes are used by Pandas to efficiently find data in DataFrames. While it may be possible to get by without Indexes, your Pandas code would take longer to run and consume more resources. The chapter deals with indexing techniques. I happened to know a lot of them, but there were a few that I didn't, especially the techniques around pivot tables.
5. Cleaning -- this chapter teaches a skill that is very fundamental to (and maybe even the bane of) a Data Scientist's job. Statistics indicate that we spend 80% of our time cleaning data. Along with the techniques themselves (remove / interpolate / ignore), this chapter contains commentary that will help you frame these decisions on your own data cleaning tasks.
6. Grouping, Joining and Sorting -- these three operations are so central to data analysis, so much so that SQL has special keywords for each operation (JOIN, GROUP BY and ORDER BY). This chapter covers various recipes to do these operations efficiently and correctly in Pandas.
7. Advanced Grouping, Joining and Sorting -- this chapter goes into greater detail on how to combine these operations to deal with specific use-cases, the so-called "split-apply-combine" technique, including the concept of a general aggregation function agg. It also shows how to do method chaining using assign.
8. Midway Project -- describes a project and asks questions that you should be able to answer from the data using the techniques you have learned so far. Comes with solutions.
9. Strings -- one reason I don't have much experience with Pandas is because it is focused on numeric tables for the most part. However, Pandas also has impressive string handling facilities via the str accessor. This chapter was something of an eye-opener for me, showing me how to use Pandas for text analysis and pre-processing.
10. Dates -- this chapter describes Pandas date and time handling capabilities. This can be useful when trying to work with time series or when trying to derive numerical features from columns containing datetime objects to combine with other numeric or text data.
11. Visualizations -- this chapter describes visualization functionality you can invoke from within Pandas, that are powered either by Matplotlib or Seaborn. This is more convenient than exporting the data to Numpy and using the two packages to draw the charts.
12. Performance -- performance has been a focus for most of the preceding chapters in this book. However, the recipes in this chapter are in the advanced tricks category, and include converting strings to categorical values, optimizing reads and writes using Apache Arrow backed formats, and the using fast special purpose functions for specific purposes.
13. Final Project -- describes a project similar to the Midway project with questions that you should be able to answer from the data using the techniques you have learned so far.

I think the book has value beyond just teaching Pandas fundamentals though. The author sprinkles insights about Data Analysis and Data Science throughout the book, around learning to structure the problem and planning the sequence of steps that are best suited for the tools at hand, the importance of critical thinking, the importance of knowing the data and interpreting the results of the analysis, etc.

Each exercise (there are 50 in all) involves downloading some dataset, dealing with subjects as diverse as tourism, taxi rides, SAT scores, parking tickets, olympic games, oil prices, etc. I think the information about the availability of such datasets (and possibly related datasets) can also be very valuable to Data Scientists for their future projects.

I think the popularity of Pandas is because of the same reason as the popularity of Jupyter Notebooks. It is a nice, self-contained platform the allows a Data Scientist to demonstrate a series of data transformations from problem to solution in a clear, concise and standard manner, not only to customers, but to other Data Scientists as well. More than any other reason, I feel that this will continue to drive the popularity of Pandas and its various incarnations, and as a Data Scientist, it makes sense to learn how to use it properly. And the book definitely fulfils its promise of teaching you how to do that.

Future of Data Centric AI -- Trip Report

I attended the Future of Data Centric AI 2023 this week, a free virtual conference organized by Snorkel AI. Snorkel.AI is a company built around the open-source Snorkel framework for programmatic data labeling. The project originally started at Stanford University's Hazy Research group, and many (all?) of the company's founders and some engineers are from the original research team. Snorkel.AI has been building and improving their flagship product, Snorkel Flow, an integrated tool for iterative data labeling and model building, so there were some presentations centered around that. In addition, its 2023, the year of generative LLMs (or GoLLuMs or Foundation Models) so Snorkel's ability to interface with these Foundation Models (FMs) also featured prominently. Maybe its a Stanford thing but presenters seem to prefer calling them FMs, so I will do the same, if only to distinguish them from the BERT / BART style large language models (LLMs).

If you are unfamiliar with what Snorkel does, I recommend checking out Snorkel and the Dawn of Weakly Supervised Machine Learning (Ratner et al, 2017) for a high-level understanding. For those familiar with the original open source Snorkel (and Snorkel METAL), Snorkel Flow is primarily a no-code web based tool to support the complete life-cycle of programmatic data labeling and model development. Because it is no-code it is usable by domain experts who don't necessarily know how to program. While the suite of built-in no-code Label Function (LF) templates are quite extensive, it supports adding programmatic LFs as well if you need them. In addition, it provides various conveniences such as cold-start LF recommendations and error analysis and recipes on how to address various classes of error to support an iterative approach to do model development almost like a programmer's edit-compile-run cycle. Over the last few months, they have added LLMs as another source of weak supervision and a possible source of LFs as well.

The last bit is important, because I think it points to the pragmatism of the Snorkel team. The FM applications ecosystem currently seems filled with pipelines that feature the FM front and center, i.e. use the FM for everything it can possibly do. Given their high infrastructure costs to run them and their high latencies, these pipelines don't seem very practical. Most of us were taught to cache (or pre-cache) as much as possible, so the customer does not pay the price during serving, or they will soon cease to be customers. Matthew Honnibal, creator of Spacy, makes a similar, though probably better argued, point in his Against LLM Maximalism blog post, where he advocates for smaller, more reliable, models for most tasks in the pipeline, and reserving the FM for tasks that truly need its capabilities. Snorkel Flow goes one step further by taking them out of the pipeline altogether -- instead using them to help generate good labels, thus benefiting from the FMs world-knowledge while still retaining the flexibility, reliability and explainability in the generated models.

However, Snorkel.AI is addressing the needs of the FM market as well, through their soon to be announced new tools -- Foundry and GenFlow -- which Alex Ratner (CEO and co-founder of Snorkel.AI) mentioned in his keynote addresses. They classify the usage of FMs into four stages -- pre-training (either from scratch or from trained weights, where it becomes more of a domain adaptation exercise), instruction tuning for behavior, fine tuning for a particular task, and distillation of the model into a smaller, more easily deployable model. As the DoReMi: Optimizing Data Mixtures Speeds Up Language Model Pretraining (Xie et al, 2023) paper shows, the mix of data used to train or adapt the FM can make a significant impact upon its quality, and Foundry and GenFlow are aimed at improving data and prompt quality for the first and second stages respectively, by ensuring optimum sampling, filtering and ranking.

Over the course of the presentation, presenters repeatedly talked about the importance of having high quality data to train models. Not surprising, since the conference has "Data-Centric AI" in its name, a term coined by Andrew Ng who was the first to emphasize this idea. However, the Snorkel team have really taken this idea to heart, and along with their customers, have developed some really cool applications, some of which they showcased in this conference. Apart from the keynotes and some panel discussions, presentations were in two parallel tracks, and I chose the ones that emphasized practice over theory, and I skipped a few, so the list below may be slightly biased. Videos of the talks will become available on the Snorkel Youtube channel in about a month, I will update the links once that happens (if I remember).

• Bridging the Last Mile: Applying Foundation Models with Data-Centric AI (Alex Ratner) -- basic idea is that FMs are analogous to generalists that (think they) know lots of things, but for specific tasks they need to be trained to do well. Alex envisions data scientists of the future that are less machine learning experts and more domain and product experts. Alex's talks contain many interesting observations, too numerous to list here, and its just the right mixture of academic and practical for lay people such as myself.
• Fireside Chat: building Bloomberg GPT (Gideon Mann and Alex Ratner) -- interesting insights into the rationale for Bloomberg GPT and the work that went into building it.
• Fireside Chat: Stable Diffusion and Generative AI (Emad Mostaque and Alex Ratner) -- lot of cool technical insights about FMs from Emad Mostaque, CEO of Stability.AI (Stable Diffusion).
• A Practical Guide to Data Centric AI -- A Conversational Use AI Use case (Daniel Lieb and Samira Shaikh) -- practical tips to building an intent classifier for conversational chatbots. Similarity function for clustering conversations was adapted from the paper Modeling Semantic Containment and Exclusion in Natural Language Inference (MacCartney and Manning, 2008).
• The Future is Neurosymbolic (Yoav Shoham) -- somewhat philosophical discussion of why FMs can never do the kind of things humans can do, and why, from the founder of AI21 Labs.
• Generating Synthetic Tabular Data that is Differentially Private (Lipika Ramaswamy) -- a somewhat technical discussion arguing for differential privacy to generate synthetic datasets that could be used to train FMs and thereby address the problem of them memorizing sensitive training data.
• DataComp: Significance of Data for Multimodal AI (Ludwig Schmidt) -- discusses DATACOMP, a benchmark which aims to improve an image-text dataset used to train multi-modal models such as CLIP, by keeping the model fixed and improving the dataset. By applying a simple quality filter on the original dataset, they were able to model that was smaller in size, took 7x less time to train, and outperformed a larger model. More details in the DATACOMP: In search of the next generation of multimodal datasets (Gadre et al, 2023) paper.
• New Introductions from Snorkel AI (Alex Ratner) -- second day keynote where Alex formally announced Snorkel Foundry and GenFlow, among other things, some of which were repeats from the previous day's keynote.
• Transforming the Customer Experience with AI: Wayfair's Data Centric Way (Archana Sapkota and Vinny DeGenova) -- this was a really cool presentation, showing how they labeled their product images programatically with Snorkel for design, pattern, shape and theme, and used that to fine tune a CLIP model, which they now use in their search pipeline. More info about this work in this blog post.
• Tackling advanced classification with Snorkel Flow (Angela Fox and Vincent Chen) -- the two big use cases where people leverage Snorkel are document classification and sequence labeling. Here they discuss several strategies for multi-label and single-label document classification.
• Accelerating information extraction with data-centric iteration (John Smardijan and Vincent Chen) -- this presentation has a demo of Snorkel flow to label documents with keywords for a specific use case (for which off the shelf NERs do not exist). The demo shows how one can rapidly reach a good score (precision and coverage) by iterating through creating and applying an LF, then training and evaluating a model on the labels created by the LF, doing error analysis to correct the issues pointed out by creating another LF, etc, until the desired metrics are reached. They called this the Data-Model flywheel.
• Applying Weak Supervision and Foundation Models for Computer Vision (Ravi Teja Mullapudi) -- talked about using Snorkel for image classification, including a really cool demo of Snorkel Periscope (an internal Labs tool) applied to satellite data to build classifiers that look for images of a particular type, using UMAP visualizations and cosine similarity distributions.
• Leveraging Data-Centric AI for Document Intelligence and PDF Extraction (Ashwini Ramamoorthy) -- a talk about information extraction from PDF documents, similar to the one listed earlier, but as with that one, Ashwini shares a huge amount of practical information that I found very useful.
• Leveraging Foundation Models and LLMs for Enterprise Grade NLP (Kristina Lipchin) -- slightly high level but very interesting take on FMs from a product manager viewpoint, echoes much of the same ideas about last mile handling covered in earlier talks, but identifies Domain Adaptation and Distillation as the primary use cases for most organizations.
• Lessons from a year with Snorkel Data-Centric with SMEs and Georgetown (James Dunham) -- this is a hugely informative talk about Georgetown University's experience with using Snorkel Flow for a year. Not only did their domain experts adapt to it readily and love the experience, both data scientists and domain experts benefited from it. Some major benefits noted are the ability to ramp up labeling efforts faster and with less risk, since it is easier to iterate on labels (adding/removing/merging classes, etc) as your understanding of the data grows, the ability to fail fast and without too much sunk cost, and overall lowering of project risk. If you are contemplating purchasing a Snorkel Flow subscription, this talk provides lots of useful information.
• Fireside chat: building RedPajamas (Ce Zheng and Braden Hancock) -- RedPajama is an open source initiative to produce a clean-room reimplementation of the popular LLaMA FM from Meta. The focus is on replicating their dataset recipe carefully, but using open source documents, and training base and instruction tuned versions of the LLaMMA model on this data that does not block commercial adoption. Ce is the head of Together Computer the company behind RedPajama, and Braden and Ce discuss the work that has been done so far in this project.

In many cases, it is not the lack of data, but a lack of labeled data that is the major hurdle to Machine Learning adoption within a company. Snorkel's support for weak supervision provides a practical path to generate labels using a programmatic approach. As someone who came to Machine Learning from Search, where featurization is basically TF-IDF and more lately using a trained tokenizer to feed a neural model, I was initially not particularly skilled at detecting features from data. However, over time, as I started looking at data, initially for error analysis and later for feature extraction in cases where labels were not available apriori, the process has become easier, so hopefully my next experience with Snorkel will be smoother. Furthermore, Snorkel's focus on FMs also provides a path to harness this powerful new resource as an additional source of weak supervision.

Understanding UMLS

I've been looking at Unified Medical Language System (UMLS) data this last week. The medical taxonomy we use at work is partly populated from UMLS, so I am familiar with the data, but only after it has been processed by our Informatics team. The reason I was looking at it is because I am trying to understand Apache cTakes, an open source NLP pipeline for the medical domain, which uses UMLS as one of its inputs.

UMLS is provided by the National Library of Medicine (NLM), and consists of 3 major parts: the Metathesaurus, consisting of over 1M medical concepts, a Semantic Network to categorize concepts by semantic type, and a Specialist Lexicon containing data to help do NLP on medical text. In addition, I also downloaded the RxNorm database that contains drug/medication information. I found that the biggest challenge was accessing the data, so I will describe that here, and point you to other web resources for the data descriptions.

Before getting the data, you have to sign up for a license with UMLS Terminology Services (UTS) - this is a manual process and can take a few days over email (I did this couple of years ago so details are hazy). UMLS data is distributed as .nlm files which can (as far as I can tell) be opened and expanded only by the Metamorphosis (mmsys) downloader, available on the UMLS download page. You need to run the following sequence of steps to capture the UMLS data into a local MySQL database. You can use other databases as well, but you would have to do a bit more work.

1. Download the Metamorphosis (mmsys) tool. Navigate to the UMLS download page and click the link for mmsys.zip. Installation consists of unzipping it into some convenient directory. For example, if you installed under /opt, your mmsys install directory would be /opt/mmsys.
2. Download the additional data files into your mmsys working directory. I chose the 2013AB UMLS Active Release Files set.
3. Start up the mmsys tool by running the ./run_linux.sh script. A GUI screen appears - click the Install UMLS button. You will be prompted for an output directory location - I chose /opt/mmsys/data.
4. The .nlm data is expanded out into three subdirectories under /opt/mmsys/data/2013AB - META, NET and LEX, which correspond to data for the UMLS Metathesaurus, Semantic Network and Specialist Lexicon respectively. Within each subdirectory, data is provided as .RRF files (basically pipe delimited text files).
5. Login to the MySQL client and create a database to hold this data. The command is: CREATE DATABASE umlsdb DEFAULT CHARACTER SET utf8;
6. The schema and data load script for the Metathesaurus can be generated using the mmsys tool using the top level menu: "Advanced", then "Copy Load Scripts to Hard Drive". You can specify a target database other than MySQL, but for the other data, you would have to adapt the provided MySQL schema and data load scripts.
7. Copy the generated script files to the META subdirectory. Update the populate_mysql_db.sh file with the MySQL root directory (/usr for me - MySQL was installed using Ubuntu's apt-get install), the database name, the database user and password. I also had to add --local-infile=1 to the mysql calls in the script because my server was not built to allow data using LOAD LOCAL INFILE. This will take a while (I left mine running over a weekend, it took more than a day to load and build the indexes on a 4 CPU box).
8. The NET subdirectory comes with its own populate_net_mysql_db.sh script. As before, the MySQL root directory, database name, database user and password need to be updated into the script, as well as pass --local-infile=1 to the mysql calls. The data loads relatively quickly, takes about 5 minutes or so.
9. The schema and data loading SQL for the LEX subdirectory can be found on this page, which also provides instructions similar to my post. Unlike the previous ones, you need to log into the MySQL client using --local-infile=1 at the mysql command, change to your database, and run the TODO script (that you downloaded) using "SOURCE mysql_lex_tables.sql;". This is also very quick, takes about 5-10 minutes.
10. For RxNorm, I downloaded the RxNorm_full_01062014.zip file and unzipped it into my /opt/mmsys/data/RxNorm directory. I then navigated to the scripts/mysql subdirectory, and updated the populate_mysql_rxn.sh script to add the MySQL root directory, database name, database user, password and dbserver (localhost). I also set the --local-infile=1 flag to the mysql command. Since the RRF files are in a different directory, all RRF file references in Load_scripts_mysql_rxn_unix.sql need to be offset by ../../rrf/. The load process runs for about 15 minutes.

One thing to note is that the database is not normalized. Information is repeated across tables and presented in different formats. The user of the data must decide how to handle this for his/her application. So what you do to reorganize the data is very much application-dependent. I actually tried to generate a database schema using SQLFairy in XFig format and modify it in Dia before I realized the futility of this exercise.

The table and column names are quite cryptic and the relationships are not evident from the tables. You will need to refer to the data dictionaries for each system to understand it before you do anything interesting with the data. Here are the links to the online references that describe the tables and their relationships for each system better than I can.

• Metathesaurus RRF manual
• Semantic Network Data Description
• Specialist Lexicon Data Description
• RxNorm data description

The tables in the Metathesaurus that are important from the cTakes point of view are MRCONSO and MRSTY, which contain information about concepts and synonymous terms and semantic types, respectively. Other tables that are important if you are looking for relationships between concepts are MRREL and MRHIER. Co-occurrences of concepts in external texts are found in MRCOC, and MRMAP and MRSMAP contains mappings between terminologies (a very complex mapping described by the docs). RxNorm seems to be structured similarly as the Metathesaurus, except the contents are drugs (although I haven't looked at RxNorm much yet, so this may not be completely accurate). For example the RxNorm analogs of the Metathesaurus tables MRCONSO and MRSTY are RXNCONSO and RXNSTY respectively.

This exercise in trying to understand the UMLS data was quite interesting. While there is some intersection between what we use at work and whats available, there is a lot we don't yet use and which can potentially be used in many interesting ways. In retrospect, I wish I had done this sooner.

Akka Content Ingestion Pipeline, Part I

I just finished attending Coursera's Reactive Programming classes in "spectator" mode (just watched the videos). The course was conducted by Martin Odersky (creator of Scala), Eric Meijer (creator of LINQ, he teaches about Monads) and Roland Kunh (Akka Tech Lead). My main draw for the course was the coverage of Akka Actors, something I have been intending to learn for a while, although I learned a lot from the other lectures as well. I first came across Scala Actors 5 years ago, but I didn't pursue it, mainly because the parallelism the approach offered was limited to a single JVM (implying a single large machine rather than many small machines). At the time, the Akka project was just getting started.

Today, Akka allows you to deploy actors across multiple JVMs on multiple machines in a network, is available in Scala and Java (important for addressing maintainability concerns in Java-only shops like mine), and provides additional supporting infrastructure via the Typesafe stack. It has progressed to the extent that it is the preferred Actor implementation for Scala 2.10+. It has a vibrant community and (reasonably) good documentation, so its not too hard to get started using it.

The example I choose as a vehicle for learning Akka is based on the Nutch pipeline. Its a pipeline I am very familiar with, we run a variant of this at work for our own content ingestion. Nutch runs on Hadoop as a series of Map-Reduce batch jobs, first fetching the pages, then parsing out key-value pairs out of them, and finally sending the key-value pairs and content off to the indexer so it can be searched by clients. Additionally (with NutchGORA) the data is persisted into a NoSQL database during the fetch and parse steps, so it can potentially be used as a content service as well. The example is non-trivial, so I decided to build it in steps and describe the evolution of this system across multiple posts, rather than describing the whole thing in one giant post 4 weeks later. Makes it easier for both of us.

The diagram below shows the actors and the message flows in our example system. There are 2 message flows (indicated by blue and red text and arrows). This is a message-passing model and looks more like the example I used for comparing various actor implementations 5 years ago than the NutchGORA model, but the business process is the same.

The top-level actor in our system is the Controller. The controller is the actor that other actors or callers from the outside interact with. The controller spawns three router actors on startup - the fetchersRouter, parsersRouter and the indexersRouter, which in turn spawn a fixed number (based on configuration) worker actors. In addition, the Controller also starts up a Reaper actor and registers the routers with it. All these actor startup is indicated by dotted green lines. From a class structure point of view, this means that the Controller, routers and the Reaper can refer to each other using references (without having to look it up from the context). The worker actors communicate only with its parent routers.

Our first message flow is the Fetch message. A Fetch message includes the URL to fetch, the current fetch depth, and any metadata included with the URL. The fetch depth is important for web crawling, where a depth > 0 indicates that outlinks must be crawled. The metadata is important for situations where you are parsing feeds and you want to carry over the title and summary from the feed rather than (or in addition to) parsing it, or supply additional data such as file create/modify dates for when you are fetching files from the local filesystem. We describe the Fetch message flow below:

1. Fetch message is sent to the controller.
2. Controller forwards the message to the Fetcher Router.
3. Fetcher Router forwards the message to one of the workers, using Round Robin routing policy.
4. If the URL is eligible to be downloaded, the Fetcher Worker downloads the URL and writes the contents and metadata into the database. Once done, it sends a FetchComplete message to the Fetcher Router that includes the database ID of the inserted record.
5. This results in the Controller receiving a FetchComplete message, to which it responds by sending a Parse message to the Parser Router.
6. The Parser Router forwards the Parse message to one of its workers.
7. The Parse Worker retrieves the contents of the file from the database using the ID, converts the file to text and parses relevant key-value pairs from it. It then writes these key-value pairs and the text content back to the database. Once done, it sends a ParseComplete message back to its router.
8. If the depth > 0, the parsing process also involves parsing the content for embedded outbound links, which are enqueued as additional Fetch requests on the Fetch Router.
9. This results in the Controller receiving a ParseComplete message, to which it responds by sending an Index message to the Index Router.
10. The Index Router forwards the Index message to one of its workers.
11. The Index Worker retrieves the key-value pairs from the database and publishes the record to a Solr index. Once done, it sends back an IndexComplete message to its parent.

The other message to handle is the Stop message, which allows actors to consume all messages that are enqueued currently, then shut them down. Here is how that works.

1. Stop message is sent to the Controller.
2. The Controller forwards the Stop Message to the Reaper. At startup, each of the routers were registered with the Reaper, so the Reaper adds their references to a List and begins monitoring them for Termination (DeathWatch).
3. The Reaper sends a PoisonPill message wrapper in a Broadcast message to the FetchRouter. Using the Broadcast wrapper ensures that the router sends a PoisonPill to each of its workers, not just the next one. A PoisonPill is placed at the end of each Worker's queue. After this, no new messages can be placed on these queues. The Workers continue to process messages till their queue is drained and then terminates. When all workers are terminated, the router terminates.
4. Because the Reaper is watching the FetchRouter, the Reaper gets a Terminated message from it, and reacts by removing the router's reference from its list. It then sends a PoisonPill Broadcast message to the next reference on its list, the ParserRouter.
5. Like the FetcherRouter, the ParserRouter terminates its Workers and then itself.
6. The Reaper gets a Terminated message from the ParserRouter, and removes it from its list, then sends a PoisonPill Broadcast to the IndexRouter.
7. Like the FetcherRouter and the ParserRouter, the IndexRouter too terminates its workers and then itself.
8. The Reaper gets a Terminated message from the IndexRouter, and removes it from its list.
9. Because its ActorRef list is now empty, the Reaper shuts down the Controller. At this point, the system has no more actors, so it shuts down also.

In addition, our system also supports a Stats message (not shown in diagram) which returns the sizes of the three process "queues". This is done by incrementing and decrementing a set of counters each time we recieve a Fetch/Parse/Index and FetchComplete/ParseComplete/IndexComplete message at the Controller respectively.

We can also send Parse or Index messages directly to the controller. Haven't thought through this completely, but we could probably also supply metadata parameters to skip certain operations, thus providing more flexibility.

Here is the code to support this functionality. First we define our messages. All our messages except for the Stop have arguments, and are hence defined as Case classes. Stop is defined as a Case object. We extend a sealed trait marker interface to prevent outside code from adding new messages.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

// Source: src/main/scala/com/mycompany/delsym/actors/DelsymMessage.scala package com.mycompany.delsym.actors import akka.actor.ActorRef sealed trait DelsymMessage //////// messages sent from outside to controller ///////// case class Fetch(url: String, depth: Int, metadata: Map[String,Any]) extends DelsymMessage case class Stats(stats: Map[String,Int]) extends DelsymMessage case object Stop extends DelsymMessage case class Register(ref: ActorRef) extends DelsymMessage ////////// messages between supervisor and worker ////////// case class Parse(id: String) extends DelsymMessage case class Index(id: String) extends DelsymMessage case class FetchComplete(id: String) extends DelsymMessage case class ParseComplete(id: String) extends DelsymMessage case class IndexComplete(id: String) extends DelsymMessage

Our next class is the Controller. The Controller instantiates the Reaper and the three Routers, then registers the Routers with the Reaper. It also sets up the counters to support the Stats message.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102

// Source: src/main/scala/com/mycompany/delsym/actors/Controller.scala package com.mycompany.delsym.actors import scala.concurrent.duration.DurationInt import com.typesafe.config.ConfigFactory import akka.actor.Actor import akka.actor.ActorLogging import akka.actor.OneForOneStrategy import akka.actor.Props import akka.actor.SupervisorStrategy import akka.actor.actorRef2Scala import akka.routing.RoundRobinRouter class Controller extends Actor with ActorLogging { override val supervisorStrategy = OneForOneStrategy( maxNrOfRetries = 10, withinTimeRange = 1.minute) { case _: Exception => SupervisorStrategy.Restart } val reaper = context.actorOf(Props[Reaper], name="reaper") val config = ConfigFactory.load() val numFetchers = config.getInt("delsym.fetchers.numworkers") val numParsers = config.getInt("delsym.parsers.numworkers") val numIndexers = config.getInt("delsym.indexers.numworkers") val queueSizes = scala.collection.mutable.Map[String,Int]() val restartChild = OneForOneStrategy() { case e: Exception => SupervisorStrategy.Restart } val fetchers = context.actorOf(Props[FetchWorker] .withRouter(RoundRobinRouter(nrOfInstances=numFetchers, supervisorStrategy=restartChild)), name="fetchers") reaper ! Register(fetchers) queueSizes += (("fetchers", 0)) val parsers = context.actorOf(Props[ParseWorker] .withRouter(RoundRobinRouter(nrOfInstances=numParsers, supervisorStrategy=restartChild)), name="parsers") reaper ! Register(parsers) queueSizes += (("parsers", 0)) val indexers = context.actorOf(Props[IndexWorker] .withRouter(RoundRobinRouter(nrOfInstances=numIndexers, supervisorStrategy=restartChild)), name="indexers") reaper ! Register(indexers) queueSizes += (("indexers", 0)) def receive = { case m: Fetch => { increment("fetchers") fetchers ! m } case m: FetchComplete => { decrement("fetchers") parsers ! Parse(m.id) } case m: Parse => { increment("parsers") parsers ! m } case m: ParseComplete => { decrement("parsers") outlinks(m.id).map(outlink => fetchers ! Fetch(outlink._1, outlink._2, outlink._3)) indexers ! Index(m.id) } case m: Index => { increment("indexers") indexers ! m } case m: IndexComplete => { decrement("indexers") } case m: Stats => sender ! queueSize() case Stop => reaper ! Stop case _ => log.info("Unknown message received.") } def queueSize(): Stats = Stats(queueSizes.toMap) def outlinks(id: String): List[(String,Int,Map[String,Any])] = { log.info("TODO: Fetch outlinks for id:{}", id) List() } def increment(key: String): Unit = { queueSizes += ((key, queueSizes(key) + 1)) } def decrement(key: String): Unit = { queueSizes += ((key, queueSizes(key) - 1)) } }

The Reaper implements the Akka DeathWatch pattern, listening for Termination messages sent by the routers. It implements the process of terminating each router sequentially on receipt of a Stop signal from the client (via the Controller).

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

// Source: src/main/scala/com/mycompany/delsym/actors/Reaper.scala package com.mycompany.delsym.actors import akka.actor.ActorLogging import akka.actor.Actor import java.util.concurrent.atomic.AtomicLong import akka.actor.Terminated import scala.collection.mutable.ArrayBuffer import akka.actor.ActorRef import akka.routing.Broadcast import akka.actor.PoisonPill class Reaper extends Actor with ActorLogging { val refs = ArrayBuffer[ActorRef]() def receive = { case Register(ref) => { context.watch(ref) refs += ref } case Stop => { refs.head ! Broadcast(PoisonPill) } case Terminated(ref) => { val tail = refs.tail if (tail.isEmpty) context.system.shutdown else { refs.clear refs ++= tail refs.head ! Broadcast(PoisonPill) } } case _ => log.info("Unknown message received.") } }

The workers are just stubs at the moment and not that interesting. All they do is log a message saying that they fired their method, implying that the message was received and processed correctly. As an example, we show the FetcherWorker below. Other workers can be found on the GitHub for this project.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

// Source: src/main/scala/com/mycompany/delsym/actors/FetchWorker.scala package com.mycompany.delsym.actors import akka.actor.ActorLogging import akka.actor.Actor import com.typesafe.config.ConfigFactory class FetchWorker extends Actor with ActorLogging { val conf = ConfigFactory.load() def receive = { case m: Fetch => { val id = fetchAndStore(m.url, m.depth, m.metadata) sender ! FetchComplete(id) } case _ => log.info("Unknown message.") } def fetchAndStore(url: String, depth: Int, metadata: Map[String,Any]): String = { log.info("TODO: fetching URL:{}", url) url } }

Akka uses TestKit and ScalaTest for testing. I am still learning these, and my ScalaTest/TestKit foo is not strong enough to write integration tests yet, so I just ran the code (using sbt run) to verify that the flow works. Heres the code for the Main method.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

// Source: src/main/scala/com/mycompany/delsym/actors/Main.scala package com.mycompany.delsym.actors import akka.actor.Props import akka.actor.ActorSystem object Main extends App { val system = ActorSystem("DelsymTest") val controller = system.actorOf(Props[Controller], "controller") (0 until 100).foreach(i => { if (i == 50) controller ! Stats(null) controller ! Fetch(i.toString, 0, Map()) }) controller ! Stop }

In addition to the excellent Reactive Programming course I already cited above, I found the Akka documentation and the Akka in Action book invaluable for figuring out Akka and writing the code above.

Thats all I have for today. As you can see, Akka provides a lot of functionality, so the application code is relatively short and uncomplicated for the functionality it provides. The code for this post is also available on my GitHub repository for my Delsym project.

BTW, if you are curious about the project name, it comes from CONtent inGESTION - Delsym is an over the counter Cough medicine, and so makes CON(tent) (in)GESTION GO away FASTER. Yes, I know, a bit far fetched, but its my project and I am sticking to the name :-).

ETL Case Study using Kettle

ETL (Extract, Transform, Load) has traditionally been the domain of data warehousing practitioners, but it can be applied to any process or set of processes that load data into databases. Data is the lifeblood of any organization. However, data by itself is not too interesting - what is interesting is the information that the data can be processed into. Many enterprise systems dedicate a significant chunk of their functionality and resources to developing programs and scripts that transform and migrate data from one form to another, so the downstream module can present it in a manner more intuitive to their clients.

Writing data transformation routines may be pervasive, but the actual transformation code is generally not very challenging. More challenging is splitting up the transformation into multiple threads and running them in parallel, since ETL jobs usually work with large data sets, and we want the job to complete in a reasonable time. Business application developers generally don't do multithreaded programming too well, mainly because they don't do it often enough. Furthermore, the transformation business logic is inside the application code, which means it cannot be sanity checked by the business person whose needs drove the transformation in the first place.

I heard about Kettle, an open source ETL Tool, from a colleague at a previous job, where he was using it to automate data transformations to push out denormalized versions of data from backend databases to frontend databases. Unfortunately, I never got a chance to use it at work, but it remained on my ToDo list as something I wanted to learn for later. Kettle started as a project by a single developer, but has since been acquired by Pentaho who sell and support a suite of open source Business Intelligence tools, of which Kettle is one, under a dual open-source/commercial license similar to MySQL.

Early in my career, I worked for the MIS group of a factory that manufactured switchboards. It occured to me that one of the processes for generating monthly factory-wide input costs would be a good candidate to convert to Kettle and understand its functionality. Part of the input costs for the factory for the month were the sum of the actual dollar amount paid out to workers. This was governed by the worker's hourly rate and the number of hours worked. The number of hours were derived from the times recorded when the worker signed in and out of the factory. The values are reported by department. The figure below shows the flow.

To replicate the process, I created a flat file for 5 employees in 2 departments (Electrical and Mechanical) which contained in and out times for these employees over a 30 day period. The original databases involved were dBase-IV and Informix with migration scripts written with Clipper and Informix-4GL, the ones in my case study were PostgreSQL and MySQL. A data flow diagram for the Kettle based solution is shown below:

The input file dump looks like this:

1 2 3 4 5

1000015 I 2006-08-01 08:07:00 1154444820 2000024 I 2006-08-01 08:09:00 1154444940 1000015 O 2006-08-01 16:05:00 1154473500 2000024 O 2006-08-01 16:08:00 1154473680 ...

The tables involved in the HRDB PostgreSQL table look like this:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

hrdb=> \d employee Table "public.employee" Column | Type | Modifiers ---------------+------------------------+----------- emp_id | integer | dept_id | integer | emp_name | character varying(255) | rate_per_hour | numeric(8,2) | hrdb=> \d timecard Table "public.timecard" Column | Type | Modifiers ----------+---------------+----------- emp_id | integer | not null tc_date | character(10) | not null time_in | integer | time_out | integer |

And the MySQL table that is populated as a result of the transformations looks like this:

1 2 3 4 5 6 7

mysql> desc input_cost; +------------+---------------+------+-----+---------+-------+ | Field | Type | Null | Key | Default | Extra | +------------+---------------+------+-----+---------+-------+ | dept_id | int(11) | | | 0 | | | input_cost | decimal(12,2) | YES | | NULL | | +------------+---------------+------+-----+---------+-------+

Kettle comes with four main components - Spoon, Pan, Chef and Kitchen. Spoon is a GUI editor for building data transformations. Pan is a command line tool for running a transformation created with Spoon. Chef is a GUI for building up jobs, which are a set of transformations that should work together, and Kitchen is again a command line tool to run jobs built with Chef.

I balked initially at having to use a GUI to design transformations. I would have preferred a scripting language or some sort of XML configuration to do this, but I guess developers have traditionally not been the target market for ETL tools. And I guess the objective of using Kettle is to not do programming for data transformations, and to a certain extent, scripting is programming. Anyway, using Spoon was pretty straightforward, and I was able to generate three transformations which could be applied to my flat file dump in sequence to produce two rows in the CostingDB MySQL table.

Each Spoon Transformation produces as output a .ktr XML file. It can also write the transformation metadata to a database repository (the recommended option). The first transformation reads the flat file, choosing rows with the "I" flag set (for incoming timecard entry), and inserts it into the HRDB.timecard table. The second transformation reads the flat file a second time, this time choosing rows with the "O" flag set (for outgoing timecard entry) and updates the time_out column in the timecard table. The reason we have two separate transformations instead of having two streams from the filter is because the two streams are going to be multi-threaded and there is no guarantee that an insert would complete before the corresponding update is applied.

<?xml version="1.0" encoding="UTF-8"?> <transformation> <info> <name/> <directory>/</directory> <log> <read/> <write/> <input/> <output/> <update/> <connection/> <table/> <use_batchid>Y</use_batchid> <use_logfield>N</use_logfield> </log> <maxdate> <connection/> <table/> <field/> <offset>0.0</offset> <maxdiff>0.0</maxdiff> </maxdate> <size_rowset>1000</size_rowset> <sleep_time_empty>1</sleep_time_empty> <sleep_time_full>1</sleep_time_full> <dependencies> </dependencies> </info> <notepads> </notepads> <connection> <name>HRDB</name> <server>localhost</server> <type>POSTGRESQL</type> <access>Native</access> <database>hrdb</database> <port>5432</port> <username>hruser</username> <password>Encrypted 2be98afc86aa7f2e4cb79a662cb81fcc8</password> <servername/> <data_tablespace/> <index_tablespace/> <attributes> <attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute> <attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute> <attribute><code>PORT_NUMBER</code><attribute>5432</attribute></attribute> </attributes> </connection> <order> <hop> <from>A</from><to>B</to><enabled>Y</enabled> </hop> <hop> <from>B</from><to>C</to><enabled>Y</enabled> </hop> <hop> <from>B</from><to>D</to><enabled>Y</enabled> </hop> </order> <step> <name>A</name> <type>TextFileInput</type> <description>Timecard Monthly Dump</description> <distribute>Y</distribute> <copies>1</copies> <accept_filenames>N</accept_filenames> <accept_field/> <accept_stepname/> <separator> </separator> <enclosure/> <enclosure_breaks>N</enclosure_breaks> <escapechar/> <header>N</header> <nr_headerlines>1</nr_headerlines> <footer>N</footer> <nr_footerlines>1</nr_footerlines> <line_wrapped>N</line_wrapped> <nr_wraps>1</nr_wraps> <layout_paged>N</layout_paged> <nr_lines_per_page>80</nr_lines_per_page> <nr_lines_doc_header>0</nr_lines_doc_header> <noempty>Y</noempty> <include>N</include> <include_field/> <rownum>N</rownum> <rownum_field/> <format>Unix</format> <encoding/> <file> <name>usr/costing/timecard-200608.txt</name> <filemask/> <file_required/> <type>Fixed</type> <zipped>N</zipped> </file> <filters> </filters> <fields> <field> <name>emp_no</name> <type>Integer</type> <format>0000000000000</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>0</position> <length>7</length> <precision>0</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>input_or_output</name> <type>String</type> <format>#</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>8</position> <length>1</length> <precision>0</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>rec_date</name> <type>String</type> <format>yyyy-MM-dd</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>10</position> <length>10</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>rec_time</name> <type>String</type> <format>HH:mm:ss</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>21</position> <length>8</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>epoch_sec</name> <type>Integer</type> <format>0000000000000</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>30</position> <length>10</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> </fields> <limit>0</limit> <error_ignored>N</error_ignored> <error_line_skipped>N</error_line_skipped> <error_count_field/> <error_fields_field/> <error_text_field/> <bad_line_files_destination_directory/> <bad_line_files_extension>warning</bad_line_files_extension> <error_line_files_destination_directory/> <error_line_files_extension>error</error_line_files_extension> <line_number_files_destination_directory/> <line_number_files_extension>line</line_number_files_extension> <date_format_lenient>Y</date_format_lenient> <date_format_locale>en_us</date_format_locale> <GUI> <xloc>99</xloc> <yloc>165</yloc> <draw>Y</draw> </GUI> </step> <step> <name>B</name> <type>FilterRows</type> <description/> <distribute>Y</distribute> <copies>1</copies> <send_true_to>C</send_true_to> <send_false_to>D</send_false_to> <compare> <condition> <negated>N</negated> <leftvalue>input_or_output</leftvalue> <function>=</function> <rightvalue/> <value> <name>constant</name><type>String</type><text>I</text><length>1</length><precision>-1</precision><isnull>N</isnull> </value> </condition> </compare> <GUI> <xloc>198</xloc> <yloc>165</yloc> <draw>Y</draw> </GUI> </step> <step> <name>C</name> <type>InsertUpdate</type> <description>C</description> <distribute>Y</distribute> <copies>1</copies> <connection>HRDB</connection> <commit>0</commit> <update_bypassed>N</update_bypassed> <lookup> <table>timecard</table> <key> <name>emp_no</name> <field>emp_id</field> <condition>=</condition> <name2/> </key> <key> <name>rec_date</name> <field>tc_date</field> <condition>=</condition> <name2/> </key> <value> <name>emp_id</name> <rename>emp_no</rename> <update>Y</update> </value> <value> <name>tc_date</name> <rename>rec_date</rename> <update>Y</update> </value> <value> <name>time_in</name> <rename>epoch_sec</rename> <update>Y</update> </value> </lookup> <GUI> <xloc>272</xloc> <yloc>113</yloc> <draw>Y</draw> </GUI> </step> <step> <name>D</name> <type>Dummy</type> <description>D</description> <distribute>Y</distribute> <copies>1</copies> <GUI> <xloc>274</xloc> <yloc>215</yloc> <draw>Y</draw> </GUI> </step> </transformation>

<?xml version="1.0" encoding="UTF-8"?> <transformation> <info> <name/> <directory>/</directory> <log> <read/> <write/> <input/> <output/> <update/> <connection/> <table/> <use_batchid>Y</use_batchid> <use_logfield>N</use_logfield> </log> <maxdate> <connection/> <table/> <field/> <offset>0.0</offset> <maxdiff>0.0</maxdiff> </maxdate> <size_rowset>1000</size_rowset> <sleep_time_empty>1</sleep_time_empty> <sleep_time_full>1</sleep_time_full> <dependencies> </dependencies> </info> <notepads> </notepads> <connection> <name>HRDB</name> <server>localhost</server> <type>POSTGRESQL</type> <access>Native</access> <database>hrdb</database> <port>5432</port> <username>hruser</username> <password>Encrypted 2be98afc86aa7f2e4cb79a662cb81fcc8</password> <servername/> <data_tablespace/> <index_tablespace/> <attributes> <attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute> <attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute> <attribute><code>PORT_NUMBER</code><attribute>5432</attribute></attribute> </attributes> </connection> <order> <hop> <from>A</from><to>B</to><enabled>Y</enabled> </hop> <hop> <from>B</from><to>D</to><enabled>Y</enabled> </hop> <hop> <from>B</from><to>C</to><enabled>Y</enabled> </hop> </order> <step> <name>A</name> <type>TextFileInput</type> <description>Timecard Monthly Dump</description> <distribute>Y</distribute> <copies>1</copies> <accept_filenames>N</accept_filenames> <accept_field/> <accept_stepname/> <separator> </separator> <enclosure/> <enclosure_breaks>N</enclosure_breaks> <escapechar/> <header>N</header> <nr_headerlines>1</nr_headerlines> <footer>N</footer> <nr_footerlines>1</nr_footerlines> <line_wrapped>N</line_wrapped> <nr_wraps>1</nr_wraps> <layout_paged>N</layout_paged> <nr_lines_per_page>80</nr_lines_per_page> <nr_lines_doc_header>0</nr_lines_doc_header> <noempty>Y</noempty> <include>N</include> <include_field/> <rownum>N</rownum> <rownum_field/> <format>Unix</format> <encoding/> <file> <name>usr/costing/timecard-200608.txt</name> <filemask/> <file_required/> <type>Fixed</type> <zipped>N</zipped> </file> <filters> </filters> <fields> <field> <name>emp_no</name> <type>Integer</type> <format>0000000000000</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>0</position> <length>7</length> <precision>0</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>input_or_output</name> <type>String</type> <format>#</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>8</position> <length>1</length> <precision>0</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>rec_date</name> <type>String</type> <format>yyyy-MM-dd</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>10</position> <length>10</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>rec_time</name> <type>String</type> <format>HH:mm:ss</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>21</position> <length>8</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> <field> <name>epoch_sec</name> <type>Integer</type> <format>0000000000000</format> <currency/> <decimal/> <group/> <nullif/> <ifnull/> <position>30</position> <length>10</length> <precision>-1</precision> <trim_type>none</trim_type> <repeat>N</repeat> </field> </fields> <limit>0</limit> <error_ignored>N</error_ignored> <error_line_skipped>N</error_line_skipped> <error_count_field/> <error_fields_field/> <error_text_field/> <bad_line_files_destination_directory/> <bad_line_files_extension>warning</bad_line_files_extension> <error_line_files_destination_directory/> <error_line_files_extension>error</error_line_files_extension> <line_number_files_destination_directory/> <line_number_files_extension>line</line_number_files_extension> <date_format_lenient>Y</date_format_lenient> <date_format_locale>en_us</date_format_locale> <GUI> <xloc>99</xloc> <yloc>165</yloc> <draw>Y</draw> </GUI> </step> <step> <name>B</name> <type>FilterRows</type> <description/> <distribute>Y</distribute> <copies>1</copies> <send_true_to>C</send_true_to> <send_false_to>D</send_false_to> <compare> <condition> <negated>N</negated> <leftvalue>input_or_output</leftvalue> <function>=</function> <rightvalue/> <value> <name>constant</name><type>String</type><text>O</text><length>1</length><precision>-1</precision><isnull>N</isnull> </value> </condition> </compare> <GUI> <xloc>198</xloc> <yloc>165</yloc> <draw>Y</draw> </GUI> </step> <step> <name>D</name> <type>Dummy</type> <description>D</description> <distribute>Y</distribute> <copies>1</copies> <GUI> <xloc>274</xloc> <yloc>215</yloc> <draw>Y</draw> </GUI> </step> <step> <name>C</name> <type>Update</type> <description/> <distribute>Y</distribute> <copies>1</copies> <connection>HRDB</connection> <commit>0</commit> <error_ignored>N</error_ignored> <ignore_flag_field/> <lookup> <table>timecard</table> <key> <name>emp_no</name> <field>emp_id</field> <condition>=</condition> <name2/> </key> <key> <name>rec_date</name> <field>tc_date</field> <condition>=</condition> <name2/> </key> <value> <name>time_out</name> <rename>epoch_sec</rename> </value> </lookup> <GUI> <xloc>273</xloc> <yloc>119</yloc> <draw>Y</draw> </GUI> </step> </transformation>

The third transformation reads the HRDB.timecard table, calculates worked hours per employee over the given time period, aggregates the worked hours per employee, applies the employee's per hour rate from the HRDB.employee table to get the dollar value to be paid out, then groups and aggregates the dollar values over department, then inserts the two rows into the MySQL CostingDB table.

<?xml version="1.0" encoding="UTF-8"?> <transformation> <info> <name/> <directory>/</directory> <log> <read/> <write/> <input/> <output/> <update/> <connection/> <table/> <use_batchid>Y</use_batchid> <use_logfield>N</use_logfield> </log> <maxdate> <connection/> <table/> <field/> <offset>0.0</offset> <maxdiff>0.0</maxdiff> </maxdate> <size_rowset>1000</size_rowset> <sleep_time_empty>1</sleep_time_empty> <sleep_time_full>1</sleep_time_full> <dependencies> </dependencies> </info> <notepads> </notepads> <connection> <name>CostingDB</name> <server>localhost</server> <type>MYSQL</type> <access>Native</access> <database>costingdb</database> <port>3306</port> <username>root</username> <password>Encrypted 2be98afc86aa7f2e4cb79ce7dc781bed6</password> <servername/> <data_tablespace/> <index_tablespace/> <attributes> <attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute> <attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute> <attribute><code>PORT_NUMBER</code><attribute>3306</attribute></attribute> </attributes> </connection> <connection> <name>HRDB</name> <server>localhost</server> <type>POSTGRESQL</type> <access>Native</access> <database>hrdb</database> <port>5432</port> <username>hruser</username> <password>Encrypted 2be98afc86aa7f2e4cb79a662cb81fcc8</password> <servername/> <data_tablespace/> <index_tablespace/> <attributes> <attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute> <attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute> <attribute><code>PORT_NUMBER</code><attribute>5432</attribute></attribute> </attributes> </connection> <order> <hop> <from>timecard</from><to>Sort(emp_id)</to><enabled>Y</enabled> </hop> <hop> <from>Sort(emp_id)</from><to>Group(emp_id)</to><enabled>Y</enabled> </hop> <hop> <from>Group(emp_id)</from><to>Lookup(deptId,rate)</to><enabled>Y</enabled> </hop> <hop> <from>Sum(Emp Cost)</from><to>Sort(dept_id)</to><enabled>Y</enabled> </hop> <hop> <from>Sort(dept_id)</from><to>Group(dept_id)</to><enabled>Y</enabled> </hop> <hop> <from>Group(dept_id)</from><to>Insert Input Costs</to><enabled>Y</enabled> </hop> <hop> <from>Remove Extra Cols</from><to>Sum(Emp Cost)</to><enabled>Y</enabled> </hop> <hop> <from>Lookup(deptId,rate)</from><to>Calculate Emp Cost</to><enabled>Y</enabled> </hop> <hop> <from>Calculate Emp Cost</from><to>Remove Extra Cols</to><enabled>Y</enabled> </hop> </order> <step> <name>Group(dept_id)</name> <type>GroupBy</type> <description/> <distribute>Y</distribute> <copies>1</copies> <all_rows>N</all_rows> <ignore_aggregate>N</ignore_aggregate> <field_ignore/> <directory>%%java.io.tmpdir%%</directory> <prefix>grp</prefix> <group> <field> <name>dept_id</name> </field> </group> <fields> <field> <aggregate>input_cost</aggregate> <subject>emp_cost</subject> <type>SUM</type> </field> </fields> <GUI> <xloc>221</xloc> <yloc>394</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Group(emp_id)</name> <type>GroupBy</type> <description/> <distribute>Y</distribute> <copies>1</copies> <all_rows>N</all_rows> <ignore_aggregate>N</ignore_aggregate> <field_ignore/> <directory>%%java.io.tmpdir%%</directory> <prefix>grp</prefix> <group> <field> <name>emp_id</name> </field> </group> <fields> <field> <aggregate>tot_worked_hrs</aggregate> <subject>worked_hrs</subject> <type>SUM</type> </field> </fields> <GUI> <xloc>102</xloc> <yloc>129</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Lookup(deptId,rate)</name> <type>DBLookup</type> <description/> <distribute>Y</distribute> <copies>1</copies> <connection>HRDB</connection> <cache>Y</cache> <cache_size>0</cache_size> <lookup> <table>employee</table> <orderby/> <fail_on_multiple>N</fail_on_multiple> <eat_row_on_failure>Y</eat_row_on_failure> <key> <name>emp_id</name> <field>emp_id</field> <condition>=</condition> <name2/> </key> <value> <name>dept_id</name> <rename>dept_id</rename> <default/> <type>Integer</type> </value> <value> <name>rate_per_hour</name> <rename>rate_per_hour</rename> <default/> <type>Number</type> </value> </lookup> <GUI> <xloc>100</xloc> <yloc>223</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Sort(dept_id)</name> <type>SortRows</type> <description/> <distribute>Y</distribute> <copies>1</copies> <directory>%%java.io.tmpdir%%</directory> <prefix>out</prefix> <sort_size>5000</sort_size> <fields> <field> <name>dept_id</name> <ascending>Y</ascending> </field> </fields> <GUI> <xloc>224</xloc> <yloc>488</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Sort(emp_id)</name> <type>SortRows</type> <description/> <distribute>Y</distribute> <copies>1</copies> <directory>%%java.io.tmpdir%%</directory> <prefix>out</prefix> <sort_size>5000</sort_size> <fields> <field> <name>emp_id</name> <ascending>Y</ascending> </field> </fields> <GUI> <xloc>104</xloc> <yloc>41</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Sum(Emp Cost)</name> <type>TextFileOutput</type> <description/> <distribute>Y</distribute> <copies>1</copies> <separator>;</separator> <enclosure>"</enclosure> <enclosure_forced>N</enclosure_forced> <header>Y</header> <footer>N</footer> <format>Unix</format> <encoding/> <file> <name>aggregated_emp_cost</name> <extention>txt</extention> <append>N</append> <split>Y</split> <add_date>Y</add_date> <add_time>Y</add_time> <zipped>N</zipped> <pad>N</pad> <splitevery>0</splitevery> </file> <fields> <field> <name>dept_id</name> <type>5</type> <format/> <currency/> <decimal/> <group/> <nullif/> <length>9</length> <precision>0</precision> </field> <field> <name>emp_cost</name> <type>1</type> <format> 0000000000.00</format> <currency/> <decimal/> <group/> <nullif/> <length>12</length> <precision>2</precision> </field> </fields> <GUI> <xloc>99</xloc> <yloc>489</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Insert Input Costs</name> <type>TableOutput</type> <description/> <distribute>Y</distribute> <copies>1</copies> <connection>CostingDB</connection> <table>input_cost</table> <commit>100</commit> <truncate>Y</truncate> <ignore_errors>N</ignore_errors> <use_batch>Y</use_batch> <partitioning_enabled>N</partitioning_enabled> <partitioning_field/> <partitioning_daily>N</partitioning_daily> <partitioning_monthly>Y</partitioning_monthly> <tablename_in_field>N</tablename_in_field> <tablename_field/> <tablename_in_table>Y</tablename_in_table> <return_keys>N</return_keys> <return_field/> <GUI> <xloc>341</xloc> <yloc>394</yloc> <draw>Y</draw> </GUI> </step> <step> <name>timecard</name> <type>TableInput</type> <description/> <distribute>Y</distribute> <copies>1</copies> <connection>HRDB</connection> <sql>SELECT emp_id, (((time_out - time_in) / 3600) + 1) AS worked_hrs FROM timecard</sql> <limit>0</limit> <lookup/> <execute_each_row>N</execute_each_row> <variables_active>N</variables_active> <GUI> <xloc>26</xloc> <yloc>41</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Remove Extra Cols</name> <type>SelectValues</type> <description/> <distribute>Y</distribute> <copies>1</copies> <fields> <remove> <name>emp_id</name> </remove> <remove> <name>tot_worked_hrs</name> </remove> <remove> <name>rate_per_hour</name> </remove> <meta> <name>dept_id</name> <rename/> <type>-</type> <length>9</length> <precision>0</precision> </meta> <meta> <name>emp_cost</name> <rename/> <type>-</type> <length>12</length> <precision>2</precision> </meta> </fields> <GUI> <xloc>100</xloc> <yloc>400</yloc> <draw>Y</draw> </GUI> </step> <step> <name>Calculate Emp Cost</name> <type>ScriptValue</type> <description/> <distribute>Y</distribute> <copies>1</copies> <script>emp_cost = tot_worked_hrs.getNumber() * rate_per_hour.getNumber(); </script> <fields> <field> <name>emp_cost</name> <rename>emp_cost</rename> <type>Number</type> <length>16</length> <precision>2</precision> </field> </fields> <GUI> <xloc>97</xloc> <yloc>312</yloc> <draw>Y</draw> </GUI> </step> </transformation>

You can run the transformations individually through Spoon using the "Run" icon. Alternatively, you can run them through the Pan tool. Here is a Pan script that runs the entire transformation:

1 2 3 4 5 6 7

#!/bin/bash KETTLE_HOME=/path/to/your/kettle/installation/here cd $KETTLE_HOME ./pan.sh -file=usr/costing/extract_in_times.ktr ./pan.sh -file=usr/costing/extract_out_times.ktr ./pan.sh -file=usr/costing/aggregate_worked_hrs.ktr cd -

Alternatively, you could use Chef GUI Tool to build up this job graphically. Chef offers some other features such as modules which do FTP, send email and so on. The job is shown graphically below, along with the generated .kjb file.

<?xml version="1.0" encoding="UTF-8"?> <job> <name/> <directory>/</directory> <logconnection/> <logtable/> <use_batchid>N</use_batchid> <pass_batchid>N</pass_batchid> <use_logfield>N</use_logfield> <entries> <entry> <name>START</name> <description>A job starts to process here.</description> <type>SPECIAL</type> <start>Y</start> <dummy>N</dummy> <repeat>N</repeat> <schedulerType>0</schedulerType> <interval>60</interval> <hour>12</hour> <minutes>0</minutes> <weekDay>1</weekDay> <DayOfMonth>1</DayOfMonth> <parallel>N</parallel> <draw>Y</draw> <nr>0</nr> <xloc>39</xloc> <yloc>28</yloc> </entry> <entry> <name>DUMMY</name> <description>A dummy entry.</description> <type>SPECIAL</type> <start>N</start> <dummy>Y</dummy> <repeat>N</repeat> <schedulerType>0</schedulerType> <interval>60</interval> <hour>12</hour> <minutes>0</minutes> <weekDay>1</weekDay> <DayOfMonth>1</DayOfMonth> <parallel>N</parallel> <draw>N</draw> <nr>0</nr> <xloc>50</xloc> <yloc>50</yloc> </entry> <entry> <name>OK</name> <description>This comparisson is true when no errors have occured.</description> <type>EVAL</type> <script>errors == 0</script> <parallel>N</parallel> <draw>N</draw> <nr>0</nr> <xloc>0</xloc> <yloc>0</yloc> </entry> <entry> <name>ERROR</name> <description>This comparisson is true when one or more errors have occured.</description> <type>EVAL</type> <script>errors != 0</script> <parallel>N</parallel> <draw>N</draw> <nr>0</nr> <xloc>0</xloc> <yloc>0</yloc> </entry> <entry> <name>Timecard Insert (IN)</name> <description>Timecard Insert</description> <type>TRANS</type> <filename>/home/sujit/tmp/Kettle-2.3.0/usr/costing/extract_in_times.ktr</filename> <transname>Insert Timecard data</transname> <directory>/</directory> <arg_from_previous>N</arg_from_previous> <exec_per_row>Y</exec_per_row> <clear_rows>Y</clear_rows> <clear_files>Y</clear_files> <set_logfile>N</set_logfile> <logfile/> <logext/> <add_date>N</add_date> <add_time>N</add_time> <loglevel>Error</loglevel> <parallel>N</parallel> <draw>Y</draw> <nr>0</nr> <xloc>195</xloc> <yloc>28</yloc> </entry> <entry> <name>Timecard Update (OUT)</name> <description/> <type>TRANS</type> <filename>/home/sujit/tmp/Kettle-2.3.0/usr/costing/extract_out_times.ktr</filename> <transname>Timecard Insert (Update OUT)</transname> <directory>/</directory> <arg_from_previous>N</arg_from_previous> <exec_per_row>N</exec_per_row> <clear_rows>Y</clear_rows> <clear_files>Y</clear_files> <set_logfile>N</set_logfile> <logfile/> <logext/> <add_date>N</add_date> <add_time>N</add_time> <loglevel>Error</loglevel> <parallel>N</parallel> <draw>Y</draw> <nr>0</nr> <xloc>351</xloc> <yloc>28</yloc> </entry> <entry> <name>Aggregate Costing Info</name> <description/> <type>TRANS</type> <filename>/home/sujit/tmp/Kettle-2.3.0/usr/costing/aggregate_worked_hrs.ktr</filename> <transname>Aggregate Costing Info</transname> <directory>/</directory> <arg_from_previous>N</arg_from_previous> <exec_per_row>N</exec_per_row> <clear_rows>Y</clear_rows> <clear_files>Y</clear_files> <set_logfile>N</set_logfile> <logfile/> <logext/> <add_date>N</add_date> <add_time>N</add_time> <loglevel>Error</loglevel> <parallel>N</parallel> <draw>Y</draw> <nr>0</nr> <xloc>507</xloc> <yloc>28</yloc> </entry> </entries> <hops> <hop> <from>START</from> <to>Timecard Insert (IN)</to> <from_nr>0</from_nr> <to_nr>0</to_nr> <enabled>Y</enabled> <evaluation>Y</evaluation> <unconditional>Y</unconditional> </hop> <hop> <from>Timecard Insert (IN)</from> <to>Timecard Update (OUT)</to> <from_nr>0</from_nr> <to_nr>0</to_nr> <enabled>Y</enabled> <evaluation>Y</evaluation> <unconditional>N</unconditional> </hop> <hop> <from>Timecard Update (OUT)</from> <to>Aggregate Costing Info</to> <from_nr>0</from_nr> <to_nr>0</to_nr> <enabled>Y</enabled> <evaluation>Y</evaluation> <unconditional>N</unconditional> </hop> </hops> <notepads> </notepads> </job>

Finally, you can run more than one job, schedule them and so on using Kitchen. Frankly, I dont see much reason to use Chef and Kitchen, since you can just write Pan scripts and schedule them via cron, but I guess the Kettle team put them in there for completeness.

My conclusion is that Spoon is fairly powerful and provides very powerful plumbing to design and run ETL jobs. I still dont like the fact that the only programming interface is a GUI, but I dont have any concrete suggestions for a scripting interface. For those whose needs are not met by the standard components provided by Spoon, there is the Injector component which can be backed by user-supplied Java code, so Kettle also provides a hook for extensibility.