At zeotap, a large amount of structured and unstructured data from data partners is transformed and converted to easily queryable, standardized data sets, which are fanned out to a variety of destinations. For this processing, the transformation jobs use Apache Spark as the distributed computing framework, with a fair share of them being batch processing jobs. Batch processing is defined as a non-continuous processing of data. A lot of the batch processing done at zeotap regularly processes files at a given frequency, with the input data being gathered in between the processing jobs.

We currently use two different types of systems to launch and manage these Spark jobs.

An introduction to Oozie, and how we use it has been given here . In this article, we will be focussing on Livy, and how we use it at zeotap.

Oozie vs Livy

Oozie is a workflow management system, which allows for launching and scheduling various MapReduce jobs. A workflow is defined as a Directed Acyclic Graph (DAG) of actions (which can include Spark Batch Jobs), which is defined in an XML File.

Livy, on the other hand, is a REST interface with a Spark Cluster, which allows for launching, and tracking of individual Spark Jobs, by directly using snippets of Spark code or precompiled jars.

Both these systems can be used to launch and manage Spark Jobs, but go about them in very different manners. Where can we use Oozie, and where can we use Livy?

Oozie

Oozie can be used when the processing flow involves multiple steps, each of which is dependent on the previous one. This is why a DAG is required to launch Oozie Workflows. A simple example of such a dependency would be of running a script to check whether the expected input files are present or not before the actual processing takes place on those input files.

When oozie launches a spark job, it first launches an ‘oozie-launcher’ container on a core node of the cluster, which in turn launches the actual Spark Job. Oozie uses this oozie-launcher container to track and wait for Spark job processing.

Pros:

• Workflow Management — Oozie supports coordinator and workflow management. It is a more generic framework, as compared to Livy (which is just used for launching and managing Spark Jobs specifically), supporting running and scheduling of workflows.
• Versatile — Oozie supports much more than just Spark jobs, and is used for managing a wide variety of Hadoop jobs, such as Pig, Hive, Sqoop, Distcp to name a few more. This gives Oozie a large variety of use cases.

Cons:

• Cluster Choke — Oozie launches an oozie-launcher container (Fig. 2), for each job it launches. When multiple jobs are launched concurrently, there is a possibility that all the oozie-launcher are running on the cluster, but none of the Spark Jobs. Now those oozie-launcher containers are waiting for their respective Spark jobs to complete, but none of the actual Spark jobs have been launched. This means the cluster is choked with oozie-launchers.
• Jar Hell — A pro for Oozie, being its versatility in launching multiple types of Hadoop jobs, also turns out to be a reason for a con. Each type of job requires its own jars and version of libraries. This can cause library jar and classpath issues, where different incompatible versions of a library can cause failures in jobs, if not properly set up.
• The above point can also cause migration between/upgrading to newer Spark versions to become a more involved process, as jar version conflicts are bound to crop up when upgrading versions, which are often difficult to resolve.
• Workflow Maintenance — Need to write workflows to launch even the simplest of jobs. (Fig. 1)

Livy

Livy can be used to launch ‘asynchronous’ Spark Jobs which are not dependent on, and also do not have any jobs dependent on them. By ‘asynchronous’, we mean that these jobs can be launched at any time, and we do not need to wait for the response of the job.

Since jobs launched through Livy are asynchronous, they can just be launched by a parent processing workflow, and then can continue on its own without waiting for the response about this job. Such jobs would often produce results that are further also asynchronously consumed. Some such examples are:

1. Data quality reporting jobs, involving stats such as row counts, column cardinality, etc. would write to a database, which is then further consumed for dashboard generation.
2. Terminal jobs of a pipeline, such as jobs to prepare and export data to an external system.

But, this does not mean that Livy’s features cannot be used to Orchestrate workflows. We could possibly wait for a job’s response, by simply polling the API. Indeed using this, at zeotap, we have leveraged Livy to launch and monitor spark jobs, and built one part of a decision making, and orchestrating framework, which supports workflow management, similar to Oozie.

Pros:

• Simple HTTP interface to launch, interact and track Spark Jobs.
• Unlikely to cause any jar issues, as it is pre-bundled with the requisite jars, and just having Spark set up correctly on the cluster is enough.
• Easy migration between Spark Versions. If the job runs on the newer Spark Version, Livy will launch it.
• Do not need to write additional workflow files, and properties to launch jobs.
• Will not choke clusters with additional tracking jobs (like ‘oozie-launcher’ — Fig. 2, 3).

Cons:

• Less versatile than Oozie for launching jobs supports only Spark jobs.
• Not a workflow management system. Purely used to launch and track single Spark Jobs. But, a workflow management system can be built around this.

Livy Features

Spark allows launching two types of data analysis options:

1. Interactive: Using spark-shell and pyspark. This launch mode allows for commands to be submitted at runtime.
2. Batch: Using spark-submit to launch an application on a cluster, with pre-compiled code, without any interaction at run time.

Livy supports these two modes using the REST interface as well.

Interactive / Session

In Interactive Mode (or Session mode as Livy calls it), first, a Session needs to be started, using a POST call to the Livy Server. This will start an Interactive Shell on the cluster for you, similar to if you logged into the cluster yourself and started a spark-shell.

Batch

In batch mode, the code to be executed needs to be already pre-written in a file; This could be a python file or compiled jar, which can then be stored on the cluster and used to launch the job.

Create Session:

curl -X POST \http://<Livy Endpoint>/sessions \
-H‘Cache-Control:no-cache’\
-H ‘Content-Type: application/json’ \
-d ‘{
“driverMemory” : “1g”,
“executorMemory” : “1g”,
“numExecutors” : 2,
“name” : “livy”,
“conf” : {
“spark.test.conf” : “conf1”
}
}’

Response:

{
“id”: 0,
“appId”: null,
“owner”: null,
“proxyUser”: null,
“state”: “starting”,
“kind”: “shared”,
“appInfo”: {
“driverLogUrl”: null,
“sparkUiUrl”: null
},
“log”: [
“stdout: “,
“\nstderr: “,
“\nYARN Diagnostics: “
]
}

In both Session and Batch mode, on launching the Spark job, Livy will respond with an Integer ID which can be used for tracking the state of the Session/Batch.

Check Session State:

curl -X GET \
http://<Livy Endpoint>/sessions/1/state

Response:

{
“id”: 1,
“state”: “idle”
}

Setting Up Livy

Setting up Livy on a Hadoop Cluster is fairly straightforward. The prerequisites to run Livy are that the SPARK_HOME and HADOOP_CONF_DIR environment variables need to be set up on the Master Node.

If these are set, then out of the box, after extracting the Livy package to the installation folder, we can run it directly, using :

./bin/livy-server start

Configuration

By default, Livy uses configuration files present in the conf directory in the install directory. This can be changed by setting LIVY_CONF_DIR environment variables before running.

The configurations mentioned below need to be added to a file livy.conf in the configuration directory.

livy.spark.master: What spark master Livy sessions should use.

• By default, it is set to local. Set it to yarn to launch using Hadoop YARN Resource Manager

livy.spark.deploy-mode : What spark deploy mode Livy sessions should use.

• It is strongly recommended to use cluster deploy mode when using Livy with YARN. This ensures that the host running the Livy server doesn’t become overloaded when multiple sessions/jobs are running.

livy.server.yarn.app-lookup-timeout : How long Livy will wait and try to launch a session/job. If the cluster is busy, Livy might not be able to launch the job within this time limit. Increase this for a larger tolerance of a busy cluster.

livy.server.recovery.mode: Recovery mode of Livy. Possible values:

• off: Default. Turn off recovery. Every time Livy shuts down, it stops and forgets all sessions.
• recovery: Livy persists session info to the state store. When Livy restarts, it recovers previous sessions from the state store.
• It is strongly recommended to set this to recovery. By default, it is in off mode. If set to off, if and whenever Livy shuts down / crashes, all sessions and jobs launched by Livy are forgotten. Furthermore, if any session is currently running, Livy will shut it down as well.

If set to recovery, ensure that the following two configs are also setup:

livy.server.recovery.state-store : Where Livy should store the state for recovery. Possible values:

• <empty>: Default. State store disabled.
• filesystem: Store state on a file system.
• zookeeper: Store state in a Zookeeper instance.

livy.server.recovery.state-store.url: For filesystem state store, the path of the state store directory. e.g. file:///tmp/livy or hdfs:///.

• For zookeeper, the address to the Zookeeper servers. e.g. host1:port1,host2:port2

At zeotap, we use filesystem state store on the local system of the master. Livy does not go down very often, and even if it does, storing the state on the local filesystem proves to be reliable enough for the use cases.

The above configurations should be enough, to allow for a stable Livy usage environment. The following configurations are required only if querying Livy and updating of the state of Spark Jobs using the API is required.

livy.server.session.state-retain.sec: How long a finished state of a session should be retained by Livy to be available for querying. By default, it is 10 minutes. Applies to both batch and sessions.

livy.server.session.timeout: This is only for sessions (and not batches launched) on Livy. This is the time on how long Livy will wait before timing out an idle session. By default, 1 hour.

Note: If you are planning to use Apache Livy on newer versions of Amazon EMR, it is already bundled, and only configurations need to be set up.

Job Tracking using Livy — Caveats

A few things to note about querying Apache Livy to track job states :

• The UI currently will show only 100 oldest jobs/sessions within the state-retain time out. The other jobs will not be visible on the UI but can be queried through the REST API.
• Livy does job tracking through integer job IDs. This could lead to mix-ups if Livy shuts down, and restarts without recovery being set, as the job IDs reset to 0 again.
• Livy logs which it returns from its APIs are usually not useful for debugging. Very limited information is available, usually just the tail end of the logs from the job. Debugging will often involve looking into the yarn application logs instead.

Apache Livy and Oozie are just two of the constituent technologies used to build the data platform at zeotap. Leveraging these two, and others, we have built a code-driven workflow management system used across multiple products here. A blog about this control plane framework will follow soon!

AUTHOR: Akshay Sasikumar — Data Engineer at zeotap

Originally published at https://zeotap.com on November 28, 2019.

Writing tests are analogous to tasting your meal before you serve it. The importance of unit testing is known at all levels in programming but more often is ignored when it comes to UI developers. This post is a brief about how you can start your journey in being a better Frontend Engineer by incorporating these key concepts about unit testing in your code.

