import com.ignition.SparkPlug
import com.ignition.frame._
import com.ignition.types._

import BasicAggregator._

val flow = DataFlow {
 
  // create an input data grid with 4 columns of the specified types and fill it with test data
  // in real life, one will probably use CassandraInput or TextFileInput etc.
  val grid1 = DataGrid(string("id") ~ string("name") ~ int("weight") ~ date("dob")) rows (
    ("j1", "john", 155, date(1980, 5, 2)),
    ("j2", "jane", 190, date(1982, 4, 25)),
    ("j3", "jake", 160, date(1974, 11, 3)),
    ("j4", "josh", 120, date(1995, 1, 10))
  )

  // create another input data grid with 1 column and two rows.
  // when a row contains only one element, you can skip ()
  val grid2 = DataGrid(string("name")) rows ("jane", "josh")

  // our first pipeline will use SQL join to merge the input grids by 'name' field.
  // SQLQuery step is a merger, i.e. it can have any number of inputs and 1 output
  // in SQL, you can refer to the inputs using names input0, input1, etc.
  val queryA = SQLQuery("""
    SELECT SUM(weight) AS total, AVG(weight) AS mean, MIN(weight) AS low
    FROM input0 JOIN input1 ON input0.name = input1.name
    WHERE input0.name LIKE 'j%'""")

  // SelectValues step is a transformer, which allows renaming the data columns,
  // change data types, retain or delete columns
  val selectA = SelectValues() rename ("mean" -> "average") retype ("average" -> "int")

  // DebugOutput() simply prints the data and passes it through to the next node, if connected
  val debugA = DebugOutput()

  // this notation means, that the outputs of steps grid1 and grid2 need to be
  // connected to the inputs 0 and 1 of step queryA, the output of queryA –
  // to the input of step selectA, and finally the output of selectA –
  // to the input of debugA 
  (grid1, grid2) --> queryA --> selectA --> debugA

  // the second pipeline also uses SQLQuery to process the data from one input
  val queryB = SQLQuery("SELECT SUBSTR(name, 1, 2) AS name, weight FROM input0")

  // BasicStats() provides aggregation functions and grouping
  val statsB = BasicStats() groupBy ("name") aggr ("weight", AVG, MAX, COUNT_DISTINCT)

  // another debug output
  val debugB = DebugOutput()

  // a simple pipeline of steps
  grid1 --> queryB --> statsB --> debugB

  // Ignition uses a lazy evaluation model, i.e. only those steps that contribute
  // to the final result will be evaluated. The last line of the DataFlow definition
  // must contain a list of terminal nodes that need to be evaluated.
  (debugA, debugB)
}

// SparkPlug helper object connects to the Spark runtime and runs the data flow.
SparkPlug.runDataFlow(flow)