agmenc / planet7

Scala library for fast ETL and reconciliation.



Fast ETL and reconciliation tool for Scala users:

  • Load and merge CSVs.
  • Rename, re-order and filter columns.
  • Sort and validate rows
  • Diff two CSVs, or any lists of comparable elements
  • Use external CSV parsing libraries such as CSVReader


Supported Scala versions:


libraryDependencies += "com.github.agmenc" %% "planet7" % "0.1.13"

Basic ETL:

In this example, we convert from a two-column input format:

Name Id
Bob A
Dave C

to a three-column output format:

Id Value Name
1 X bob
3 X dave

We also manipulate the data:

  • Change the names to lower case
  • Default "Value" to "X"
  • Map the characters "A" and "C" to the numerals "1" and "3".
"We can map data and specify default values by column" in {
  val twoColumns = Csv("""

  val threeColumns = Csv("""

  def alphaToNum(alpha: String): String = alpha match {
    case "A" => "1"
    case _ => "3"

  val result = twoColumns
    .columnStructure("ID", "Value", "Name")
      "Value" -> (_ => "X"),      // Default value (for the empty column we added)
      "ID" -> alphaToNum,         // Mapping function
      "Name" -> (_.toLowerCase)   // Mapping function

  result.rows.toList must equal(threeColumns.rows.toList)

All CSV Features

  "All available CSV-manipulation features" in {
    import planet7.tabular._

    val csv = Csv(new File(inputPath))
      .assertAndAbort(                      // Fail-fast validations
        Validations.rowNotTruncated)        // Let's check early, before we select columns
        "Company",                          // Keep any column we name; remove the rest
        "Company account" -> "Company ID",  // Rename "Company account" to "Company ID"
        "First name",
        "Postcode" -> "Zip code")
        "Zip code" -> postcodeLookupTable,  // Specify a (String) => String to change data
        "Surname" -> (_.toUpperCase))
      .assertAndReport(                     // Reported validations are appended to the row
        "Zip code" -> validZipCode _)       // Report any invalid zip codes
      .columnStructure(ignore("Zip code"))  // Drop column, now we've validated against it

    write(sort(csv), outputPath)            // Sort the output and write to disk

    Diff(Csv(new File(outputPath)), Csv(new File(modelAnswerPath)), NaiveRowDiffer) mustBe empty
  } // ```

### More Examples

These examples are taken from [CsvSpec.scala]( They are not exhaustive, so see [CsvSpec.scala]( for a full range of working examples.

**Extract and remodel a CSV:**

"Extract a CSV, remodel it, and convert the data" in {
  import planet7.tabular.CompanyAccountsData._

  // CSV file with header: First name,Surname,Company,Company account,Postcode,Pet names
  val someFile = asFile("before.csv")

  // Retain only three of the original columns, in a different order, renaming
  // "Postcode" to "Zip code", and adding "Fee owing"
  val reshapedCsv = Csv(someFile)
    .columnStructure("Surname", "First name", "Postcode" -> "Zip code", "Fee owing")
      "Zip code" -> postcodeLookupTable,  // Map the old postcodes to zip codes, using a Map
      "Surname" -> (_.toUpperCase),       // Make all surnames upper case
      "Fee owing" -> (_ => "0.00")        // Add a default value for "Fee owing" of 0.00

  // Now convert the data to your data model, or export to a feed, or reconcile against another source, etc.
  // reshapedCsv.rows map ( ... )

Export a CSV:

Materialises the data. Use this in tests or for small files, but consider iterating through Csv.rows and writing to a BufferedWriter or similar, to avoid sucking all your data into the heap at once.

val allYourData: String = export(reshapedCsv)

Use an external CSV parser:

Pimp external libraries into TabularDataSources:

implicit def fromCsvReader(reader: CSVReader): TabularDataSource = new TabularDataSource {
  override val header = reader.readNext() match {
    case Some(items) => Row(items.toArray)
    case None => throw new NoDataInSourceException(reader.toString)

  override def rows = => Row(items.toArray))

  override def close() = reader.close()

Now work with Csv in the usual way:

"We can use external parsers such as (the incredibly slow) CSVReader" in {
  import LargeDataSet._

  val csv = Csv(

  csv.header must equal(expectedHeader) must be (expectedFirstRow)


Convert two CSVs to a canonical format and Diff them, aggregating and formatting the output for easy reconciliation:

"We can Diff Csv instances and generate readable output" in {
  import planet7.Diff
  import planet7.tabular.CompanyAccountsData._

  val before = Csv(asFile("before.csv"))
    .columnStructure("First name", "Surname", "Company", "Company account" -> "Company ID", "Postcode")
      "Postcode" -> postcodeLookupTable,
      "Company" -> (_.toUpperCase)

  val after = Csv(asFile("after_with_diffs.csv"))
    .columnStructure("First name", "Surname", "Company", "Company ID", "Postcode")

  val diffs: Seq[(Row, Row)] = Diff(before, after, RowDiffer(before.header, "Company ID"))

  // The resulting diffs are yours to play with. Let's group them: missing rows, added rows, or just plain different rows.
  val summary = diffs.groupBy {
    case (row, EmptyRow) => "Missing"
    case (EmptyRow, row) => "Added"
    case (row1, row2) => "Diffs"

  // We can Diff rows which have changed. We zip the header information with each row, so that we know the names of the fields which changed.
  val fieldDifferences = summary("Diffs") map {
    case (leftRow, rightRow) => NonSortingDiff( zip, zip, FieldDiffer)

  // Let's print the name of the field which changed, and the before and after values
  val readableDiffs = fieldDifferences map (FieldDiffer.prettyPrint(_).mkString(", "))
  printSummary(summary, readableDiffs)
  assert(readableDiffs === List(
    "Postcode: 43205 -> 432666, Company: ENIM SIT AMET INCORPORATED -> ENIM SIT AMET LIMITED",
    "Postcode: 22656 -> 22756"


All structure and filtering operations are lazy. Aside from the header row, the datasource is only read when the Csv is exported, or you iterate through the Csv.rows iterator.


External parsers, large datasets or naive coding can slow down CSV transformations. Don't optmise prematurely; instead, time your operations and fix only those that need fixing:

"Users of the planet7 library can gauge the performance impact of external parsers such as CsvReader" in {
  import planet7.timing._

  val timer = new Timer(2)
  import timer._

  for (i <- 1 to 5) {
    t"overallTime" {
      val csvReader = t"shouldBeQuick" {"before.csv")) }
      val csv = t"shouldAlsoBeQuick" { Csv(csvReader) }
      t"veryExpensive" { export(csv) }

  timer.overallTime.average must be < 150.0

More examples:

See CsvSpec.scala for a full range of working examples