maestro: a distinguished conductor

The maestro library is responsible for providing convenient APIs for marshalling and orchestrating data around for etl type work.

The primary goal of maestro is to provide the ability to make it easy to manage data sets with out sacrificing safety or robustness. This is achieved by sticking with strongly-typed schemas describing the fixed structure of data, and working on APIs for manipulating those structures in a sensible way that it can scale to data-sets with 100s of columns.

Scaladoc

• Github Pages: https://commbank.github.io/maestro/index.html
• Extended documentation: https://github.com/CommBank/maestro/tree/master/doc

maestro is designed to work with highly structured data. It is expected that all data-sets manipulated by maestro at some level (maybe input, output or intermediate representations) have a well defined wide row schema and fixed set of columns.

At this point, maestro supports thrift for schema definitions.

maestro uses the thrift schema definition to derive as much meta-data and implementation of custom processing (such as printing and parsing) as it can. It then provides APIs that use these "data type" specific tools to provide generic "tasks" like generating analytics views.

This is not the place for a full thrift tutorial, so I will skip a lot of the details, but if you don't understand thrift or would like more complete documentation then http://diwakergupta.github.io/thrift-missing-guide/ is a really good reference.

So if a dataset was going to land on the system, we would define a schema accurately defining the columns and types:

#@namespace scala au.com.cba.omnia.etl.customer.thrift

struct Customer {
  1  : string CUSTOMER_ID
  2  : string CUSTOMER_NAME
  3  : string CUSTOMER_ACCT
  4  : string CUSTOMER_CAT
  5  : string CUSTOMER_SUB_CAT
  6  : i32 CUSTOMER_BALANCE
  7  : string EFFECTIVE_DATE
 }

This is a simplified example, a real data set may have 100s of columns, but it should be enough to demonstrate. The important points here are that order is important, the struct should be defined to have fields in the same order as input data, and types are important, they should accurately descript the data (and will be used to infer how the data should be parsed and validated).

A pipeline is defined in terms of a cascade. This terminology comes from the underlying technology used, but it is easier to think of it as a set of jobs (with an implied partial-ordering, based on data dependencies).

A cascade can be made up of scalding jobs, hive queries (future), shell commands, or maestro tasks. The hope is that most pipelines can be implemented with just maestro, and the corner cases can be easily handled by hive or raw scalding jobs.

A pipeline built only from maestro tasks.

import com.twitter.scalding._
import au.com.cba.omnia.maestro._
import au.com.cba.omnia.etl.customer.thrift._

class CustomerCuscade(args: Args) extends CascadeJob(args) with MaestroSupport[Customer] {
  val maestro = Maestro(args)

  val delimiter = "|$|"
  val env           = args("env")
  val domain        = "CUSTOMER"
  val input         = s"${env}/source/${domain}"
  val clean         = s"${env}/processing/${domain}"
  val outbound      = s"${env}/outbound/${domain}"
  val dateView      = s"${env}/view/warehouse/${domain}/by-date"
  val catView       = s"${env}/view/warehouse/${domain}/by-category"
  val features      = s"${env}/view/features/ivory"
  val errors        = s"${env}/errors/${domain}"
  val now           = yyyyMMdd.format(new java.util.Date)
  val byDate        = Partition.byDate(Fields.EFFECITVE_DATE)
  val byCategory    = Partition.byFields(Field.CUSTOMER_CAT, Fields.CUSTOMER_SUB_CAT)
  val filters       = Filter.exclude(Fields.EFFECTIVE_DATE)
  val cleaners      = Clean.all(
    Clean.trim
  )
  val validators    = Validator.all(
  )
  val filter        = RowFilter.keep

  def jobs = List(
    maestro.load[Customer](delimiter, input, clean, errors, now, cleaners, validators, filter),
    maestro.view(byDate, clean, dateView),
    maestro.view(byCategory, clean, catView),
    maestro.ivory(clean, features),
    maestro.sqoop(clean, output, filters)
  )
}

Maestro allows you to write directly to Hive tables and to run queries on Hive tables as part of a Maestro job.

Create a HiveTable to describe/reference a specific hive table. This is required by the other hive related methods as identifier. The source and sink methods on the HiveTable provide Scalding sources and sinks for typed pipes to read from or write to. name provides a fully qualifed name that can be used inside hql.

viewHive allows the Maestro job to write out the data to a partitioned hive table in parquet similar to view. However, it also creates the hive table if it doesn't already exist. Otherwise, it just verifies the schema.

hiveQuery enables the running of a hive query as part of the Maestro job. Apart from the query string it also expects a list of input hive tables and optionally an output table. This is required for Maestro to properly schedule interdependent flows. It is solely use for scheduling and has no effect on the query itself. In order to run a hive query the Maestro job needs to extend MaestroCascade. See the example for more details.

Currently it is not possible for our implementation to read in data from the partition columns. Instead it is expected that all the data is solely contained inside the core columns of the table itself. It is, therefore, not possible to partition on the same column as a field of the thrift struct (instead a duplicate column with a different name is required). Partition columns can only be used for hive performance reasons and not to carry information.