Overview

1.Importance of Unit testing

2.Sample App

• Setup
• Writing Unit Tests for Presentational components
• Writing Unit Tests for Services
• Writing Unit Tests for Container Components

3.Conclusion

Importance of Unit testing

Writing Unit Tests does seem like an overhead when a you can just test the functionality by using it. For the times you are in such a dilemma you can keep these few points in mind:

1. Unit tests not only improve quality but decrease time to debug : Unit tests help you understand what parts of the app are working as intended and what parts aren’t and hence allow you to narrow down on the cause of bugs much quicker than using console.logs or debuggers.
2. We are JS developers!! : We all as developers have either built test UI components and rough Html to test an underlying logic/service or delayed testing till our Presentational components are done. Writing a unit test allows you to iteratively build a functional component without unnecessary test UI elements.
3. Freedom to collaborate : Working in a team I have often noticed members working on isolated silos of functionality and with a large code base there is a never ending fear of breaking some working code during refactoring and bug fixing. This should and can be avoided if you write proper unit tests along with the code which detects any breakage in case of changes for developers who may work on the code later.
4. No Low level Documentation a unit test declares the purpose of a given unit of code. This reduces the requirement for a developer to explicitly document code ( would also recommend declarative style of programming for all JS developers) and product teams can focus more on the look and feel of the application than on the functionality.
   Using Test frameworks like Jest also allows you to test Frontend code on your CI/CD environments which is a plus to the point no. 3 as it helps generate regular reports on your code health and test coverage.

Here are some key guidelines that you should keep in mind while writing unit tests :

1. Understanding the type of unit tests that should be written depends on the type of app component( Presentational, Logic Containers,Services,etc). Understanding what should be tested really helps in reasoning the extra effort you are taking in writing unit tests at each level.
2. Write Functional JS and try to break down your app into Presentational and Logic components as much as possible. This really helps in improving the focus of your unit tests and also decreases the time taken to write them.
3. Write Tests along with the code. This is By far the most Important one !! I can’t stress enough on how painful it has been for me to revisit old code and add unit tests for already developed components. It requires both time and effort to figure out what you have written and what to test. When Tests are written our aim should be to write code that pass tests rather than the other way around.
4. Practice writing tests before you dive into writing your app. Most developers avoid writing tests because they either don’t know or are not completely sure about some basics like Mocking a Class , testing an async call, mocking http calls etc. Get rid of these confusions and myths with practice. So practice unit testing as much as you practice writing application code.

Having Understood the importance of writing tests we are going to go through an example Angular App and write some unit tests around it with Jest.

Why Jest ?

Jest is beautiful testing framework that provides a uniform and non-browser based unit testing options for multiple javascript frameworks.

Find more about them here.

Also a shoutout to jest-angular-preset library that makes it easy to use jest with angular. With jest I get three great features which aren’t present with default angular testing setup : Snapshot testing, Unit tests that can run without browser and AutoMocking. I suggest everyone to understand these to use this wonderful framework to its fullest.

Setup :

If you have never used angular before please follow the official angular setup guide here

Our App will have three major Component : AppComponent, ListingService, ListRowComponent. But before we start writing our components and test cases, We have to setup jest.

Steps to setup jest :

Use this quick guide to do the initial setup, remove karma based code, and run jest.

Jest allows you to store your config in either a jest field in your package.json or in a separate file jest.config.js

I would suggest everyone should go through the official config guide once to know what kind of configurations your project can have and might need. To help you guys I would recommend at-least focusing on the following fields: setupFilesAfterEnv, coverageDirectory, coverageReporters, transformIgnorePatterns, modulePathIgnorePatterns, moduleNameMapper, testPathIgnorePatterns

Here is jest.config.js from our sample app

module.exports = {
    "preset": "jest-preset-angular",
    "setupFilesAfterEnv": ["<rootDir>/setupJest.ts"],
    globals: {
      "ts-jest": {
        tsConfig: '<rootDir>/tsconfig.spec.json',
        "diagnostics":false,
        "allowSyntheticDefaultImports": true,
        "stringifyContentPathRegex": "\\.html$",
        astTransformers: [require.resolve('jest-preset-angular/InlineHtmlStripStylesTransformer')],
      }
    },
    coverageDirectory:'<rootDir>/output/coverage/jest',
    transformIgnorePatterns: ["node_modules/"],
    "coverageReporters": [
      "text",
      "json",
    ],
    "reporters": [
      "default",
    ],
    snapshotSerializers: [
      'jest-preset-angular/AngularSnapshotSerializer.js',
      "jest-preset-angular/AngularSnapshotSerializer.js",
      "jest-preset-angular/HTMLCommentSerializer.js"
    ],
    "transform": {
      '^.+\\.(ts|html)$': 'ts-jest',
      "^.+\\.js$": "babel-jest",
    },
    modulePathIgnorePatterns: [],
    moduleNameMapper: {},
    testPathIgnorePatterns:['sampleCodes/'],
  };

Here is my tsconfig.spec.ts

{
  "extends": "./tsconfig.json",
  "compilerOptions": {
    "outDir": "./out-tsc/spec",
    "types": ["jest", "node"],
    "emitDecoratorMetadata": true,
    "allowJs": true
  },
  "files": [
    "src/polyfills.ts"
  ],
  "include": [
    "src/**/*.spec.ts",
    "src/**/*.d.ts"
  ]
}

Note: Don’t simply copy and paste the code but understanding the config really helps you in setting up your entire config for your project on your own.

I would also suggest install jest globally

npm install -g jest

This really helps when running jest cli commands required for snapshot testing ( like updating snapshots using jest -u)

Finally run jest and check if the basic tests that are auto created with ng generate are running using

jest --coverage

Here is a great guide on how to test components and improve our test cases and how the DOM Testing library helps in this

Writing Unit Tests for Presentational components

If you are in practice of writing Pure Presentational components then you are awesome!!. If you aren’t I suggest you start practicing on how to divide your app code into Logic Containers and Presentational Components.

Jest has the ability to use Snapshot testing for testing the UI components. Read more about Snapshot testing here

This saves time spent writing DOM queries. As per the documentation one should commit these snapshots with your code so you can verify how your UI components should be rendered in DOM.

When Not to use snapshots ?

If the component is basic and simple enough ,snapshot testing should cover most of your UI tests, though avoid using it with Presentational Components like Lists where you would want to check the total number of rendered rows or in components where verification of business logic representation is required.

Below Find Sample ListRowComponent

@Component({
  selector: 'app-list-row-component',
  templateUrl: './list-row-component.component.html',
  styleUrls: ['./list-row-component.component.scss'],

})
export class ListRowComponentComponent implements OnInit {

  @Input() firstName:string;
  @Input() lastName:string;
  @Input() gender:string;
  @Output() rowClick = new EventEmitter();

  getClass(){
    return {
      'blue':this.gender==='male',
      'green':this.gender==='female'
    }
  }
  constructor() { 
  }
  ngOnInit() {
  }
}

Below Find Sample ListRowComponent.spec file

describe('ListRowComponentComponent', () => {
  let component: ListRowComponentComponent;
  let fixture: ComponentFixture<ListRowComponentComponent>;

  beforeEach(async(() => {
    TestBed.configureTestingModule({
      declarations: [ ListRowComponentComponent ]
    })
    .compileComponents();
  }));

  beforeEach(() => {
    fixture = TestBed.createComponent(ListRowComponentComponent);
    component = fixture.componentInstance;
    fixture.detectChanges();
  });

  it('should create', () => {
    expect(component).toBeTruthy();
  });

  it('should render the component with blue color class',()=>{
    component.firstName = 'James'
    component.lastName = 'Bond'
    component.gender = 'male'
    fixture.detectChanges()

    expect(fixture).toMatchSnapshot();
  })
  it('should render the component with green color class',()=>{
    component.firstName = 'James'
    component.lastName = 'Bond'
    component.gender = 'female'
    fixture.detectChanges()

    expect(fixture).toMatchSnapshot();
  })

  it('should emit events onClick',done=>{
    let buttonClicked = false
    component.rowClick.subscribe(()=>{
      buttonClicked =true;
      expect(buttonClicked).toBeTruthy()
      done();
    })
    var btn = getByTestId(fixture.nativeElement,'row-click');
    simulateClick(btn);
  })
});

Note: If you notice I’m using data-testid to query the button in the unit test above. I would suggest all developers to bring this into practice , It makes our tests very resilient to change and robust in nature.

Writing Unit Tests for Services

First here are some concepts that were confusing to me before I stared writing unit tests for services or containers :

Mocking Dependencies. There are a lot of great tutorials available with a simple Google search on this but most use component constructors or promote using auto-mocking features of Jest for Mocking dependencies. It depends on your preference which method you use ,For me mocking dependencies while using Angular’s Dependency Injection to instantiate a component were key and I found a really good way to do that .

You can go through this great article about the same

Mocking Store : It is suggested that we should write getters and selectors for ngrx store ( https://ngrx.io/ ) in services , so that your components are reusable along with the store. This means mocking a Store in service becomes very important.

describe('Service:StoreService', () => {
  let backend: HttpTestingController;

  beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [HttpClientModule, HttpClientTestingModule, RouterTestingModule],
      providers: [
        provideMockStore({ initialState }),
      ],
      schemas:[NO_ERRORS_SCHEMA]
    });
    backend = TestBed.get(HttpTestingController);
  });

  afterEach(inject(
    [HttpTestingController],
    (_backend: HttpTestingController) => {
      _backend.verify();
    }
  ));

know more

Using Marble testing : Lastly Most Services you will create in your angular projects will use RxJs. To test your services and logic container components properly, understanding how to test these Observables ( best done using jasmine-marbles ) is essential.

Here is a great article by Micheal Hoffman that will help you get a good understanding about the same

Sample Service

@Injectable({
  providedIn: 'root'
})
export class ListingService {

  constructor(
    public http: HttpClient
  ) { }

  public getHeaderWithoutToken() {
    return new HttpHeaders()
      .append('Content-Type', 'application/json')
      .append('Accept', 'application/json');
  }

  public getHeader(tokenPrefix = '') {
    let headers = this.getHeaderWithoutToken();
    return { headers };
  }

