DBTimes

Data Warehouse Source Loader library. 1

Configuration file. 1

Configuration schema versions. 1

Configuration file at a glance. 2

Configuration file notes. 4

Metadata columns and types of file versioning. 4

Non-versioned with no unique key. 4

Non-versioned with unique key. 4

Incremental Load Query. 6

Incremental Load Unique keys. 7

Full load vs. Incremental load vs. Initial load. 7

Sample configuration file. 8

Object DataSourceLoader 10

Methods. 10

validateConfig. 10

loadData. 11

Callback for Custom Location. 11

Callback for Post Process. 12

Example of using Source Loader 13

Data Warehouse Source Loader library

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

Configuration file

Configuration schema versions

Configuration schema versions are maintained independently from sourceloader library versions.

Library version will only include in its build supported configuration schema versions.

Data Warehouse Source Loader Schema version 1

Data Warehouse Source Loader Schema version 2

· Added Schema Evolution

Configuration file at a glance

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

· 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.

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,

          "metadataColumnsPrefix": "_Src_",

          "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" },

…

],

        "schemaEvolution": {
          "allowMetadataColumnsPrefixChange": true,
          "columnAdd": {
          "isAllow": true
          },
         "columnDelete": {
            "isAllow": true
          },
          "columnTypeChange": {
            "isAllow": true
          }
        },

"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

Metadata columns and types of file versioning

When the Loader creates a parquet file with source data it adds metadata columns after the last source column. To avoid collision with actual data columns metadata columns can be prefixed by a string defined in "fileDestination.metadataColumnsPrefix" field.

Which metadata columns are added depend on the type of versioning requested for a given source.

There can be three types of versioning:

· Non-versioned with no unique key

· Non-versioned with unique key

· Versioned

Non-versioned with no unique key

This type of versioning occurs when "fileDestination.isVersioned" is false and there are no columns with "isUniqueKey" set to true.

For this type of versioning loader creates a separate copy of the source data for each effective date.

The following metadata columns are added to source columns:

· RowTimestamp – set to load timestamp

· CreatedOn – set to load timestamp

· CreatedBy – set to user id

· EffectiveDate – set to the date of the load, or the value from column designated as effective date, or from file name, or from configuration

Non-versioned with unique key

This type of versioning occurs when "fileDestination.isVersioned" is false and there are columns – one or more - with "isUniqueKey" set to true

For this type the Loader creates a latest copy of the source for each present unique key value.

The following metadata columns are added to source columns:

· RowUniqueKey – set to a concatenation of all fields with "isUniqueKey" set to true

· RowMergeKey– currently not used

· RowHash – set to MD5 hash for all source columns

· RowTimestamp – set to load timestamp

· CreatedOn – set to load timestamp

· CreatedBy – set to user id

Versioned

This type of versioning occurs when "fileDestination.isVersioned" is set to true.

For this type the Loader creates a separate copy or version of the row when any of values changed for new effective date.

The following metadata columns are added to source columns:

· RowUniqueKey – set to a concatenation of all fields with "isUniqueKey" set to true

· RowMergeKey– currently not used

· RowHash – set to MD5 hash for all source columns

· RowTimestamp – set to load timestamp

· CreatedOn – set to load timestamp

· CreatedBy – set to user id

· Version – a number starting with 1

· EffectiveDateStart – the effective date of the load when the row created

· EffectiveDateEnd – the far future date for the latest version or the date 1 less of the EffectiveDateStart of the next version

Schema evolution

Schema evolution handles scenario when the source schema changes for consecutive loads.

Processing of following schema changes can be enabled in schema evolution configuration:

· Changing the prefix for metadata columns

· Adding new columns

· Deleting columns

· Changing of column type – new type must be compatible with the old type via CAST. The exception is when new type is Boolean. The following old values are converted to true: “1”, “y”, “yes”, “true”, and to false: “0”, “n”, “no”, “false” – irrespective of string case

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

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

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

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,

          "metadataColumnsPrefix": "_Src_",

          "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()