In order for the job to work the hive-site.xml needs to be on the classpath when the job is initiated and on every node.

• Writing out hive files currently only works if the metastore is specified as thrift endpoint instead of database.

    <property>
      <name>hive.metastore.uris</name>
      <value>thrift://metastore:9083</value>
    </property>
  

• In order to run queries the hive-site.xml need to include the yarn.resourcemanager.address property even if the value is bogus.

    <property>
      <name>yarn.resourcemanager.address</name>
      <value>bogus</value>
    </property>
  

• In order to run queries with partitioning the partition mode needs to be set to nonstrict.

    <property>
      <name>hive.exec.dynamic.partition.mode</name>
      <value>nonstrict</value>
    </property>
  

You can start with the example hive-site.xml. To use this either install it on your cluster, or add it to your project's resources directory so that it is included in your jar.

import scalaz.{Tag => _, _}, Scalaz._

import com.twitter.scalding._, TDsl._

import org.apache.hadoop.hive.conf.HiveConf

import au.com.cba.omnia.maestro.api._, Maestro._
import au.com.cba.omnia.maestro.core.codec._
import au.com.cba.omnia.maestro.example.thrift._

class CustomerCascade(args: Args) extends MaestroCascade[Customer](args) {
  val env           = args("env")
  val domain        = "customer"
  val inputs        = Guard.expandPaths(s"${env}/source/${domain}/*")
  val errors        = s"${env}/errors/${domain}"
  val conf          = new HiveConf
  val validators    = Validator.all[Customer]()
  val filter        = RowFilter.keep
  val cleaners      = Clean.all(
    Clean.trim,
    Clean.removeNonPrintables
  )

  val dateTable =
    HiveTable(domain, "by_date", Partition.byDate(Fields.EffectiveDate) )
  val idTable =
    HiveTable(domain, "by_id", Partition(List("id"), Fields.Id.get, "%s"))
  val jobs = Seq(
    new UniqueJob(args) {
      load[Customer]("|", inputs, errors, Maestro.now(), cleaners, validators, filter) |>
      (viewHive(dateTable, conf) _ &&&
        viewHive(idTable, conf)
      )
    },
    hiveQuery(
      args, "test",
      s"INSERT OVERWRITE TABLE ${idTable.name} PARTITION (id) SELECT id, name, acct, cat, sub_cat, -10, effective_date FROM ${dateTable.name}",
      List(dateTable, idTable), None, conf
    )
  )
}

Maestro uses the metadata available from the thrift definition to generate out significant supporting infrastructure customized for your specific record type. This gives us the ability to refer to fields for partitioning, filtering and validation in a way that can be easily be checked and validated up front (by the compiler in most circumstances, and on start-up before things run in the worst case) and that those field references can have large amounts of interesting metadata which allows us to automatically parse, print, validate, filter, partition the data in a way that we know will work before we run the code (for a valid schema).

Tasks are not a concrete concept, it is just a name used to indicate that a function returns a scalding job for integration into a cascade. Tasks differ from raw Jobs in that they rely on the metadata generated by Maestro and can generally only by executed via their scala api and are not intended to be standalone jobs that would be run by themselves.

Partitioners are really simple. Partitioners are just a list of fields to partition a data set by.

The primary api is the list of fields you want to partition on:

Partiton.byFields(Fields.CUSTOMER_CAT, Fields.CUSTOMER_SUB_CAT)

The api also has special support for dates of the yyyy-MM-dd form:

Partiton.byDate(Fields.EFFECTIVE_DATE)

This will use that field, but split the partitioning into 3 parts of yyyy, MM and dd.

Filters again are really simple (hopefully). By default filters allow everything. A filter can then be refined via either blacklists (i.e. exclude these columns) or whitelists (i.e. only include these columns).

Examples:

 // everything except effective date
Filter.exclude(Fields.EFFECTIVE_DATE)

 // _only_ effective date and customer id
Filter.include(Fields.EFFECTIVE_DATE, Fields.CUSTOMER_ID)

Validators start to get a bit more advanced, but hopefully not too bad. A Validator can be thought of as something that is a function from the record type to either an error message or an "ok" tick of approval. In a lot of cases this understanding can be simplified to saying it is a combination of a Field to validate and a Check to apply. There are a few builtin checks provided, if you want to do custom checking you can fail back to defining a custom function.

Validator.all(
  Validator.of(fields.EFFECTIVE_DATE, Check.isDate),
  Validator.of(fields.CUSTOMER_CAT, Check.oneOf("BANK", "INSURE")),
  Validator.of(fields.CUSTOMER_NAME, Check.nonempty),
  Validator.of(fields.CUSTOMER_ID, Check.matches("\d+")),
  Validator.by[Customer](_.customerAcct.length == 4, "Customer accounts should always be a length of 4")
)

The best tip for advanced pipelines is to look carefully at how the maestro tasks are implemented. You have the same set of tools available to you and can jump down to the same lower-level of abstraction by just implementing a scalding job, and extending the Maestro class.