DBTimes

Data Warehouse Source Comparer library. 1

Configuration file. 1

Configuration schema. 1

Configuration file at a glance. 1

Sample configuration file. 4

Configuration file notes. 7

Comment fields. 7

Multiline values. 7

Object DataSourceComparer 8

Public Interface. 8

validateConfig. 8

compare. 8

Example of using Source Comparer 9

Data Warehouse Source Comparer library

Data Warehouse Source Comparer is used to compare tabular results from various sources. It can be used for reconciliation or to verify processing.

Configuration file

Configuration schema

Data Warehouse Source Comparer Schema version 1

·         Comparer configuration is described inside "sourceCompare" object

· Compare scenario is the unit of compare. Compare scenarios are defined as elements of "compareScenarios" array.

· Configuration file can have additional attributes if needed at the root level in addition to "sourceCompare" 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": "Data Source Comparer",

    "spark.master": "local",

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

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

  },

  "sourceCompare":

    "sourceCompareConfigVersion": "1",

    "logFileDir": "/dw/data/logs/sourceComparer/",

    "compareScenarios":

        "name" : "Compare some file with the source",

        "isActive": true,

        "isDebugDwLib" : false,

        "floatThreshold" : 0.01,

        "maxSameDifferences" : 10,

        "compareResult" :

          "file" :

            "dir": “/dw/data/compare-result/",

            "csv" :

              "header" : true

},

          "schema" :

            { "colName" : "Id", "isUniqueKey" : true },

            { "colName" : "Dt", "isUniqueKey" : true },

            { "colName" : "AnotherValue", "isInclude" : true  }

},

"leftSource" :

{

"moniker" : "Some name for the left source",

          "file" :

            "path": "/dw/data/sources/parquet_file",

            "parquet" : {}

},

OR

"file" :

  "path": "/dw/data/sources/some_file.csv",

  "csv" :

"mode" : "PERMISSIVE",

"header" : true,

"encoding" : "ISO-8859-1",

…

},

},

OR

"dbms" :

  "sqlServer" : {

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

    "dbtable" : "…",

    "numPartitions" : 10,

    "query": "…”,

    "prepareQuery": "…”,

…

},

  "user": "…",

  "password": "…",

},

OR

"dbms" :

"oracle" : {

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

    "dbtable": "…",

    "numPartitions": 10,

    "query": "…”,

    "prepareQuery": "…",

…

},

   "user": "…",

   "password": "…",

},

OR

"dbms" :

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

…
}

  "user": "…",

  "password": "…",

},

          "schema" :

            { "colName" : "SomeName", "colType" : "String", "isPartitionColumn" : true },

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

…

],

          "selectClauseWithFieldsToCompare": "SELECT SomeIdName AS Id, InitialDate AS Dt, SomeValue, SomeOtherValue, AnotherValue, UpdSrc "

},

"rightSource" :

{

"moniker" : "Some name for the right source",

          "file" :

            "path": "/dw/data/sources/parquet_file",

            "parquet" : {}

},

OR

"file" :

  "path": "/dw/data/sources/some_file.csv",

  "csv" :

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

…

},

},

OR

"dbms" :

  "sqlServer" : {

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

    "dbtable" : "…",

    "numPartitions" : 10,

    "query": "…”,

    "prepareQuery": "…”,

…

},

  "user": "…",

  "password": "…"

},

OR

"dbms" :

"oracle" : {

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

  "dbtable": "…",

  "numPartitions": 10,

  "query": "…”,

  "prepareQuery": "…",

…

},

 "user": "…",

 "password": "…"

},

OR

"dbms" :

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

…

}

  "user": "…",

  "password": "…",

},

          "schema" :

            { "colName" : "SomeName", "colType" : "Long", "isPartitionColumn" : true },

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

…

],

          "subsetQuery": "SELECT Id, Dt, SomeValue, SomeOtherValue, AnotherValue, UpdSrc FROM ___ "

        "name" : "Some other scenario",

         … elements for other scenario

},

…

Sample configuration file