  public doGet(url,header=this.getHeader()){
    return this.http.get(url,header);
  }
  public getList() : Observable<UserModel[]>{
    return this.doGet('http://example.com/users')
    .pipe(
      map((res:any[])=>{
        return res.map(toUserModel)
    }))
  }
}

Testing a service using jest

describe('ListingServiceService', () => {
  let service: ListingService;
  let backend: HttpTestingController;

  beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [HttpClientModule, HttpClientTestingModule],
      providers: [
        ListingService
      ],
      schemas:[NO_ERRORS_SCHEMA,CUSTOM_ELEMENTS_SCHEMA]
    });
    backend = TestBed.get(HttpTestingController);
    service = TestBed.get(ListingService);
  });

  afterEach(inject(
    [HttpTestingController],
    (_backend: HttpTestingController) => {
      _backend.verify();
    }
  ));

  it('should be created', () => {
    expect(service).toBeTruthy();
  });

  const url = 'http://example.com/users';
  test('should fetch a list of users',done=>{
    service.getList()
    .subscribe(data=>{
      expect(data).toEqual(outputArray)
      done()
    })
    backend.expectOne((req: HttpRequest<any>) => {
        return req.url === url && req.method === 'GET';
      }, `GET all list data from ${url}`)
      .flush(outputArray);
  })
});

Writing Unit Tests for Container Components

Container Components are complex components and often this complexity can lead to confusion towards how to write unit tests for a container component. To avoid this you can take the shallow and deep testing approach of writing unit tests.

You can learn more about this approach here

Sample App container component

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})

export class AppComponent implements OnInit{
  title = 'my-test-app';
  list$ : Observable<UserModel[]>;
  constructor(
    private listService :ListingService,
  ){
  }
  ngOnInit(){
    this.initListService()
  }
  initListService(){
    this.list$ =  this.listService.getList();
  }
  onClicked(user){

  }
}

Setting up the Container for unit tests

let fixture : ComponentFixture<AppComponent>;
  let service : ListingService;
  let component : AppComponent;
  beforeEach(async(() => {
    TestBed.configureTestingModule({
      declarations: [
        AppComponent
      ],
      providers:[
        {provide:ListingService,useClass:MockListService}
      ],
      schemas: [CUSTOM_ELEMENTS_SCHEMA, NO_ERRORS_SCHEMA]
    }).compileComponents();
  }));
  beforeEach(()=>{
    fixture = TestBed.createComponent(AppComponent)
    component = fixture.debugElement.componentInstance;
    service = fixture.componentRef.injector.get(ListingService);
    fixture.detectChanges()
  })

Writing Shallow Tests

Unit tests for testing only parts which are isolated from other components in the current container like whether all DOM components written as part of this component’s template are being rendered as desired, component is being set up by fetching data from services and the component outputs are working as intended.

it('should create the app', () => {

    expect(component).toBeTruthy();
  });

  it('should render title in a h1 tag',() => {
    const compiled = fixture.debugElement.nativeElement;
    expect(queryByTestId(compiled,'app-title')).not.toBeNull();
    expect(queryByTestId(compiled,'app-title').textContent).toEqual(component.title)
  });

  test('should fetch the user list from the listing service',async(()=>{
    const spy = jest.spyOn(service,'getList');
    var expectedObservable = cold('-a',{a:outputArray})
    spy.mockReturnValue(expectedObservable)
    component.ngOnInit()
    fixture.detectChanges()
    expect(spy).toHaveBeenCalled();
    expect(component.list$).toBeObservable(expectedObservable)
    getTestScheduler().flush()
    fixture.detectChanges()
    component.list$.subscribe((o)=>{
      fixture.detectChanges()
      var list = fixture.nativeElement.querySelectorAll('app-list-row-component')
      expect(list.length).toEqual(outputArray.length)
    })
    spy.mockRestore()
  }))

Writing Deep Tests

Set of unit tests where the aim is to check the interaction in the component between the child / internal components and the providers and dispatchers attached to the component.

test('should call onClicked when app-list-row-component is clicked',()=>{
    const spy = jest.spyOn(service,'getList');
    var expectedObservable = cold('a',{a:outputArray})
    spy.mockReturnValue(expectedObservable)
    component.initListService()
    getTestScheduler().flush()
    var onClicked = spyOn(component,'onClicked').and.callThrough();
    component.list$.subscribe((o)=>{
      fixture.detectChanges()
      var row0 = fixture.debugElement.query((el)=>{
        return el.properties['data-testid'] === 'row0'
      }).componentInstance as ListRowComponentComponent
      row0.rowClick.emit();
      expect(onClicked).toHaveBeenCalledWith(outputArray[0])
    })
  })

Conclusion

Through this article I hope to have given the reader a brief knowledge of the key concepts required to integrate Unit testing into your Frontend code and also some tips on how to write unit tests for complex components and the way you should design your application so it becomes easy to maintain a healthy codebase.

You can find the entire code for the sample app used in this post here

Please feel free to fork and practice unit testing using this setup.

AUTHOR:

Divye Marwah — Full stack developer | Software Engineer at Zeotap

Monitoring is one of the critical yet under-appreciated aspects of infrastructure. Nobody cares when your servers/applications are running fine, but as soon as your website is down or your application is not responding, the first question that comes to our mind is “‘Why is there no monitoring enabled for this piece?”

For starters, the purpose of monitoring majorly falls into the below categories:

1. Alerting: the most important thing is knowing when something goes wrong. The first step is alerting humans when a defined event occurs aka when some threshold is breached.
2. Debugging: we refer to this when humans look into the alert, investigate and troubleshoot.
3. Trending: when humans identify the trend of the event, whether it’s an anomaly or expected, ex: how the systems are used and usage changes over time. This helps in RCA and capacity planning.

At zeotap we use AWS for our cloud infrastructure and our resources are majorly deployed in three regions: Asia Pacific (Mumbai), EU (Ireland) and US (North Virginia). We have on-demand EC2 instances running across all the three regions and we primarily use spot instances for our EMR jobs.

Earlier we used datadog (enterprise version) for our infra monitoring but we found it expensive as our tech stack grew. We found cloudwatch not flexible enough to monitor our applications, so we decided to go with an open source monitoring tool which would fit our needs much better.

Prometheus

Prometheus is a popular open source, metrics-based monitoring system. It is written in GO, so you have a single binary executable, download the binary and run prometheus, as simple as that.

It stores the metrics scraped in a time series database(TSDB) and provides PROM Query Language (PromQL) that lets you query over the metrics to see how your applications, services and servers are performing.

How Prometheus works

The Prometheus service is called Prometheus server and the things it monitor/scrapes are called targets. The targets could be a server, service, database, http endpoint, script etc. Each unit of a target such as CPU, memory, disk, network usage is called a metric. These metrics should be exposed in a prometheus format.

Prometheus works on pull approach. It pulls/scrapes metrics from the targets over the HTTP on regular intervals and stores the metric in TSDB. The metric stored can be queried in the Prometheus UI or we can integrate it to Grafana for rich visualization.

Prometheus has a configuration file called prometheus.yml, where you can specify the scrape interval, alert rules, and targets under scrape configs.

You can create multiple jobs under scrape_configs such as

• job_name: EC2 metrics,
• job_name: DB metrics

and specify the respective targets under them.

You can simply query the metrics stored in prometheus TSDB by typing PromQL expression in the prometheus UI.

How the metrics can be generated in prometheus format

Prometheus provides client libraries in all popular languages which you can instrument in your own code. Some of the popular softwares like Kubernetes and docker are already instrumented with Prometheus client libraries. For other applications, we can use third party integrations to expose their metrics to prometheus.

These third-party integration tools are called exporters. They collect the system, application metrics and report them in a Prometheus format on a specified port and path. It has to be installed in respective machines where you want to expose metrics to be monitored by prometheus.

For example, servers don’t have any mechanism to expose metrics (system stats such as cpu, memory, disk) in prometheus format, so we can use an exporter to get those metrics.

At zeotap, we use an exporter tool called telegraf from influxdata. The reason we chose telegraf is, it has integration with all the major softwares Aerospike, Elasticsearch, Mesos, Kafka, Zookeeper which are part of zeotap’s tech stack.

Below block is a configuration file of telegraf agent.

The telegraf agent runs at the interval of 60s, it collects host metrics such as CPU, disk, memory, network stats, etc., and outputs it in a Prometheus format on the port “:9999” and in the path “/metrics”.

[global_tags]
  type = "EC2"

[agent]
  interval = "60s"
  round_interval = true
  metric_batch_size = 1000
  flush_interval = "60s"
  debug = false
  quiet = true
  logfile = "/var/log/telegraf/telegraf.log"

[[outputs.prometheus_client]]
  listen = ":9999"
  path = "/metrics"
  expiration_interval = "60s"

[[inputs.cpu]]
  percpu = true

[[inputs.disk]]
  ignore_fs = ["tmpfs", "devtmpfs", "devfs"]

[[inputs.mem]]

[[inputs.net]]

If we hit the “ipaddress or hostname:9999/metrics” we can see the metrics exposed by telegraf agent.

Telegraf agent updates the metrics as per the interval specified in the telegraf.conf.

Prometheus scrapes these metrics in an interval defined in the prometheus.yml and stores the data locally in a custom TSDB.

Service Discovery

So far, we have seen how Prometheus works and how a target can expose their metrics to Prometheus.

In the Prometheus section, we learned that we can specify the targets under scrape_configs. This is easy to do when you have a handful of servers. What if you have 100s of servers though? Manually adding all the server names in prometheus.yml is not scalable.

Prometheus has a service discovery feature which allows prometheus to automatically discovers the new instances and scrapes them without manual intervention. We tried using Consul (service discovery tool), but it requires separate agent to be installed in all the target machines. We wanted to go for an agent-less mechanism for service discovery.

As our infrastructure runs on AWS, we use EC2 service discovery in our monitoring setup. We have created an IAM role with EC2ReadOnlyAccess and attached in our Prometheus server.

Below code block shows the config to discover the EC2 instances in eu-west-1 region.

We have added tags “Environment=production” ‘Monitoring = yes” in all of our production instances.

