DBTimes

Data Warehouse Source Loader library. 1

Configuration file. 1

Configuration schema. 1

Configuration file at a glance. 2

Configuration file notes. 4

Comment fields. 4

Multiline values. 4

File Name Pattern – see also namePattern. 4

Incremental Load Query- see also dbtable, query, "aggregation.pipeline". 5

Incremental Load Unique keys – see also uniqueKeys. 6

Full load vs. Incremental load vs. Initial load. 6

Sample configuration file. 6

Object DataSourceLoader 9

Public Interface. 9

validateConfig. 9

loadData. 9

Callback for Custom Location. 10

Callback for Post Process. 10

Example of using Source Loader 11

Data Warehouse Source Loader library

Data Warehouse Source Loader is used to maintain the Data Warehouse Staging area.

Configuration file

Configuration schema

Data Warehouse Source Loader Schema version 1

· Load configuration is described inside "sourceLoad" object

· Load action is the unit of the load. Load actions are defined as elements of "loadActions" array.

· Configuration file can have additional attributes if needed at the root level in addition to "sourceLoad" object.

Configuration file at a glance

This configuration shows a compressed version by leaving out some options. Refer to Configuration schema for complete definition.

Optional attributes are highlighted:

· "isRemoveDuplicateRows" : false, highlighted in grey and are shown with default values, or

· "loadControl", highlighted in yellow and are not required, or

· “sparkParams”, highlighted in magenta is an example of additional application specific attributes that are not used by the library.

  "sparkParams":

  {

    "spark.app.name": "NFL Data Source Loader",

    "spark.master": "local",

    "spark.driver.host": "localhost",

    "spark.sql.session.timeZone": "UTC"

  },

  "sourceLoad":

    "sourceLoadConfigVersion": "1",

    "loadActions":

        "name" : "Load some file",

        "isActive": true,

        "fileLoadControl" :

          "parquet" : {},

          "dir" : "/dw/data/control/"

},

"fileSource" :

  "isRemoveDuplicateRows": false,

  "effectiveDate" : "2020-03-24",

OR DBMS source

"dbmsSource" :

  "isRemoveDuplicateRows": false,

  "effectiveDate" : "2020-03-24",

          "fileSystemLocation" :

            "dir": "/dw/data/sources/…",

            "namePattern": "my-file\\.csv"

},

OR

"internetLocation" : {

  "url": "https://…"

  "user": "…",

  "password": "…",

  "isDisableSslVerification": true

},

OR

"customLocation" :
{
"packageName": "...",
"moduleName": "...",
"methodName": "someMethodForCustomLocation",

"methodRetuns": "FileData",
"applicationSpecific": { }
},

  "sqlServer" : {

    "url" : "jdbc:sqlserver://… ",

    "dbtable" : "…",

    "numPartitions" : 10,

    "query": "…”,

    "prepareQuery": "…”,

…

},

OR

"oracle" : {

  "url" : "jdbc:oracle:thin://… ",

  "dbtable": "…",

  "numPartitions": 10,

  "query": "…”,

  "prepareQuery": "…",

…

OR

DB

"mongoDb" :
{

  "connection.uri" : "jdbc:sqlserver://… ",

  "database": "…",

"collection": "…",
"aggregation.pipeline": [
    "[",
    " {",
    "    $project:",
    "       {",
    "        _id: 0,",
    "        numberValue: \"$numberValue\",",
    "        stringValue: \"$_string_Value_\",",
    "        booleanValue: \"$booleanValue\",",
    "        dateValue: \"$dateValue\",",
    "        nullValue: \"$nullValue\"",
    "      }",
    " }",
    "]"
    ],
"aggregation.allowDiskUse": false,

…
}

  "csv" :

    "mode" : "PERMISSIVE",
    "header" : true,
    "encoding" : "ISO-8859-1",

"dateFormat" : "MM/dd/yyyy",

…

},

  "processedFilesDir": "/dw/data/sources/processed/"

},

  "user": "…",

  "password": "…",

  "incrementalLoad" :

    "uniqueKeys": "( SELECT SomeName, SomeOtherName FROM my-table ) AS tbl"

    "delimForWatermarkColumns": ":::",

},

        "fileDestination" :

          "isVersioned" : true,

          "parquet" : {},

          "path" : "/dw/data/…/file-name",

          "previousCopyPath" : "/dw/data/…/prev-copy-file-name"

},

        "schema" :

          { "colName": "SomeName", "colType": "Integer", "isUniqueKey": true, "isPartitionColumn": true }

          { "colName": "SomeOtherName", "colType": "String" },