{
"sparkParams": {
    "spark.app.name": "Source Reconciliation"
},
"sourceCompare": {
    "sourceCompareConfigVersion": "1",
    "compareScenarios": [
      {
        "name": "Compare Team file",
        "isActive": false,
        "isDebugDwLib": false,
        "floatThreshold": 0.01,
        "compareResult": {
          "file": {
            "dir": "/dw/data/compare-result/",
            "csv": {
              "header": true
            }
          },
          "schema": [
            {
              "colName": "TeamName",
              "isUniqueKey": true
            },
            {
              "colName": "SeasonAbbreviation",
              "isUniqueKey": true
            },
            {
              "colName": "SeasonYear",
              "isInclude": true
            }
          ]
        },
        "leftSource": {
          "moniker": "Original",
          "file": {
            "path": "/dw/data/nfl/team-info/nfl-teams.csv",
            "csv": {
              "header": true
            }
          },
          "schema": [
            {
              "colName": "TeamName",
              "colType": "String"
            },
            {
              "colName": "SeasonAbbreviation",
              "colType": "String"
            },
            {
              "colName": "SeasonYear",
              "colType": "Integer"
            },
            {
              "colName": "SeasonTeamName",
              "colType": "String"
            },
            {
              "colName": "ActiveSince",
              "colType": "Integer"
            }
          ],
          "subsetQuery": "SELECT TeamName, SeasonAbbreviation, SeasonYear FROM ___"
        },
        "rightSource": {
          "moniker": "SQL Server",
          "dbms": {
            "sqlServer": {
             "url": "jdbc:sqlserver://my-server\\SQLSERVER2019:1433;database=my-db-name;integratedSecurity=true;",
              "dbtable": "( SELECT *, CONVERT( VARCHAR( 200), CONVERT( BINARY(8), [RowVersion], 1 ), 2 ) AS RowVersionStr FROM [dbo].[ARMResetTest] ) as tbl",
              "numPartitions": 10
            }
          },
          "subsetQuery": "SELECT TeamName, SeasonAbbreviation, SeasonYear FROM ___"
        }
      },
      {
        "name": "Compare with SQL Server",
        "isActive": true,
        "isDebugDwLib": false,
        "floatThreshold": 0.01,
        "compareResult": {
          "file": {
            "dir": "/dw/data/compare-result/",
            "csv": {
              "header": true
            }
          },
          "schema": [
            {
              "colName": "SecId",
              "isUniqueKey": true
            },
            {
              "colName": "AdjDt",
              "isUniqueKey": true
            },
            {
              "colName": "RPB"
            },
           {
              "colName": "LoanLifeFloorLo"
            },
            {
              "colName": "NumLoans"
            },
            {
              "colName": "EffDt",
              "isInclude": false
            },
            {
              "colName": "UpdSrc"
            }
          ]
        },
        "leftSource": {
          "moniker": "SQL Server",
          "dbms": {
            "sqlServer":
            {
              "url": "jdbc:sqlserver://my-server\\SQLSERVER2019:1433;database=my-db-name;integratedSecurity=true;",
              "__connectionUri": "jdbc:sqlserver://DESKTOP-T3QOCRG\\SQLSERVER2019:1433;database=embs;",
              "__user": "my_user_id",
              "__password": "my_password",
              "dbtable": "( SELECT *, CONVERT( VARCHAR( 200), CONVERT( BINARY(8), [RowVersion], 1 ), 2 ) AS RowVersionStr FROM [dbo].[ARMResetTest] ) as tbl",
              "numPartitions": 10
            }
          },
          "schema": [
            {
              "colName": "SecId",
              "colType": "Integer",
              "isPartitionColumn": true
            }
          ],
          "subsetQuery": "SELECT SecId,AdjDt,RPB,LoanLifeFloorLo,NumLoans,EffDt,UpdSrc FROM ___ "
        },
        "rightSource": {
          "moniker": "Parquet file",
          "file": {
            "path": "/dw/data/data/nfl/staging/armresettest.parquet",
            "parquet": {}
          },
          "subsetQuery": "SELECT SecId,AdjDt,RPB,LoanLifeFloorLo,NumLoans,EffDt,UpdSrc FROM ___ WHERE _SrcDt_EffectiveDateEnd = '2099-01-01'"
        }
      },
      {
        "name": "Compare to itself parquet",
        "isActive": true,
        "isDebugDwLib": false,
        "floatThreshold": 0.01,
        "compareResult": {
          "file": {
            "dir": "/dw/data/compare-result/",
            "csv": {
              "header": true
            }
          },
          "schema": [
            {
              "colName": "SecId",
              "isUniqueKey": true
            },
            {
              "colName": "AdjDt",
              "isUniqueKey": true
            },
            {
              "colName": "RPB"
            },
            {
              "colName": "LoanLifeFloorLo"
            },
            {
              "colName": "NumLoans"
            },
            {
              "colName": "EffDt",
              "isInclude": false
            },
            {
              "colName": "UpdSrc"
            }
          ]
        },
        "leftSource": {
          "moniker": "left Parquet file",
          "file": {
            "path": "/dw/data/data/nfl/staging/armresettest.parquet",
            "parquet": {}
          },
          "subsetQuery": "SELECT SecId,AdjDt,RPB,LoanLifeFloorLo,NumLoans,EffDt,UpdSrc FROM ___ WHERE _SrcDt_EffectiveDateEnd = '2099-01-01'"
        },
        "rightSource": {
          "moniker": "right Parquet file",
          "file": {
            "path": "/dw/data/data/nfl/staging/armresettest.parquet",
            "parquet": {}
          },
          "subsetQuery": "SELECT SecId,AdjDt,IF( SecId = 9997, RPB - 1, RPB) AS RPB,LoanLifeFloorLo,IF( SecId = 1388360, NumLoans + 4, NumLoans ) AS NumLoans,EffDt,UpdSrc FROM ___ WHERE SecId NOT IN ( 9998,9999)"
        }
      }
    ]
}
}

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

"__ subsetQuery": "SELECT * FROM ___"

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

Object DataSourceComparer

package com.dbtimes.dw.sourcecomparer

object DataSourceComparer

Methods

validateConfig

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

Arguments:

appConfig: com.typesafe.config.Config – configuration that has "sourceCompare" 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)

compare

compare(appConfig: Config): Unit – compare sources for all active scenarios. Spark session must be created prior to calling this method.

Arguments:

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

Returns:

Unit

Example:

val configFileName = …

val srcComparerConfig = ConfigFactory.load(configFileName)

val spark = sparkSessionBuilder.getOrCreate()

DataSourceComparer.compare(srcComparerConfig)

spark.stop()

Example of using Source Comparer

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.sourcecomparer.DataSourceComparer

def main(args: Array[String]) {

  val configFileName = args(0)

  val srcComparerConfig = ConfigFactory.load(configFileName)

// validate configuration

val errors: Seq[String] = DataSourceComparer.validateConfig(srcComparerConfig)

if ( !errors.isEmpty ) {
val errorMessage = errors.mkString( "Comparer Configuration ERROR(s)\n *", "\n *", "" )
throw new RuntimeException(errorMessage)
}

  // configure and start Spark

…

  val spark = sparkSessionBuilder.getOrCreate()

  DataSourceComparer.compare(srcComparerConfig)

  spark.stop()