So when Prometheus queries our EC2 instances in the EU region, AWS provides all list of instances in EU. Prometheus then checks whether the instances are having the tags “Environment=production, Monitoring=yes”. It will drop the instances which don’t have the tags, from its targets list and start scraping the matching instances.

We are passing tag keys ‘Team, Application, Owner, Role, Project’ as metadata which will be added as a label to the metrics of the targets. This helps us in querying the list of instances belong to particular team, projects.

scrape_configs:

  - job_name: 'EC2-EU-region'
    ec2_sd_configs:
      - region: eu-west-1
        port: 9999
        refresh_interval: 10m
    relabel_configs:
      - source_labels: ['__meta_ec2_tag_Environment']
        regex: 'production'
        action: keep
      - source_labels: ['__meta_ec2_tag_Monitoring']
        regex: 'yes'
        action: keep
      - source_labels: ['__meta_ec2_availability_zone']
        target_label: 'Zone'
      - source_labels: ['__meta_ec2_private_ip']
        target_label: 'instance'
      - source_labels: ['__meta_ec2_tag_Team']
        target_label: 'Team'
      - source_labels: ['__meta_ec2_tag_Application']
        target_label: 'Application'
      - source_labels: ['__meta_ec2_tag_Name']
        target_label: 'Name'
      - source_labels: ['__meta_ec2_tag_Role']
        target_label: 'Role'
      - source_labels: ['__meta_ec2_tag_Owner']
        target_label: 'Owner'
      - source_labels: ['__meta_ec2_instance_state']
        target_label: 'Instance_Status'
      - source_labels: ['__meta_ec2_instance_type']
        target_label: 'Instance_Type'

This is customizable and Prometheus allows you to configure how metadata from service discovery is mapped to monitoring targets and their labels using relabelling. You can add a tag with key ‘monitoring_worthy’, value as ‘yeah’ or ‘nah’ and tell Prometheus to keep only ‘yeah’ and drop ‘nah’.

Refresh interval indicates that this service discovery happens every 10 minutes, so any new instance provisioned with above tags will be discovered in 10 minutes. Likewise if the instance is terminated, it will be removed from AWS, so when the refresh happens, the instance will not be identified and Prometheus will not scrape it.

We are using terraform to provision instances with necessary tags and run Ansible playbook to install/configure telegraf in the instances. So this way the monitoring configuration for instance is done in less than 5 minutes and the instance comes under monitoring in 10 minutes.

Alerting

We have got instances (targets) which are exposing metrics using telegraf agent (exporters), EC2 service discovery to provide the list of targets to prometheus, prometheus is configured to scrape the metrics from the targets, so the only missing piece is alerting.

Alerting with Prometheus is separated into two parts.

• Alerting rules in Prometheus send alerts to an Alertmanager.
• The Alertmanager then manages those alerts, including silencing, aggregation and sending out notifications via methods such as email, PagerDuty and Slack.

The main steps to setting up alerting and notifications are:

• Create alerting rules in Prometheus
• Configure Prometheus to talk to the Alertmanager
• Setup and configure the Alertmanager

Alerting rules:

Alerting rules allow you to define alert conditions based on PromQL and attach a severity label.

Below is an example rule, to alert if disk usage percent is above 95 for 15 minutes.

- alert: Critical Disk usage
    expr: disk_used_percent > 95
    for: 15m
    labels:
      severity: page
    annotations:
      summary: "Critical disk usage > 95%"
      description: "Critical disk usage on host {{$labels.instance}} : (current value: {{ $value }})."

This alert rule file has to specified in the prometheus.yml.

Prometheus evaluates the rules as per evaluation interval and if any one of the alert is triggered, it will forward the alert with severity label to the alertmanager, configured under alerting section.

# Alertmanager configuration
alerting:
  alertmanagers:
    - scheme: http
    static_configs:
      - targets:
        - 'alertmanager-ipaddress:9093'

rule_files:
  - "/etc/prometheus/alertmanager/hostmetrics_rules.yml"

Alert Manager:

The alertmanager installation is as straightforward as prometheus. It comes up with an executable binary file and a configuration file. Alertmanager can be installed in the same instance where prometheus is running.

In configuration, you can specify the receiver of the alerts (team that needs to be notified) and the communication channels (pagerduty, email, slack) based on the severity of the alert.

We are using three channels for alerting — slack for all types of alerts, email for critical alerts and pagerduty for severe alerts (if the application is down, or the instance is down).

Below block is the configuration of alertmanager.

route:
  repeat_interval: 3h
  receiver: slack-notifications
  routes:
    - match:
        severity: warning
      repeat_interval: 1h
      continue: true
      receiver: slack-notifications
    - match:
        severity: critical
      repeat_interval: 30m
      continue: true
      receiver: slack-notifications
    - match:
        severity: page
      repeat_interval: 10m
      continue: true
      receiver: slack-notifications
    - match:
        severity: critical
      repeat_interval: 30m
      continue: true
      receiver: email-notifications
    - match:
        severity: page
      repeat_interval: 5m
      continue: true
      receiver: pagerduty-notifications

receivers:
  - name: slack-notifications
    slack_configs:
      - channel: 'prometheus-alerts'
        api_url: ''
        send_resolved: true
        icon_url: https://avatars3.githubusercontent.com/u/3380462
        title: '{{ template "custom_title" . }}'
        text: >-
          {{ range .Alerts }}
            *Alert:* {{ .Annotations.summary }} - `{{ .Labels.severity }}`
            *Description:* {{ .Annotations.description }}
            *Details:*
            {{ range .Labels.SortedPairs }} *{{ .Name }}:* `{{ .Value }}`
            {{ end }}
          {{ end }}

- name: email-notifications
    email_configs:
      - to: receiver@example.com
        from: alerts@example.com
        smarthost: "smtp.sendgrid.net:587"
        send_resolved: true

- name: pagerduty-notifications
    pagerduty_configs:
      - service_key: ""

Below is one of the alert notifications we received on Slack for the critical disk usage event in one of the servers.

Grafana

As we mentioned in the introduction, trending helps you to identify how your systems, applications performing over time. Though prometheus comes up with a graph console it is not user friendly.

Grafana is an open source metric analytics & visualization suite. It is most commonly used for visualizing time series data for infrastructure and application analytics. We can integrate Grafana with prometheus and visualize the metrics captured by prometheus.

All we have to do is add prometheus as a data source and create dashboards for the important metrics.

For dashboard graphs, you can use the same PromQL expression (ex: disk_used_percent, cpu_usage_user).

Grafana supports variables, so you can specify metric labels as variables (region, application, instance) and pass the variables in the PromQL expression which enables multi selection in the graphs.

Summary

Monitoring ideally should liberate the ops engineers from spending time in monitoring systems, applications and notify only when human intervention is required. One should not spend every day adding, removing servers to monitoring and it should be taken care of by the system itself. It should be flexible enough to customize the alerts, versatile enough to monitor servers, applications, services, processes, etc, and not to mention, it should not burn too much of your company’s money either.

We found that Prometheus with the combination of telegraf, alertmanager, grafana is ticking all the boxes for us (opensource, customisable, alerts integration, visualisation) and the cherry on top is that we are running this in a single t2.medium instance.

About The Author

Kannan Anandakrishnan is a devops Engineer with 7 years of experience in managing large infrastructures and data platforms.

He is a Senior DataOps Engineer at zeotap, currently managing the infrastructure of zeotap with keen focus on Infrastructure as Code, building immutable infrastructure and automation.

We’re hiring for devops engineers at zeotap. Check out our open positions here.

In the previous post, we learnt why and how to transform our real world, high- dimensional-categorical data to a lower dimensional continuous space, ‘Embeddings’. In this post, we shall learn two ways to generate samples from these embeddings.

PART II : Generating samples

The idea of sampling methods is to be able to sample all such points in an appropriate number that best explains the variance in data, such that, it gives a close representation of input data. The two techniques defined below take embeddings of training data as input and outputs samples. Check the previous post to learn how we reached to embeddings from high dimensional categorical input.

1. RBF-based Sampling

With this sampling technique, we are going to exploit embeddings to our advantage, as embeddings can be used to attain a quantifiable value by calculating the Euclidean distance between two embedding vectors. This is then used to find similarity between two training data points. The basic idea is to pick more samples that are closer to the mean and move radially outwards.

Let’s dig into the specifics of RBF:

What is RBF ?

Radial Basis Function is a real-valued function, Φ, whose value only depends on the distance from some other point, c, called a center, so that:

The norm is usually Euclidean distance, although other distance functions are also possible. Reference:https://en.wikipedia.org/wiki/Radial_basis_function

What is Gaussian RBF ?

There exist multiple types of RBFs. Here’s Gaussian RBF, where x is the sample picked, μ is the mean (the c vector from above) and Φ is a defined function, check Fig. 2. Gaussian RBF is used to our advantage in the calculation of probability as the output of this function is contained between 0 and 1 for positive inputs. The output value represents the probability of acceptance of sample, x.

This definition of probability calculation will be clearer by taking two boundary conditions. If the input (candidate sample, x) is equal to μ (from Fig. 2), then the output of this RBF will be 1, i.e., the probability of selection is 1. As the distance between the input and the prototype grows, the response falls off exponentially towards 0. This ensures more selection of data points that are closer to the mean.

Graph for RBF:

Algorithm:

1. Calculate the mean (μ) of embeddings for all dimensions.

2. Pick a batch of samples and calculate the mean, μ(z), of embeddings of the batch.

3. Calculate the probability of selection from Gaussian rbf, refer Fig. 2. Taking μ from Step1 and x as μ(z) from Step2.

4. Accept/Reject the batch of samples based on the output value from Step 3 with Roulette wheel selection.

Notes :

• Roulette wheel selection is an acceptance criterion in which a random value is uniformly sampled from a unit distance [0,1]. This value is then compared to the sample probability selection from Gaussian RBF. If the output of RBF (probability of acceptance) is greater than the random sample from, the candidate sample is selected.
• Tweaking the values of β in RBF to get the desired ratio of high probability values vs low probability values is a challenge. Otherwise, it results in biased sampling, favoring more of high probability values.

2. Variant of Stratified Sampling

Stratification/Stratified sampling is the process of dividing members of the population into homogeneous subgroups (strata) before sampling. This is followed by applying simple random sampling or systematic sampling within each stratum. The objective is to improve the precision of the sample by reducing sampling errors. Reference:https://en.wikipedia.org/wiki/Stratified_sampling