…

],

"postProcess" :

{

"packageName": "...",

"moduleName": "...",

"methodName": "someMethodForPostProsess",

          "stgSources" :

            { "moniker" : "PlayByPlay",

              "description" : "Play by play s/sheet for all games",

              "fileSource" : { "path" : "C:\\ASCODE\\DATA\\NFL\\STAGING\\stgPbp.parquet",  "parquet" : { } }

},

…

],

          "applicationSpecific": { }

},

},

        "name" : "Load some other source",

…

},

…

Configuration file notes

Comment fields

Any field that starts with two underscores is a comment and is ignored. A value for the comment field can be of any type.

For example, this field is a comment

"__previousCopyPath old" : "/dw/staging/team.prev.parquet"

Multiline values

Since JSON does not allow for multiline fields and some fields, like SQL SELECT or Mongo DB aggregation, can be very long, configuration allows for some fields to have value of a string or array of string.

Example of a long field with array value

"aggregation.pipeline": [
"[",
" {",
"    $project:",
"       {",
"        _id: 0,",
"        numberValue: \"$numberValue\",",
"        stringValue: \"$_string_Value_\",",
"        booleanValue: \"$booleanValue\",",
"        dateValue: \"$dateValue\",",
"        nullValue: \"$nullValue\"",
"      }",
" }",
"]"
],

File Name Pattern – see also namePattern

A regular expression of file names to process. The files can have effective date as part of the name. In that case the files are processed in a date ascending order.

The effective date must use capturing groups with specific names for date parts: effectiveDate, year, month and day. Month and day are optional and each will be defaulted to 1 if absent.

(?<effectiveDate>(?<year>20[0-9][0-9])(?<month>[0-1][0-9])(?<day>[0-3][0-9]))

The examples of namePattern with effective date (the second one has a year only, so month and day will be defaulted to 1):

pbp-(?<effectiveDate>(?<year>20[0-9][0-9])-(?<month>[0-1][0-9])-(?<day>[0-3][0-9])).*\.csv

pbp-(?<effectiveDate>(?<year>20[0-9][0-9]))\.csv

The example of namePattern without effective date:

NFL-Teams\\.csv

All the files in the source directory that match the pattern will be processed and moved to processedFilesDir

If for example the source directory has files

"combine.csv"

"pbp-2013.csv"

"pbp-2013 - Copy.csv"

"pbp-2014.csv"

"pbp-2014 - Copy.csv"

"pbp-2015 - Copy.csv"

"pbp-2016 - Copy.csv"

The following patttern "pbp-(?<effectiveDate>(?<year>20[0-9][0-9]))\.csv", will result in processing of two files

"pbp-2014.csv"

"pbp-2013.csv"

The effective date for the first file will be 2014-01-01, and for the second – 2015-01-01.

Incremental Load Query- see also dbtable, query, "aggregation.pipeline"

When load action defines incremental load, the query attribute should have a statement that loads data incrementally based on a watermark column or columns. The watermark column is a column whose previously loaded MAX value is used to determine the new rows for the incremental load.

Example 1 - SQL Server:

"dbtable" : "( SELECT *, CONVERT( VARCHAR( 200), CONVERT( BINARY(8), [RowVersion], 1 ), 2 ) AS RowVersionStr FROM [dbo].[ARMResetTest] WHERE [RowVersion] > 0x:::RowVersionStr::: ) AS tbl",

where, RowVersionStr is the column in the result set and is used as watermark. The loader will substitute :::RowVersionStr::: (see loadActions.dbmsSource.incrementalLoad.delimForWatermarkColumns ) with the maximum value of previously loaded data from the "fileDestination". For example, if the maximum value for the attribute is 000000000002E784, the above statement will be replaced with "( SELECT *, CONVERT( VARCHAR( 200), CONVERT( BINARY(8), [RowVersion], 1 ), 2 ) AS RowVersionStr FROM [dbo].[ARMResetTest] WHERE [RowVersion] > 0x000000000002E784 ) AS tbl".

