DBTimes

Data Warehouse Source Loader library. 1

Configuration file. 1

Configuration schema. 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

Actual - sourceloader-config-schema.json

As a document - Data Warehouse Source Loader

Configuration file at a glance

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

· The load configuration is described inside the "sourceLoad" object.

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

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

Optional attributes are highlighted:

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

· "loadControl", highlighted in yellow and 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":

    "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" :

            { "id" : "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 collisions with actual data columns, metadata columns can be prefixed with a string defined in the "fileDestination.metadataColumnsPrefix" field.

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

There can be three types of destination file 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, the 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 the user ID

· EffectiveDate – set to the date of the load, or the value from the column designated as the effective date, or from the file name, or from the 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 the latest copy of the source for each unique key value present.

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 an MD5 hash of all source columns

· RowTimestamp – set to load timestamp

· CreatedOn – set to load timestamp

· CreatedBy – set to the user ID

· EffectiveDate – set to the date of the load

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 the values change for a 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 an MD5 hash of all source columns

· RowTimestamp – set to load timestamp

· CreatedOn – set to load timestamp

· CreatedBy – set to the user ID

· Version – a number starting with 1

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

· EffectiveDateEnd – the far-future date for the latest version, or the date one day before the EffectiveDateStart of the next version

Schema evolution

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

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

· Changing the prefix for metadata columns

· Adding new columns

· Deleting columns

· Changing the column type – the new type must be compatible with the old type via CAST. The exception is when the new type is Boolean. The following old values are converted to true: “1”, “y”, “yes”, “true”, and these to false: “0”, “n”, “no”, “false” – irrespective of letter 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 multiline fields, and some fields, such as SQL SELECT or MongoDB aggregation, can be very long, the configuration allows some fields to have a value that is a string or an array of strings.

Example of a long field with an array value

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

File Name Pattern

A regular expression for the file names to process. The files can have an effective date as part of the name. In that case, the files are processed in ascending date 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 default to 1 if absent.

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

Examples of namePattern with an effective date (the second one has only a year, so month and day will default 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

Example of namePattern without an effective date:

NFL-Teams\\.csv

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

For example, if the source directory has these 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 pattern "pbp-(?<effectiveDate>(?<year>20[0-9][0-9]))\.csv" will result in two files being processed:

"pbp-2013.csv"

"pbp-2014.csv"

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

Incremental Load Query

When a load action defines an 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 that is used as the 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 that is used as the 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 that selects the 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 a 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, either a full or an incremental load can be used. For an 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

The "sparkParams" object in this file is not part of the loader configuration schema. An application can have arbitrary fields outside the "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 the configuration against the schema.

Arguments:

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

Returns:

Unit

Throws an exception if validation fails.

Example:

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

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

loadData

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

Arguments:

appConfig: com.typesafe.config.Config – the configuration

Returns:

Unit – the method logs errors and warnings and throws an exception if there are validation errors.

Example:

// validate configuration
val appConfig = ConfigFactory.load(configFileName)
DataSourceLoader.validateConfig(appConfig: Config)

Callback for Custom Location

For a 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 a Scala object or a Java static method.

signature:

               callbackForCustomLocation( param: Option[Config] ): String

parameter:

               value of "applicationSpecific" object; if the "applicationSpecific" object is not defined in the 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, additional processing is needed on the source data before it can be saved to a destination file.

A separate object in the configuration defines the post process, e.g.,

"postProcess" :

{

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

  "moduleName": "SrcLoaderNFL",

  "methodName": "postProsessPlayers",

  "stgSources" :

    { "id" : "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 a Scala object or a Java static method.

signature:

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

parameter 1:

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

parameter 2:

               loaded source

returns:

               post-processed  source

If "stgSources" is defined in the configuration, each of the sources will be available inside the callback as a view whose  name matches the "id" value.

For example, for the 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 a main method 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()