Stratified Sampling:

Stratified sampling is performed on a latent space (i.e. the embedding space) and not directly on input data. Like the previous method, this sampling technique also exploits a property of embeddings, which is, attaining a quantifiable figure to find a similarity between two embedding vectors. The algorithm is defined in detail below.

Notations:

μ — Mean of embeddings

M — Number of strata points to be defined

N — Total number of input examples

f — Fraction of data to be taken from each strata

K — Number of samples desired

Algorithm:

1. Calculate mean (μ) of embeddings for all dimensions.

2. For every vector of embedding, calculate Euclidean distance of it from the μ. Check Fig. 5

3. Lay out these distances on a number line with minimum and maximum distances found from Step 2. as its extremes. Check Fig. 6

4. Divide this line of distance into M uniform buckets/strata.

5. Take a fraction, f, of samples from each strata (mini-samples)

6. Combine all such M mini-samples to form a big sample.

Estimation of parameters:

Parameters N and μ are known.

How to define M and f ?

1. First, fix on a number of samples desired, say, K < N.
2. Now, fix on a value of the number of strata points, M. The bigger the M, the better the sampling. But as M increases, so does the computational workload. There’s a trade-off between accuracy and computation here.
3. f*N ~ K. Considering this as an equation, we can calculate the value of f.

Convergence criteria for the above two sampling techniques: The mean(μ) and standard deviation(σ) of the embeddings of the sample-set and the entire data should be of the same range. The closer they are, the better the approximation.

In conclusion, a real-world high dimensional, sparse and categorical data is first transformed to a lower dimensional continuous space, embeddings. These embeddings are then taken as input for sampling techniques. RBF based sampling and variant of stratified sampling are two techniques that were discussed in this post.

Thanks for reading! If you have any feedback or questions, please leave a response below!

ABOUT THE AUTHOR

Shreya Jain is a Senior Data Scientist at zeotap. Based out of Bangalore, Shreya’s core areas of work include Machine Learning modelling, Probabilistic modelling, Neural Networks and Recommendation systems. After her studies at Birla Institute of Technology and Science Pilani, she worked at Samsung Electronics as a Software Engineer for two years before deciding to embark in an exciting data journey with zeotap!

In a real world scenario, we often come across data with high sparsity, a large number of features with a high categorical dimensionality. Due to such complexity of the data, defining a distribution for the same is a challenging task. As the distribution is not a standard one, basic sampling techniques cannot be applied here.

We overcome this challenge by transforming this complex data into a lower dimensional continuous space called Embeddings, followed by sampling techniques that approximate this data distribution.

Part I: Problems and solutions for real-world data

Part II: Generating samples

Why should I be interested in dataset samples ?

1. With the samples (representation of the dataset), we can run the numbers to quantify various estimates real quick, and simply extrapolate those numbers for the entire population.
2. The samples are meant to capture the majority of the variance of the data. Hence, training any model on this sampled data would reflect if there’s something terribly wrong with the architecture of the model.
3. In the case of unsupervised learning, it’s difficult to gather intuition in the form of correlations about the features. Hence, the manual dropout of features or manually observing samples would help gain meaningful data insights.
4. Hyper-parameter tuning for AutoML requires training on efficient samples.

PART I : Problems and solutions for real-world data

A) Problems associated with real-world data

First things first, problems associated with widely sparse and high dimensional data are immense. To make matters worse, categorical inputs with no ordinal relationships amongst themselves are also encountered. Let’s call this data — DEVIL(Data Everyone Vexes In Life) and learn more about it.

1. Computational workload

High dimensional input poses various problems. To start with, the computation cost increases exponentially as the number of possible combinations of the variables increases. This would result in an explosion of computational workload as depicted in Fig. 1 below.

2. Dimensional Redundancy

There might exist strong collinearity between features as in most cases, attributes are unlikely to be independent of one another. This introduces dimensional redundancy.

The problem of high dimensional data can be taken care of by feature engineering, but we’ll take an algorithmic approach to solve this issue.

3. Fitting data into standard distribution

In higher dimensions, see Fig. 2, data is more sparse and more flung out in the outer region than in the center of the hypercube. Hence, defining a standard probability distribution (i.e. a hypersphere) for such data (all points in the hypercube) which would contain most of its points in/around the mean of the probability distribution (say, multivariate Gaussian distribution), gets difficult as dimension increases. See, the standard probability distribution cheatsheet. As defining a probability distribution for this data is not straightforward, basic sampling techniques won’t be applied here.

The three points above constitute what’s known as theCurse of Dimensionality. Long story short, the common theme of these problems is that when the dimensionality increases, the volume of the space increases so fast that the available data becomes sparse. This sparsity is problematic for any method that requires statistical significance. In order to obtain a statistically sound and reliable result, the amount of data needed to support the result often grows exponentially with the dimensionality. Also, organizing and searching data often relies on detecting areas where objects form groups with similar properties; in high dimensional data, however, all objects appear to be sparse and dissimilar in many ways, which prevents common data organization strategies from being efficient. Reference:https://en.wikipedia.org/wiki/Curse_of_dimensionality

4. Ineffectiveness of similarity measures

Referring to Fig.2, if a constant number of examples is distributed uniformly in a high-dimensional hypercube, beyond some dimensionality most examples are closer to a face of the hypercube than to their nearest neighbor. If we approximate a hypersphere, (a distribution) by inscribing it in a hypercube, in high dimensions almost all the volume of the hypercube is outside the hypersphere, i.e. the points essentially become uniformly distant from each other. So, similarity measures based on distance functions (like Euclidean metric) are rendered less effective.

Similarity measures are even less effective when dealing with categorical variables instead of continuous, as the value would just indicate the presence/absence of certain values and no insights on the degree of similarity. The combination of multi-variate features for distance functions would further take away any knowledge of which particular features went missing.

5. Insignificance of joint probability distribution

With high dimension features and sparse categorical data, there would be a huge number of possible combinations of feature values. So, the joint probability of most of the combinations would be too low to gather any meaningful insights out of it. Referring to Fig. 3, the joint probability distribution is expressed as a probability mass function, as for our discrete case.

Hence, it makes sense to describe our joint probability distribution in the form of continuous random variables instead of discretizing every variable. So, probability density distribution (continuous) needs to be defined to get some significant values.

6. Random sampling limitations on complex data

Simple random sampling can lead to under-sampling or over-sampling of data. This is because input data are not all alike so it requires more percentage of samples to be picked up from the more frequent type of data points and vice versa for less frequent ones. Hence, a class imbalance is another reason to move into continuous space, so that input data properties can be exploited (using embeddings, defined below) to pick a percentage of samples in accordance to their existence.

Because of all the above-mentioned problems associated with high dimensional and sparse categorical input, we compress the DEVIL into lower-dimensional continuous space called Embeddings, which is explained in section B below.

B) Solutions to real-world data problems

An embedding is a relatively low-dimensional space into which you can translate high-dimensional vectors. Ideally, an embedding captures some of the semantics of the input by placing semantically similar inputs close together in the embedding space, quantifiable with the help of distance metric. Reference: https://en.wikipedia.org/wiki/Embedding

This transformation to continuous space, wherein two data points are comparable, is the basis of the definition of our sampling techniques. We shall read about it in detail in the next post. Let’s continue with embeddings now.

Visualization of embeddings:

I.Tabular form

In Fig. 5, we see that the left table with categorical and continuous data has 6 features which are reduced/compressed into 3 continuous features in the latent space. The latent space is vectors of 3 dimensions, D1, D2 and D3.

II. Embedding space

Position (distance and direction) in the vector space can encode semantics in a good embedding. For example, the above visualizations of real embeddings show geometrical relationships that capture semantic relations like the relationship between a country and its capital.

Embedding generation techniques for DEVIL:

1. Non-Linear dimensionality reduction techniques: Auto-Encoder github link , What are auto-encoders?
2. Co-occurrence based models (GloVe architecture, github link )
3. Linear dimensionality reduction techniques: PCA (unsupervised)

To learn more about embeddings, check this out.

In summary, we learned about the problems linked with high dimensional, sparse and categorical input and how we overcome these challenges by converting them into a lower dimensional continuous space called embeddings. These embeddings are then going to help us in our pursuit of sampling.

Now, proceed on to the second post in this series to learn about various sampling techniques.

ABOUT THE AUTHOR

Shreya Jain is a Senior Data Scientist at zeotap. Based out of Bangalore, Shreya’s core areas of work include Machine Learning modeling, Probabilistic modeling, Neural Networks and Recommendation systems. After her studies at Birla Institute of Technology and Science Pilani, she worked at Samsung Electronics as a Software Engineer for two years before deciding to embark in an exciting data journey with zeotap!

At zeotap, we are using AWS for all our cloud processing needs. By early 2018, we found out that our AWS expenditure had been increasing at an alarming rate with a CAGR of 8.05%.

This was worrisome. For almost every startup, the cost is one of the top priorities in infra setup (this is where cloud platforms are saviors for them) and saving big expenses also contributes to revenue for a startup. It is a big problem if we are not able to attribute the spends and analyze their worthiness, and that’s when we decided the problem had to be tackled. After closely analyzing it and having implemented a solution, we were able to reduce our expenditure by 50%. We came out with three strategies, and the savings with each can be found in the chart below.

Now, isn’t this intriguing! Let us tell you the whole story of how we were able to cut down our spending in such a drastic way.

Prologue

It all started over a beer 🍺 while having long conversations about AWS costs. We knew something had to be done and fast. Finally, the time came around when we heard zeotap was announcing its first hackathon [1], and that’s when we knew that was just the right project for us.

Problem Statement

Though we were spending hundreds of thousands of dollars, there were some basic questions that we could not answer:

1. Why is our cost increasing so much?
2. Who is actually spending so much?
3. Are we following industry practices on having correct visibility on expenditure?

After the questions were clear, we knew the scope of this project should be to provide the following features:

• Transparency over services: When you have multiple modules and products running across teams, it’s difficult to find out the break up of cost per module/product/team or owner. If we know who is spending how much, we can ask the right questions and make better management calls like ROI calls over projects.
• Monitoring resource utilization across services: We need to monitor our resources properly to know whether we are over-provisioning.
• Providing better alternatives with respect to cost: Choose from multiple billing models like RI and Storage/Compute instances from AWS.

