Data processing process

Bub uses Spark to process massive user behaviors generated by services in modules. Redundant or simple summaries are filtered and exported to different PostgreSQL tables to store normalized basic data. For example, LoginTable and ShoppingTable record user login/purchase behavior respectively. Then set up different services in the secondary processing module, such as S1, S2. S1 accesses the database directly, while S2 accesses both the database and the statistical results of S1, and also relies on the local CSV file.

The characteristics of the summary

All right, with that said, let’s get straight to the dry stuff.

The installation

Tablesaw Contains multiple libraries. Tablesaw-core and tablesAW-jsplot are recommended. The former is the base library and the latter is used to render diagrams. Other information, such as TablesAW-HTML, tablesAW-JSON, and tablesAW-BreakerX, supports data format change and can be selected as required. In fact, the need to write two lines of code on the line, lightweight and flexible. Tablesaw-core is used as an example to describe how to install the Jar package.

Maven repository configuration:

<dependency>
  <groupId>tech.tablesaw</groupId>
  <artifactId>tablesaw-core</artifactId>
  <version>0.37.3</version>
</dependency>
Copy the code

SBT warehouse configuration:

libraryDependencies += "tech.tablesaw" % "tablesaw-core" % "0.37.3"
Copy the code

1. Import CSV directly

// Read the CSV file input.csv automatically infer the schema
val tbl = Table.read().csv("input.csv")

// Read the contents of the table
println(tbl.printAll())

/ / check the schema
println(tbl.columnArray().mkString("\n")) 
Copy the code

The output table is as follows:

Can be found to be more perfect speculation, and good support for Chinese. The output schema is:

Date column: date

String column: name

A Double column: working hours

An Integer column: pay

Tablesaw supports the following data types: SHORT, INTEGER, LONG,FLOAT,BOOLEAN,STRING,DOUBLE,LOCAL_DATE,LOCAL_TIME,LOCAL_DATE_TIME, INSTANT, TEXT, SKIP. Most column and normal table types are no different, but there are two types that need to be emphasized:

• INSTANT. Timestamps, accurate to nanoseconds, introduced in Java 8.
• The SKIP. Specifies that columns are ignored and not read in.

3. Encoding setting schema and data filling

This approach is suitable for a variety of scenarios and can import data from different data sources at run time.

The basic process is:

• Create an empty table with a name

• Schema: Adds columns with specified names and data types in sequence to the table.

• Append data to a table by row. The elements in each row are added separately to the specified column.

  To make some changes to the previous example, assuming that the data is coming from the network, the data structure serialized to local memory is:

// Define the local memory structure for data source conversion in the form of case class
case class RowData(date: LocalDate, name: String, workTime: Double, salary: Double)
Copy the code

Create a function to add the collected data set to the table:

// @param tableName specifies the tableName
// @param colNames list of column names in the table
// @param colTypes a list of data types for each column of the table
// @param rows
def createTable(tblName: String, colNames: Seq[String], colTypes: Seq[ColumnType], rows: Seq[RowData) :Table = {
  // Create table name
  val tbl = Table.create(tblName)

  // Create schema: add columns in order
  val colCnt = math.min(colTypes.length, colNames.length)
  val cols = (0 until colCnt).map { i =>
    colTypes(i).create(colNames(i))
  }
  tbl.addColumns(cols: _*)

  // Add data
  rows.foreach { row =>
    tbl.dateColumn(0).append(row.date)
    tbl.stringColumn(1).append(row.name)
    tbl.doubleColumn(2).append(row.workTime)
    tbl.doubleColumn(3).append(row.salary)
  }

  tbl
}
Copy the code

The above illustrates the complete process of adding data: create tables, add columns, and append elements to columns. All creation and deformation can be done based on these three basic operations.

1. Traversal and deformation

For example, print the names of all records in the demo table in order:

// Get the name column, indexed by the column name
val nameCol = tbl.stringColumn("name")

// traversal by line number
(0 until nameCol.size()).foreach( i =>
      println(nameCol.get(i))
)

// Use the traversal interface provided by column directly
nameCol.forEacch(println)
Copy the code

In addition to traversal, another common use is to deform a column into another column: type invariant value change; Type changes. Taking working hours as an example, we regard working hours no less than 8 as full attendance:

// Get the time column based on the index of the column
val workTimeCol = tbl.doubleColumn(2)

// deformation 1: map, the output column type is the same as the input column type
val fullTimeCol = workTimeCol.map { time =>
  // The time type is Double, so we need to convert the result to Double, otherwise the compilation fails
  if (time >= 8)
    1.0
  else
    0.0
}

// Deformation 2: mapInto, input/output columns can be of different data types, but the target columns of the same size need to be created in advance
val fullTimeCol = BooleanColumn.create("Full-time", 
                                       workTimeCol.size()) // Create a Boolean column that records the attendance tag
val mapFunc: Double2BooleanFunction = 
  (workTime: Double) => workTime >= 8.0                    // Create a mapping function based on SAM
workTimeCol.mapInto(mapFunc, fullTimeCol)                  / / deformation
tbl.addColumns(fullTimeCol)                                // Add columns to the table
Copy the code

The output is:

3. The filter

Tablesaw Column filtering criteria are defined as Selection. You can set the criteria to and, or, or Not. Each type of column provides a direct interface generation condition prefixed with “is”. Select * from work hours = 8 and pay < 1000 between 2019-01-09 and 2019-01-10:

// Set time filtering criteria
val datePattern = DateTimeFormatter.ofPattern("yyyy-MM-dd")
val dateSel = tbl.dateColumn(0)
                 .isBetweenIncluding(LocalDate.parse("2019-01-09", datePattern),
                                     LocalDate.parse("2019-01-10", datePattern))
// Set time filter criteria
val workTimeSel = tbl.doubleColumn(2).isEqualTo(8.0)
// Set the compensation filtering criteria
val salarySel = tbl.doubleColumn(3).isLessThan(1000)
// filter table for all conditions
tbl.where(dateSel.and(workTimeSel).and(salarySel))
Copy the code

The output results are as expected:

The connection

Join two tables with a common column name. The basic way is to concatenate data from other columns in the same table with a common column key. There are differences according to different combinations of ways:

• Inner. The intersection of data in public columns, other filtering.
• Outer. Data in the public column is union, and missing data is set to a default null value. Specifically, it can be divided into three categories:
  • LeftOuter. Result table public column data is exactly the same as the left table, not filtered in it, missing setting null value.
  • RightOuter. Result table public column data and right table exactly the same, not in the filter, missing set null value.
  • FullOuter. The common column data of the result table is the union of the two tables.

For example, add a new table tBL2 to record where each person works:

Connect to the demo table in inner mode:

val tbl3 = tbl.joinOn("name").inner(tbl2)
Copy the code

The contents of TBL3 are:

You can see that both Tom and Jerry are filtered out by the intersection join of names.

table

Unlike group aggregation, when grouped by column, you may not need to aggregate the same set of data into a single value, but save it for more complex operations. In this case, you need to split tables. The interface is simple: tbl.spliton (col…) Sets the set of column names for a subtable. Such as:

// Divide the tables by name and location, and save the generated child tables to the List
tbl.splitOn("name"."Location").asTableList()
Copy the code

summary

Tablesaw Bug briefly introduced the functions and usage of tablesAW. Based on my own experience, I like it the most:

• Unified and clear API interface
• Interactive charts are easy to generate and can be connected to the Web

In addition, tablesAW development and maintenance is in full swing, and we expect more interesting features to be added in the future.