How to Build an ETL Pipeline for ElasticSearch Using Segment and Iron.io

The following tutorial assumes that you've installed Segment into your website. From here we'll walk through switching on the IronMQ integration and then running IronWorker to transform the data before loading it into ElasticSearch.

Copy your Iron.io Credentials from the Dashboard

1. Connecting IronMQ to Segment

With Segment and Iron.io, building a reliable ETL pipeline into ElasticSearch is simple. The initial extract process, often the origin of many headaches, is already handled for you by piping the data from Segment to IronMQ.

Flipping on the IronMQ integration within the Segment dashboard automatically sends all of your Segment data to a queue named "segment". All you need to do to initiate the process is create a project within the Iron.io HUD and enter the credentials within Segment.

Enter Your Iron.io Credentials into Segment

2. Transforming the Data

Now that we have our Segment data automatically sending to IronMQ, it's time to transform it prior to loading it into ElasticSearch.

Let's say we want to filter out identified users based on their plan type so only paid user data gets sent to ElasticSearch. In the case of building customer dashboards, this allows us to maintain a collection of purely usable data, making our indexing and searching more efficient for the eventual end use case.

We're going to create a worker to pull messages from the queue in batches, filter the data, and then load into an ElasticSearch instance. For simplicity's sake, we're going to create a Heroku app with the Bonsai add-on, a hosted ElasticSearch service. Leveraging IronWorker's scheduling capabilities, we can check for messages on the queue at regular intervals.

With IronMQ and IronWorker, we can also ensure that we're not losing any data in this process and that we're not overloading our ElasticSearch cluster with too much incoming data. Buffering and, buffering and...

Segment Data
Before we get to our worker, let's examine the data from the Segment that gets sent to the queue. The segment is vendor agnostic, making it very simple to interact with the exported data. The Tracking API that we'll be working with consists of several methods: identify track, page, screen, group, and alias. You can dive into the docs here. We use the Ruby client within the Iron.io website to see where our users are going. Any page we want to track, we just place this line on the Controller.

Analytics.track( user_id: cookies[:identity], event: "Viewed Jobs Page" )

Here is a typical basic identify record in JSON format... parsed, filtered, and prettied. This is plenty enough for us to make our transformation before loading into ElasticSearch.

{
 “action”  : “Identify”,
 “user_id” : “123”,
 “traits”  : {
  “email”        : “ivan@iron.io”,
  “name”         : “Ivan Dwyer”,
  “account_type” : “paid”
  “plan”         : “Enterprise”,
 },
 “timestamp” : “2014-09-10-02T00:30:00.276Z”
}

Worker Setup
Now let's look at creating the task that performs the business logic for the transformation. With IronWorker, you create task-specific code within a "worker" and upload it to our environment for highly scalable background processing. Note that this example uses Ruby, but IronWorker has support for almost every common language including PHP, Python, Java, Node.js, Go, .NET, and more.

For this worker, the code dependencies will be the irommq, elasticsearch, and JSON gems. It’s a quick step to create the .worker file that contains the worker config and we'll put the Bonsai credentials in a config.yml file.

Our worker in this example will be very simple as a reference point. With the flexible capabilities of IronWorker, you are free to build out any transformations you'd like and load them to any external service you prefer. The Segment queue can grow rapidly (because of traffic), so we'll want to schedule this worker to run every hour and clear the queue each time. If your queue growth gets out of hand, you can create a master worker that splits up the processing to slave workers that can run concurrently, significantly cutting down the total processing time. Just another reason keeping the ETL process within Iron.io is the way to go.

Once we initiate both the IronMQ and ElasticSearch clients, we can make our connections and start working through the queue data before loading to Bonsai. We know from our Segment integration that the queue is named "segment", and that our paid users are tagged with an "account_type" trait. This allows us to easily loop through all the messages and check whether or not each record meets our requirements. Post the data to Bonsai and then delete the message from the queue. Pretty simple.

3. Upload and Schedule our Worker

Now we can use the IronWorker CLI tool to upload our code package.

$ iron_worker upload segment

Add a New Scheduled Task

Once the code package is uploaded and built, we can go to the HUD and schedule it to run. Select the "SegmentWorker" code package and set the timing. Every hour at regular priority seems fine for this task as this is a background job with no time sensitivity. What's that mem1 cluster you ask? Why that’s our new high memory worker environment meant for heavy processing needs.

Now our worker is scheduled to run every hour with the first one being queued up right away. We can watch its progress in the Tasks view.

View Task Status in Realtime

Once it's complete, we can check the task log to see our output. Looks like our posting to ElasticSearch was successful.

Detailed log output

Just to be sure, let's check our Bonsai add-on in Heroku. Looks like we successfully created an index and populated it with documents. Now we can do what we like within ElasticSearch.

Bonsai add-on within Heroku

There you have it. With the Iron.io integration on Segment, you can build your own ETL pipeline any way you'd like.