Example 2 – MongoDB:

"aggregation.pipeline": [

  "[",

  "  {",

  "     $match: { \"$expr\": { $gt: [ \"$GameDate\", { \"$toDate\": \":::GameDate:::\" } ] } } ",

  "  },",

  "  {",

  "    $project:",

  "       {",

  "        _id: 0,",

  "         \"_id_str\" :{ \"$toString\": \"$_id\" },",

  "         GameId: \"$GameId\",",

  "         GameDate: \"$GameDate\",",

  "         Quarter: \"$Quarter\",",

  "         Minute: \"$Minute\",",

  "         Second: \"$Second\",",

  "         OffenseTeam: \"$OffenseTeam\",",

  "         DefenseTeam: \"$DefenseTeam\"",

  "      }",

  "  }",

"]"

],

where, GameDate is the column in the result set and is used as watermark. The loader will substitute :::GameDate::: (see loadActions.dbmsSource.incrementalLoad.delimForWatermarkColumns ) with the maximum value of previously loaded data found in "fileDestination".

Incremental Load Unique keys – see also uniqueKeys

Unique keys is a query to select unique keys from the database. Unique keys are needed to delete rows that are no longer in the source.

Example for SQL Server

        "uniqueKeys" : "( SELECT SecId, AdjDt FROM [dbo].[ARMResetTest] ) AS tbl"

Example for MongoDB

        "uniqueKeys": "[ { $project : { _id: 0,  \"_id_str\": { \"$toString\": \"$_id\" } } } ]"

Full load vs. Incremental load vs. Initial load

There are two ways to load data from DBMS source – full and incremental. Full load should not be confused with initial load. Full load means that the entire source table is read. Full load is always used for the very first or initial load. For subsequent loads full or incremental load can be used. For the incremental load the configuration must have the incremental load section.

	Order of the load in time
	Initial Load	Subsequent load
Full Load	a	a
Incremental Load		a

Table 1. Applicability of full/incremental load

Sample configuration file

"sparkParams" object in this file is not part of loader configuration schema. Application can have arbitrary fields outside of "sourceLoad" object.