Data Analysis and Problem Scoping

We started out by analyzing our AWS billing data. In order for us to limit the scope of our problem, we first identified the cost distribution per AWS service per region. You can find the analysis below:

This graph shows the cost distribution per service and region. Just by looking at the graph, we understood right away, what to focus on. For our hackathon, we limited the cost optimization scope to EC2 [2] and RDS [3] alone which accounted for 80% of the cost. In this article, we will only talk about EC2 since RDS has a very similar story.

EC2 provides two types of instances: On-Demand [4] and Spot Instances [5]. Out of the total cost of EC2, more than 70% was attributed to On-Demand EC2 Nodes. Hence, we chose to look more closely at on-demand EC2/RDS spend to recommend more optimal instance types in order to reduce our expenditure.

Literature Survey

After reading through people’s experience of how they actually save costs over a cloud infra and what best practices are recommended, we came out with the following points:

• Right-Sizing: Allocating just the best instance for the task. An instance is best for the task when it is optimally used in terms of memory and CPU (i.e an instance should be neither too large to be underutilized nor too small to miss the SLA of the running job).
• Monitoring: For right resizing, we need to continuously monitor our instances (their usage) on the regular interval and calculate the optimal required capacity of the instances.
• Reserved Instance: This is another technique where AWS provides significant discounts (up to 70%) compared to the on-demand instance. AWS provides different purchasing options for Reserved Instances based on customers’ comfort. For more details on RI, see the reference [6].
• Dayparting: This is another great way to reduce the spends. Dayparting means to scale up and down as per requirement. i.e scale down/stop/terminate the machines when not in use. Alternatively, for scheduled jobs/services we can launch the cluster when required and terminate once the job is done.

We picked the top 3 strategies in our project. In order to implement them, we needed a better understanding of the AWS pricing model [7] and the AWS reserved Instances [6]. Not only this, but we also needed to find ways to collect the resource utilization data. We found two viable sources to do so, Cloudwatch [8] and Prometheus [9]. Cloudwatch, though natively provided by AWS, is a paid solution while Prometheus is an open source solution. We, therefore, decided to go ahead with Prometheus.

Data Collection and Analysis Architecture

The diagram above shows the complete stack and architecture we used to come up with our solution. We used the following tools:

1. Java for data collection and munging the data from AWS and Prometheus
2. Athena [10] for computations
3. DOMO [11] for visualization

The following datasets were collected in order to arrive at the optimized recommendations.

EC2 Instance Data

Tagging was a prerequisite for this data. As per our process, all EC2 instances should be tagged with Team-Name, Owner, and Project. We collected all the relevant data from the instances using AWS SDK, which comprised of Instance ID, Instance Type, Launch Date Time, Availability Zone, Tenancy, Private IP Address, Status, Region and tags data.

AWS master pricing data

AWS provides an API to fetch the current on-demand pricing. We extracted this data to check for instances that have a similar configuration but are cheaper. We then checked all the instance types that were provided by AWS, and what their On Demand and RI (Reserved Instances) price were under various models.

Utilization Data from Prometheus

We found out that IO (Network and Disk) was seldom a problem for us but the CPU and Memory were. Hence, we extracted out the utilization data for CPU and Memory using Prometheus. Prometheus collects metrics every 5 seconds. To aggregate these metrics, we experimented with max, min, average and percentile calculations.

Percentile [12] came out as the best option and gave better recommendations. Percentile intuition is explained in the algorithm section below. We took the 80th percentile of resource utilization data over a 15 days window, and we pulled host address, timestamp, aggregate utilization value per instance. This was possible since Prometheus has inbuilt support for percentile aggregation.

EC2 Recommendation Algorithm

For providing size recommendations for any given instance, we required the following data:

1. AWS master pricing data
2. Prometheus data (Usage)
3. EC2 Instance data

From the AWS master pricing data, we needed to extract the following details for all the available instances on AWS:

• InstanceType (e.g. m4.2xlarge)
• Available CPU and Memory (# of cores and memory)
• Region (as we have different prices in different regions)
• On-demand price per unit
• RI price per unit for no upfront

Using AWS APIs, we can extract all the instance details running in our AWS account.

Algorithm Example

Consider we have one EC2 instance running in our AWS account with the following configuration:

Instance type: c3.4xlarge

Available CPU: 16 core

Available Memory: 30GB

Price (per hour): $0.840

From the Prometheus data for this instance, we observe the following metrics:

CPU usage: 25% (80 percentile metrics used here)

Memory usage: 30% (80 percentile metrics used here)

* 80 percentile or above means that 80% of the time the CPU usage is less than 25%.

Using this instance usage information, we calculated, the required CPU and Memory for the same instance based on its usage would be:

Required CPU: 16 * 25% = 4 core

Required Mem: 30 * 30% = 9 GB

As we had taken the 80th percentile, we added an additional buffer on the CPU and Memory of the recommended instance as the below formula shows:

Actual Memory Usage = Instance Memory * Usage

Buffer Memory = sqrt(Actual Memory Usage)

Recommended Memory = Actual Memory Usage + Buffer Memory

The same calculation can be done for CPU cores as well.

Why is the buffer memory calculated using the Square root function? Consider the following two examples.

Case 1. The actual memory usage is 4GB. According to the formula, the buffer memory will come out to be 2GB which is more than sufficient in this case.

Case 2. Let’s say the actual memory usage is 128GB. Again as per the formula, the buffer memory will come out to be approximately 128= 11GB.

In both the cases, we see it is a reasonable buffer to add.

Square root curve accentuates lower data points and penalizes high data points giving us an acceptable value of buffer size.

The same applies to the CPU usage as well, and now we have the following recommended instance properties as per the above formula:

Recommended CPU = 4 + √4 = 6 core

Recommended Memory = 9 + √9 = 12 GB

If we have to visualize this, see the bubble chart of AWS instances based on its CPU (x-axis), Memory (y-axis), and Price (size of bubble). The goal is to find the top N closest and cheapest instance available with the above configurations. The black bubble in the below diagram shows the required Instance with the configurations we calculated above. We find the cheapest top 3 recommendations which satisfy the required configurations and sort by the price of the instance.

As a result, our recommendations would be following:

• t3.2xlarge (CPU — 8 core, Mem — 32 GB and Price : $0.332)
• c5.2xlarge (CPU — 8 core, Mem — 16GB and Price : $0.340)
• t2.2xlarge (CPU — 8 core, Mem — 16GB and Price : $0.371)

If we resize our current instance based on the above recommendations, we would end up saving around:

Savings per hour = $0.840 — $0.332 = $0.51 per hour

Savings (in %age) = $0.51 / $0.84 * 100 = 60% approx

And this saving is for just one instance. We can apply the same algorithm on all the on-demand instances (EC2 or RDS) based on its usage and save the company a whole lot of money.

Conclusion

AWS provides a variety of ways to reduce the infrastructure cost. We have to choose the right model and the right size for each of the AWS services. For example, we should consider using Spot Instances for EMR related jobs and Reserve Instances (RIs) for long-running instances with known utilization. Further, understanding and analyzing the billing data over a period of time helps identify the optimal mix of various pricing model’s to address your infrastructure requirements.

Transparency in infra is another very important aspect to achieve infra cost optimizations and assist in ROI calls for the services deployed. Tagging the instance is one of the mandatory practices for transparency. Regular monitoring of the infra, removing or resizing underutilized resources, goes a long way in reducing spends. We are running this recommendation algorithm once a month and looking at more automated ways of applying the optimization recommendations to our cloud infrastructure.

References

[1] Zeotap Hackathon — https://www.zeotap.com/hackathon

[2] EC2 — https://aws.amazon.com/ec2/details/

[3] RDS — https://aws.amazon.com/rds/

[4] On Demand — https://aws.amazon.com/ec2/pricing/on-demand/

[5] Spot Instances — https://aws.amazon.com/ec2/spot/

[6] AWS Reserved Instances — https://aws.amazon.com/ec2/pricing/reserved-instances/

[7] AWS pricing model — https://aws.amazon.com/pricing/

[8] Cloud Watch — https://aws.amazon.com/cloudwatch/

[9] Prometheus — https://github.com/prometheus/prometheus

[10] Athena — https://aws.amazon.com/athena/

[11] Domo — https://www.domo.com/solution/data-visualization

[12] Percentile — https://en.wikipedia.org/wiki/Percentile

About The Authors

Rakesh Sharma and Aman Verma are both Senior Engineers that work on the Data Engineering team which handles the massive scale of data at zeotap. They work on the core data platform zeoCore, responsible for ingesting, refining and distributing data from our multiple data partners. Rakesh studied computer science and engineering and had a stint with Minjar Cloud Solutions before joining Zeotap. His areas of expertise include data engineering, distributed systems, and AWS Services amongst others. Aman studied electronics and communications engineering and worked at one of the largest e-commerce startups in India Snapdeal before moving to Zeotap. His areas of expertise include data engineering, distributed systems, and data platforms.

I joined the company last year in April into a team of Data Engineers. The team consisted of 3 interns and 3 Data Engineers collaborating using the version control system, Git. We didn’t have much of a structure in place to peer review the code that went into production. We did use pull requests on Github at the time, but we had all the issues in our git repositories that an early stage startup might have. We added inessential files into the repositories (such as .DS_Store) along with hardcoded paths to local files, no proper commit messages, no logical commits to name a few.

Since then, we have come a long way. Our team has grown from 6 engineers to 14 engineers. We have made significant improvements to our code review process. We ensure that at least two engineers see every piece of code before it goes into production. Our git commits are much more logical and always accompanied with much more meaningful and detailed commit messages than before. Through all of this, one of the decisions we had to make was to choose the right Git Workflow, specifically whether to “merge” pull requests or use “rebase and merge” instead.

In this article, I first discuss the “rebase” workflow that we follow at zeotap in the Data Engineering team. Then, I argue that the “rebase” workflow builds better team dynamics compared to the “merge” workflow. While the “rebase” workflow is trickier to follow, it has numerous benefits. For instance, it forces the developers to make logical commits resulting in better code readability and code ownership. It also fits well with the agile software development strategy by ensuring frequent commits and pull requests.

The Choice

At zeotap, every developer works on a particular part/feature of a project, building commits in their own branch. Once a logical portion of the feature is complete, the developer raises a pull request on Github. Then, at least two different peer engineers review this pull request before it can be merged into the master/sprint branch.

While merging the pull request, we consider two choices. We can either “merge” the pull request resulting in a merge commit, which is a special kind of commit merging a developer branch into the master/sprint branch [https://www.atlassian.com/git/tutorials/merging-vs-rebasing]. A sequence of merge commits in a Git repository builds a complex tree of commits leading up to the current version of the code.

Another choice is to “rebase” all the commits into a developer branch on top of the master/sprint branch and merge the pull request using a fast-forward merge strategy. [https://ariya.io/2013/09/fast-forward-git-merge]. This results in a structure where it looks as if all the commits were made on the master/sprint branch directly and the developer branch never really existed. In practice, we actually delete the developer branch once the pull request is merged.

Rebase Workflow at zeotap

Before we compare both the workflows, below is the “rebase” workflow that we follow at zeotap. In this workflow, we never force-update the master/sprint branch, since only developers can force-update their branch.

Rebase vs Merge

Now, let’s compare both the “merge” and “rebase” workflows.

Rebase is Hard

First of all, let me accept that “rebase” is complex, destructive and takes time to get used to. If there are conflicts, they need to be resolved for each commit that is rebased on the current branch. Once “rebase” is complete, one has to destructively force push to the remote branch resulting in not a safe practice. Github does have mechanisms available to ensure that the sprint/master branch is never modified once committed, nevertheless, the process is complex.

On the other hand, “merge” is relatively easier, requires little effort and git does most of the work for you. It creates a “merge” commit explicitly marking the merge of the developer branch into the sprint/master branch. This can lead to conflicts as well, but those are resolved in a single step for all commits in a branch unlike with the “rebase” approach.

Logical Commits

While “rebase” is hard and destructive, it reinforces the idea of logical commits. By logical commits, I mean that each commit has a complete logical unit of code such as fixing a bug, adding a feature or a part of it. This has really helped us improve the readability of our commits in a git repository. When a developer has to rebase their branch on the master/sprint branch, the sequence of new commits on the developer’s branch are applied on top of the master/sprint branch. If these commits are not logical, it could be tedious to resolve conflicts in this process. Logical commits allow us to distinctly identify how to resolve the conflicts based on the logical change that has been made in a given commit. This reinforces the idea of building logical commits while writing code. On the contrary, in the case of the “merge” workflow, the conflicts need to be resolved in the “merge” commit and no such need of creating logical commits ever arises.

Developer’s Responsibility / Ownership

When git is unable to perform “merge” due to conflicts in the code, it typically informs the developer, and he has to take the matter into his own hands. In the case of the “rebase” workflow, this is done while the pull request is getting reviewed. Whereas, in the case of the “merge” workflow, the conflicts are sometimes resolved while merging the pull request. What we really liked about the “rebase” workflow is that it takes away the choice of merging later. It becomes the responsibility of the developer to ensure all the changes go are done correctly before the pull request is merged. The developer has the responsibility of rebasing his changes on the master/sprint branch before creating the pull request compared to relying on Git or code reviewers for merging the changes in the pull request.

Commit History & Reverting Commits

This is a straightforward consequence of using the “rebase” workflow. It provides a linear history of commits in a repository which is easy to reason about. Also, from the history, it is easier to understand how the current state of the code is achieved. It is also much easier to undo commits from the linear history of commits. Whereas, with the “merge” commit, a much more complex tree structure of commits is built, which makes it difficult to identify and revert.

Agile World

This is the final advantage that I’d like to discuss in this article. I believe that the “rebase” approach fits better into the world of Agile Software Development. Zeotap’s environment is extremely fast-paced and therefore requirements and code tend to change very frequently. In such a fast paced environment, the “rebase” workflow helps us keep up with the code changes made by peer developers. We create and merge our code frequently through pull requests to ensure that our branch does not lag behind the master/sprint branch. Given that “rebase” requires resolving conflicts commit by commit for all the commits in the pull request, the fewer the commits, the easier it becomes to resolve conflicts. It, therefore, pushes developers to merge changes as frequently as possible to reduce the workaround effort when resolving conflicts. Sometimes when we do find conflicts, we also engage in discussions among ourselves to understand the reason behind them. This always leads us into becoming a much more collaborative team.

Conclusion

The “Rebase” and “Merge” approach are both excellent strategies for the Git workflow which can be adopted at the workplace. Nevertheless, I think the major difference is that the “rebase” workflow is destructive and takes some time to get used to whereas the “merge” workflow is easier to use and doesn’t require training. Despite that, the “rebase” workflow has a slight advantage over “merge” in an agile environment: it requires developers to think more deeply about their commits in order to build logical commits and take ownership of each pull request, providing a linear history which is easy to reason about.

We have been using the “rebase” workflow for more than a year now. Whenever developers join the team, we help onboard them with our Git workflow. Once they get used to this workflow, the rest of the benefits simply follow.

About the Author

Aman Mangal, Software Engineer at zeotap, is one of the many team members that works with a tremendous amount of data on a daily basis. His main tasks include managing the different partners’ data ingestion into zeotap’s systems as well as monitoring and optimizing the data pipeline for smooth functioning. After completing his computer science studies both in India and the US, Aman had the opportunity to contribute to companies like Nokia Bell Labs, Zerostack and Lumiata before joining zeotap. His areas of expertise include data engineering, distributed systems and linux containers amongst others.

For more info on Aman’s profile, visit his Linkedin page here.

At Zeotap we source data from countless enterprise partners to then refine, blend and convert it to high-performing data segments to be used for targeting. Last year the number of profiles under management grew 10X from 250M to 2.5Bn.

An introduction to our data pipeline was given here — Oozie! It’s pouring data!. In this article we will now further elaborate on our profile store, its purposes and our choice on tech. Our profile store is where every data feed finds it place. It is the heart of our advertising intelligence.

Key Terms

Profile — all attributes of a user ranging from demographic to interest and intent.

Profile store — a denormalized store of all user profiles

Segment — a collection of users which belong to a particular group satisfying a criteria.

Use Cases & Scaling Up

We already have dozens of data points for each user and this number is constantly growing. Some regions have data in more than 2 Bn rows and there are by now in excess of 7000 queries to treat the data.

Following are some of the significant data segment requirements of this database / store:

1. Creation and Export — Given a combination of attributes as filters, the database returns the matching rows or users. It also can export these segments to a file location so that they can be consumed across multiple channels.
2. Revenue Attribution — Allocating the generated revenue per segment to zeotap’s different data partners based on their contributions, data quality parameters and different prices for different attributes
3. Size Estimator — Calculating the approximate number of users in a segment in real-time
4. Analytics — Running aggregate queries to create real-time dashboards and reports on live segments

Data Model

The below table depicts the overall structure of our data profile store:

There are three main types of columns in this table:

1. ZUID — Zeotap Unique ID — a unique and purely internal randomly assigned user ID
2. User Attributes — such as age, gender and interest categories
3. Data Partner ID — data partner that the attribute orginates from

Query Patterns

Assuming that the table name is profileStore and that the attributes are the ones listed above, typical queries to build a meaningful user segment for the advertising market looks like this.

Segment creation and export queries

Segment revenue attribution queries

The query above will calculate the proportion of attributes shared by each data partner from the final database for a specific user segment — called “All female users in the profileStore” in this case. Imagine this running for 500+ columns and you can picture the complexity and scale of this task.

Segment size estimation queries

We estimate the size of a specific segment by using a representative data sample from the profile store. The query on this sample would look like this:

select count(distinct zuid) from profileStoreSample where age > 25 AND gender = ‘Male’

Important to note here is that since the filters / criteria used to create a segment can vary enormously between applications, any of the hundreds of columns with the respective AND / OR / Other combinations can be relevant.

The result of any query then, is a list of ZUIDs. The less filters we apply, the bigger the size of the segment. A simple segment comprising only males can easily be in the billions. As a result of such scale, we need an efficient way to export these results to an easily accessible stored format.

Column Oriented & MPP Databases

Now that we have described our technical and to some degree commercial use cases and some sample query patterns, let’s take a look at the core technology behind the ProfileStore: our Column Oriented Databases and MPP (Massively Parallel Processing).

Column-Oriented Databases

Column-oriented databases store all the values of a particular column together on one disk. When a query is needed, only a few disk blocks which contain the required columns are fetched to calculate the result. This requires significantly less disk I/O as opposed to row-oriented databases. A couple of smart techniques such as compression and late materialization are additionally used. They further reduce disk I/O and improve speed. [2][3]

The main difference between row and column oriented storage types is shown below:

Figure 1: Difference in storage mechanism for row and column-oriented database for a sample table

MPP Database

In Massively Parallel Processing (MPP) databases, data is partitioned across multiple servers or nodes with each server/node having memory/processors to treat the data. All communication is done via a network interconnect. Hence, disk-level sharing or contention are of no concern which is why this approach is also referred to as a ‘shared-nothing’ architecture. Leveraging parallelism, the queries can run multiple times faster than on a single host.

Amazon Redshift

During the scaling journey we’ve carefully chosen Redshift as our ProfileStore. Redshift is a database provided by AWS for data warehousing and OLAP workloads and stores its data as laid out above in a column oriented fashion on disks. It is also a MPP database with a shared-nothing architecture.

At zeotap, using Redshift goes along with the following main benefits:

AWS Managed

Redshift is a fully managed database by AWS. Operability is easier than maintaining an owned equivalent database such as Greenplum.

Bulk Load and Unload

Data can be copied from S3 to Redshift in multiple formats such as CSV, AVRO, GZIP using a COPY command. Since most of our data is stored in S3, this is very convenient. Also, loading plain CSV data into Redshift is much faster than AVRO. [4]

Redshift also has a feature to automatically store the query results to an S3 location using the UNLOAD command. This suits us very well since the results from our queries are usually very large and need to be consumed across multiple channels.

Postgres Extension

You can directly query AWS redshift using Postgres extensions such as dblink and postgres_fdw. This is useful in one of our products which is constantly sampling data from Redshift to estimate segment sizes. For more details about extension, take a look at our dedicated blog post on note: [5].

Redshift @Zeotap

The largest cluster we currently have comes from the US. This cluster is running with 10 nodes and also has the largest number of user records. We use dc2.large types of nodes since they come with SSD and are faster compared to a Dense Storage type of nodes.

Limitations and Future Work

While Redshift is generally a very good fit, one of its current advantages is also one of its biggest limitations. And that is that it does not support data structures such as JSON or Map. As a result, table modelling is limited by a flat structure of columns using trivial types. To support a one-to-many relationship, we have to model two separate tables with a loosely imposed Foreign key and use JOIN to combine multiple layers of results. When one of the tables is much smaller than another, Redshift uses HASH JOIN and otherwise MERGE JOIN to compute the results. One problem we have observed is that MERGE JOIN is slow when both tables are sufficiently large to put the memory to a challenge.

Another limitation is that with growing amounts of data, the time needed to COPY — as the only way to upload data into Redshift — processing times increase as well. Therefore, it is not too suitable for streaming or near-real time use cases.

Now luckily, at zeotap we don’t like to live with limitations, which is why we’ll dive in detail into how we overcame these challenges using other databases such as Presto (SQL query engine for files stored on HDFS or S3) in our next blog article. Stay tuned and reach out if you also never settle with the average. It’s how innovation is born and mind and skillset evolve.

References

1. http://use-the-index-luke.com/sql/anatomy
2. http://db.csail.mit.edu/projects/cstore/abadi-sigmod08.pdf
3. https://en.wikipedia.org/wiki/Column-oriented_DBMS
4. https://blog.stitchdata.com/redshift-database-benchmarks-copy-performance-of-csv-json-and-avro-9062f71f814
5. https://aws.amazon.com/blogs/big-data/join-amazon-redshift-and-amazon-rds-postgresql-with-dblink/
6. http://docs.aws.amazon.com/redshift/latest/dg/c_columnar_storage_disk_mem_mgmnt.html
7. http://docs.aws.amazon.com/redshift/latest/dg/c_high_level_system_architecture.html
8. http://docs.aws.amazon.com/redshift/latest/dg/t_Loading_data.html
9. http://docs.aws.amazon.com/redshift/latest/dg/c_unloading_data.html

About the author:

Chaitanya Bendre, Senior Software Engineer at zeotap, is one of our colleagues who handles tremendous amounts of data on a daily basis. He works on our core data engineering platform, zeoCore, which ingests and refines data from multiple partners to distribute it across a wide range of buying and other application platforms. After completing his degree in Engineering at BITS Pilani, Chaitanya had the opportunity to work with companies like Goldman Sachs and Juspay on cloud computing and data platform related projects. His areas of interests include data engineering, functional programming and distributed systems.

zeotap makes large scale, deterministic data assets easily available within the digital advertising ecosystem and other industries. Our data engineering team transforms unstructured raw data that we receive from a multitude of data partners into structured readily queryable data which can then be monetized through different distribution channels. In this article, I will begin with a high-level introduction to our data pipeline and elaborate on how we use Apache Oozie as a distributed controller for our data platform with some customisations to make it work for our use case.

Our whole data platform runs on Amazon Web Services (AWS) making heavy use of a diverse set of tools and services provided by AWS. We receive data from various data partners every day in designated S3 buckets in pre-decided formats. We then process it on AWS Elastic MapReduce (EMR) using Apache Spark through various stages in the pipeline. The processed data is uploaded to Amazon Redshift Clusters (an OLAP MPP Database) and made available for efficient querying by different teams as well as by various internal APIs that we have built on top of it. All of this is scheduled and monitored through Apache Oozie that comes along with Amazon’s EMR service.

Apache Oozie

Apache Oozie is a workflow scheduler for the Hadoop ecosystem that can schedule various MapReduce jobs across an EMR cluster with the help of yarn resource manager. Each Oozie workflow consists of a directed acyclic graph of control flow nodes called actions. A workflow can be executed once using Oozie controller or scheduled periodically using an Oozie Coordinator. An Oozie workflow is described in an XML file. Those are usually not a favourite to work with, but in Oozie they are supported with some really powerful abstractions such as workflow parametrization, fork-join action nodes for parallel execution, a rich library of EL (Expression Language) expressions and email notifications so we can quickly get over it.

Data Pipeline in Oozie

In order to set up our data pipeline in Oozie, we composed each stage of the pipeline as a sub-workflow consisting of one or more Spark actions. We also made use of HDFS actions to create the necessary folder structure and shell actions to ensure that the expected data is in fact available. We wrapped all the sub-workflows in yet another parent workflow to run them in sequence and perform the bookkeeping operations as needed. The decoupling of a parent workflow with sub-workflows for each stage enabled us to run one or more stages independently for testing and debugging purposes.

We parametrized the parent workflow to process data for every data partner using Oozie variables. We set up Oozie coordinators to run the corresponding parent workflow every day at specific times of the day. Here, we made use of EL expressions to compute paths to a folder in an S3 bucket containing the latest data for a given data partner. Finally, we added another parent workflow consisting of a Spark action for aggregating data across all data partners followed by a Shell action to upload aggregated data to Redshift.

Problems Faced — Oozie on EMR

Oozie Data Trigger

We wanted to use data triggers in our data pipeline and run a workflow as soon as all the input data would be available. Oozie’s synchronous data triggers didn’t work for our use case as a workflow can only be triggered at a fixed time interval. To circumvent the issue, we simply wrote a Shell script polling data availability every 5 minutes to check whether processed data had been written by the previous workflow.

Duplicate Action Execution

We use the autoscaling feature of AWS EMR for task nodes to minimize our cost which created an issue at the final stage of the pipeline. While the upload shell script is running, AWS scales down the cluster due to low memory requirements leading to failure of the YARN container running the script. Even when YARN brilliantly restarts the container on a different node, it delays the pipeline unnecessarily.

Looking deeper into the issue, we found out that Oozie runs two containers for each shell action, a driver container and a task container. AWS EMR schedules the driver container only on core nodes whereas task containers can run anywhere in the cluster. As we do not use auto-scaling for core nodes in the EMR cluster, the driver container continues to execute, whereas the task container dies when EMR scales down the cluster. As the driver container keeps running, it restarts the task container on failure.

To resolve the issue, we wanted to figure out a way to schedule the task container on core nodes as well. Interestingly, Oozie allowed us to uberize containers for shell actions. This meant an action could be executed inside the driver container without running any task containers. As the driver container would be scheduled only on core nodes, we would not face the issue anymore.

This is achieved by setting the property oozie.action.launcher.mapreduce.job.ubertask.enable to true in the Oozie configuration module.

Shared Libraries

Oozie shared libraries don’t have the appropriate Spark libraries required to run a Spark action job. We resolved the issue by manually adding/replacing the following libraries to the Oozie shared library folder (Oozie version 4.3.0 on EMR emr-5.6.0) -

HDFS Action

While running the pipeline, we observed that the HDFS actions available in Oozie do not work for EMRFS (HDFS equivalent library for S3) due to the unavailability of the EMRFS assembly JAR in the Oozie local libraries. To deal with this issue, we simply added a soft link to the EMRFS assembly JAR already available on the EMR Master node in the Oozie local jar directory.

This is of course just a sneak peak of all the challenges we face while dealing with data at zeotap. The fun part of all of this is that on top of being constantly challenged, we end up overcoming all of this together with the rest of the team members while learning from each other and building the team’s relationship. This definitely helps with keeping motivation high, and one can just feel it in the air.

About the author

Aman Mangal, Software Engineer at zeotap, is one of the many team members that works with a tremendous amount of data on a daily basis. His main tasks include managing the different partners’ data ingestion into zeotap’s systems as well as monitoring and optimizing the data pipeline for smooth functioning. After completing his computer science studies both in India and the US, Aman had the opportunity to contribute to companies like Nokia Bell Labs, Zerostack and Lumiata before joining zeotap. His areas of expertise include data engineering, distributed systems and Linux Containers amongst others.

For more info on Aman’s profile, visit his Linkedin page here.

Want to join the zeotap team? Check out our career site here!

Follow us on: Linkedin, Facebook, Twitter.

Today

Ieva leads zeotap´s business development efforts in North America. As part of the role Ieva and the team engage and test dozens of data partners every week against zeotap´s strict high-quality data standards. Her day-to-day responsibilities also involve leading our identity business globally and working on other strategic business development opportunities.

Background

Ieva joined zeotap in the winter of 2015 and moved to company´s headquarters in Berlin from the ancient halls of University of St. Andrews, Scotland, where she finished her Master’s degree in Finance. Ieva set foot in St. Andrews with an ambition to pursue a career in investment banking, but it was the University´s entrepreneurship professor who inspired her to move into the growing data and mobile advertising industry. Prior to that, Ieva was with the management consulting firm EY, working on a variety of data-related projects across the Europe.

zeotap journey

Ieva is one of our colleagues who reinvented her own role several times while zeotap kept growing. Over the 2 years she wore many hats and ventured from telecom operator partnerships into building agency and brand relationships in Spain and Italy. When zeotap was just getting ready to enter the US, Ieva’s ambition was to go for the highest quality data owners in the biggest advertising market in the world. With her move to the NYC, she also opened our US office in Chelsea, Manhattan.

Also, during her time with zeotap, Ieva had a chance to dive into the Indian culture — during her visit to the Bangalore office, her colleagues helped her discover the colourful customs this country has to offer.

Fun fact

Over the years, wherever she lived, Ieva has always dedicated a part of her free time to Corporate Social Responsability and volunteering projects — from her hometown Riga, where she was mentoring youth in SOS Children Villages and developing advanced teaching materials for high school students, approved and taught in schools across the country, to helping out in soup kitchens and Christmas events in Atlanta. With her move to the NYC, Ieva will also be kicking off volunteering efforts as part of the University of St Andrews Alumni Club Board. Volunteering aside, Ieva is a big fan of a healthy and active lifestyle, but having spent almost a year in Glasgow, she would never say no to a naughty but oh-so-good deep-fried Mars candy bar if you offered her one. Or a generous serving of a good old Scottish classic of scotch-soaked haggis, neeps, and tatties.

Want to join the zeotap team? Check out our career site here!

Follow us on: Linkedin, Facebook, Twitter.