DBTimes
Data Warehouse Source
Comparer library
Configuration file at a
glance
Example of using Source
Comparer
Data Warehouse Source Comparer is used to compare tabular results from various sources. It can be used for reconciliation or to verify processing.
Actual
- sourcecomparer-config-schema.json
As a document - Data Warehouse Source Comparer
This configuration shows a compressed version, leaving out some options. Refer to the Configuration schema for the complete definition.
· The comparer configuration is described inside the "sourceCompare" object.
· A compare scenario is the unit of comparison. Compare scenarios are defined as elements of the "compareScenarios" array.
· The configuration file can have additional attributes at the root level, if needed, in addition to the "sourceCompare" 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": "Data Source Comparer",
"spark.master": "local",
"spark.driver.host": "localhost",
"spark.sql.session.timeZone": "UTC"
},
"sourceCompare":
{"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"
:
{
"label" : "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":
"…", … "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"
:
{
"label" : "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", … }, }, |
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 }, … ] }}
{
"sparkParams": {
"spark.app.name": "Source
Reconciliation"
},
"sourceCompare": {
"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":
{
"label": "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":
{
"label": "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":
{
"label": "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":
{
"label": "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":
{
"label": "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":
{
"label": "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)"
}
}
]
}
}
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 ___"
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\"",
" }",
" }",
"]"
],
package com.dbtimes.dw.sourcecomparer
object DataSourceComparer
validateConfig(appConfig: Config): Unit - validates the configuration against the schema.
Arguments:
appConfig: com.typesafe.config.Config – the configuration that has "sourceCompare" 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)
compare(appConfig: Config): Unit – compares sources for all active scenarios. A Spark session must be created before calling this method.
Arguments:
appConfig: com.typesafe.config.Config – the configuration that has "sourceCompare" as one of its fields. The 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()
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.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()}