{
"sparkParams":
{
    "spark.app.name": "NFL Data Source Loader",
    "spark.master": "local",
    "spark.driver.host": "localhost",
    "spark.sql.session.timeZone": "UTC"
},
"sourceLoad":
{
    "sourceLoadConfigVersion": "1",
    "loadActions":
    [
      {
        "name" : "Load Team file",
        "isActive": true,
        "fileLoadControl" :
        {
          "parquet" : {},
          "dir" : "/dw/data/control/"
        },
        "fileSource" :
        {
          "isRemoveDuplicateRows": false,
          "fileSystemLocation" :{
            "dir": "/dw/data/sources/teaminfo",
            "namePattern" : "NFL-Teams\\.csv"
          },
          "csv" :
          {
            "header" : true
          },
          "processedFilesDir": "/dw/data/sources/processed/"
        },
        "fileDestination" :
        {
          "isVersioned" : false,
          "parquet" : {},
          "path" : "/dw/data/staging/team.parquet",
          "previousCopyPath" : "/dw/data/staging/team.prev.parquet"
        },
        "schema" :
        [
          { "colName" : "TeamName","colType" : "String" },
          { "colName" : "SeasonAbbreviation", "colType" : "String" },
          { "colName" : "SeasonYear", "colType" : "Integer" },
          { "colName" : "SeasonTeamName", "colType" : "String" },
          { "colName" : "ActiveSince", "colType" : "Integer" }
        ]
      },
      {
        "name" : "Load Players file",
        "isActive": true,
        "fileLoadControl" :
        {
          "parquet" : {},
          "dir" : "/dw/data/control/"
        },
        "fileSource" :
        {
          "isRemoveDuplicateRows": false,
          "fileSystemLocation" :{
            "dir": "/dw/data/sources/players",
            "namePattern" : "players_2013-12-12.csv"
          },
          "csv" :
          {
            "header" : true
          },
          "processedFilesDir": "/dw/data/sources/processed/"
        },
        "fileDestination" :
        {
          "isVersioned" : false,
          "parquet" : {},
          "path" : "/dw/data/staging/stgplayers.parquet",
          "__ optional-field previousCopyPath" : "/dw/data/staging/stgpbp.prev.parquet"
        },
        "schema" :
        [
          { "colName" : "name", "colType" : "String", "isUniqueKey" : true },
          { "colName" : "first_name", "colType" : "String" },
          { "colName" : "last_name", "colType" : "String" },
          { "colName" : "birth_city", "colType" : "String" },
          { "colName" : "birth_state", "colType" : "String" },
          { "colName" : "birth_country", "colType" : "String" },
          { "colName" : "birth_date", "colType" : "String", "isUniqueKey" : true },
          { "colName" : "college", "colType" : "String" },
          { "colName" : "draft_team", "colType" : "String" },
          { "colName" : "draft_round", "colType" : "String" },
          { "colName" : "draft_pick", "colType" : "String" },
          { "colName" : "draft_year", "colType" : "Integer" },
          { "colName" : "position", "colType" : "String" },
          { "colName" : "height", "colType" : "String" },
          { "colName" : "weight", "colType" : "Integer" },
          { "colName" : "death_date", "colType" : "String" },
          { "colName" : "death_city", "colType" : "String" },
          { "colName" : "death_state", "colType" : "String" },
          { "colName" : "death_country", "colType" : "String" },
          { "colName" : "year_start", "colType" : "Integer" },
          { "colName" : "year_end", "colType" : "Integer" }
        ]
      },
      {
        "name" : "Load ARMReset table",
        "isActive": true,
        "fileLoadControl" :
        {
          "parquet" : {},
          "dir" : "/dw/data/control/"
        },
        "dbmsSource" :
        {
          "effectiveDate" : "2020-05-13",
          "sqlServer" : {
            "url" : "jdbc:sqlserver://DESKTOP\\SQLSERVER2019:1433;database=testdb;integratedSecurity=true;",
            "dbtable" : "( SELECT *, CONVERT( VARCHAR( 200), CONVERT( BINARY(8), [RowVersion], 1 ), 2 ) AS RowVersionStr FROM [dbo].[ARMResetTest] WHERE [RowVersion] > 0x:::RowVersionStr::: ) AS tbl",
            "numPartitions" : 10
          },
          "incrementalLoad" :
          {
            "uniqueKeys" : "( SELECT SecId, AdjDt FROM [dbo].[ARMResetTest] ) AS tbl",
            "delimForWatermarkColumns": ":::"
          }
        },
        "fileDestination" :
        {
          "isVersioned" : true,
          "parquet" : {},
          "path" : "/dw/data/staging/armresettest.parquet",
          "previousCopyPath" : "/dw/data/staging/armresettest.prev.parquet"
        },
        "schema" :
        [
          { "colName" : "SecId", "colType": "Integer", "isUniqueKey": true, "isPartitionColumn" : true },
          { "colName" : "AdjDt", "colType": "Date", "isUniqueKey": true }
        ]
      }
    ]
}
}

Object DataSourceLoader

package com.dbtimes.dw.sourceloader

object DataSourceLoader

Methods

validateConfig

validateConfig(appConfig: Config): Unit - validates configuration against schema. Schema version to validate against is defined in "sourceLoadConfigVersion" field in the configuration. All configuration errors must be fixed before proceeding with the load.

Arguments:

appConfig: com.typesafe.config.Config – configuration that has "sourceLoad" as one of the fields. Configuration can include other fields – they will be ignored.

Returns:

Unit

Throws an exception if validation failed

Example:

import com.typesafe.config.{ConfigFactory, Config}

val appConfig = ConfigFactory.load(configFileName)
// validate configuration. If there any configuration error the method will throw an exception
DataSourceLoader.validateConfig(appConfig: Config)

loadData

loadData(appConfig: Config): Unit - load data into destination (i.e., staging) files based on the configuration.

Arguments:

appConfig: com.typesafe.config.Config – configuration

Returns:

Seq[String]- collection of strings with errors. If there are no errors the collection will be empty

Example:

// validate configuration
val appConfig = ConfigFactory.load(configFileName)
val errors: Seq[String] = DataSourceLoader.validateConfig(appConfig: Config)
if ( !errors.isEmpty ) {
val errorMessage = errors.mkString( "Loader Configuration ERROR(s)\n *", "\n *", "" )
throw new RuntimeException(errorMessage)
}

Callback for Custom Location

For custom file location the configuration would look like this

"fileSource" :
{
"customLocation" :
{
    "packageName": "com.dbtimes.nflsourceloader",
    "moduleName": "SrcLoaderNFL",
    "methodName": "sampleForCsvFileCustomLocation",
    "methodReturns": "FileData",
    "applicationSpecific": { }
},
"csv" :
{
    "header" : true,
    "multiLine": true,
    "quote": "\"",
    "escape": "\""
},
"processedFilesDir": "/data/nfl/processed/"
},

The callback function must be implemented in Scala object or Java static method.

signature:

               callbackForCustomLocation( param: Option[Config] ): String

parameter:

               value of "applicationSpecific" object; if "applicationSpecific" object is not defined in configuration - None

returns:

               file path or file content depending on configuration "methodReturns" - "FilePath" or "FileData”

For example, for "methodReturns": "FileData"

def sampleForCsvFileCustomLocation( param: Option[Config] ): String = {
"""name,age,city
Alice,30,"New
York"
Bob,25,"Los
Angeles""""
}

or, for "methodReturns": "FilePath",

def sampleForCsvFileCustomLocation( params: Option[Config] ): String = {
s"/some-path/csv-file.csv"
}

Callback for Post Process

In some instances it is needed to do some additional processing on the source data before that data can be saved to a destination file.

A separate object in configuration would define post process, e.g.,

"postProcess" :

{

  "packageName": "com.dbtimes.nflsourceloader",

  "moduleName": "SrcLoaderNFL",

  "methodName": "postProsessPlayers",

  "stgSources" :

    { "moniker" : "PlayByPlay",

      "description" : "Play by play s/sheet for all games",

      "fileSource" : { "path" : "C:\\ASCODE\\DATA\\NFL\\STAGING\\stgPbp.parquet",  "parquet" : { } }

],

  "applicationSpecific": {

    "field1": "value1",

    "field2": "value2",

  }

},

The callback function must be implemented in Scala object or Java static method.

signature:

               callbackForPostProcess( param1: Option[Config], param2: DataFrame ): DataFrame

parameter 1:

               value of "applicationSpecific" object; if "applicationSpecific" object is not defined in configuration - None

parameter 2:

               loaded source

returns:

               post processed source

If "stgSources" is defined in configuration, each of the sources will be available as a view with the name matching the "moniker" value.

For example, for above "postProcess" object the callback may look like this:

def postProcessPlayers( appSpecificAsOption: Option[Config], dfSource: DataFrame ): DataFrame = {
val spark = SparkSession.builder().getOrCreate() // this gets previously created session
val appSpecific = appSpecificAsOption.get
val field1 = appSpecific.getString("field1") // result - "value1"
val field2 = appSpecific.getString("field2") // result - "value2"
val viewSource = "Players" + java.util.UUID.randomUUID.toString.replace("-", "_" )
val sql = s"""
          |SELECT *
          |FROM $viewSource
          |WHERE draft_team IN (SELECT DISTINCT OffenceTeam FROM PlayByPlay ) """.stripMargin
val df = spark.sql(sql)
df
}

Example of using Source Loader

Create a new project with the main like this. The first argument is the configuration file.

import org.apache.spark.sql.SparkSession

import com.typesafe.config.{ConfigFactory, Config}

import com.dbtimes.dw.sourceloader.DataSourceLoader

def main(args: Array[String]) {

  val configFileName = args(0)

  val appConfig = ConfigFactory.load(configFileName)

// validate configuration
val errors: Seq[String] = DataSourceLoader.validateConfig(appConfig: Config)
if ( !errors.isEmpty ) {
val errorMessage = errors.mkString( "Loader Configuration ERROR(s)\n *", "\n *", "" )
throw new RuntimeException(errorMessage)
}

  // configure and start Spark

…

  // run the loader to load data into staging area 

  val spark = sparkSessionBuilder.getOrCreate()

  DataSourceLoader.loadData(srcLoaderConfig)

  spark.stop()