DBTimes
Configuration file at a
glance
effDateColumnNameInDatesToProcess
getLastProcessedStgSourceTimestamp
isStgSourceChangedSinceLastLoad
effDateColumnNameInDatesToProcess
getLastProcessedStgSourceTimestamp
isStgSourceChangedSinceLastLoad
Data Warehouse ETL library
Data Warehouse ETL (DW ETL) is a library to implement ETL process for dimensional model. Dimensional model design should follow Kimball methodology (see “The Data Warehouse Toolkit” by Kimball/Ross).
The library implements ETL process for the dimensional model based on configuration and using specific code needed for each dimension and fact table.
The library’s Dim and Fact classes generically handle various aspects of dimension design and lifecycle.
Dim class:
- Load dimension from multiple sources
- Historical load of data on initial load for both Type 1 and Type 2 dimensions
- Re-load dimension from specific point in time
- Type 1 and Type 2 changes, and by extension Type 3 and Type 4. Type 2 dimension allows for type 1 columns, so it more like Type 7, but we still call it Type 2
- Generation of surrogate keys
- Algorithmic surrogate keys
- Determination of dates for historical load. The dimension may need to be processed for multiple dates on model re-generation (on initial load or when the design changed) or to catch-up when several days are skipped for any reason
- Load dimension for multiple effective dates in bulk or one at a time.
- Setting keys on a fact table with the option to override the default implementation
- Full and incremental dimension load
Fact class:
- Assignment of foreign keys
- Load fact table from multiple sources
- Load multiple dates in bulk or one at a time.
- Full and incremental fact table load
- Supports different strategies of incremental fact table processing
- Creation of table with key setting columns to troubleshoot Unknown keys and setting keys in general
Configuration file
Configuration schema
Data Warehouse ETL Schema version 1
· Comparer configuration is described inside "dwEtl" object
· Configuration file can have additional attributes if needed at the root level in addition to " dwEtl" 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"
},
"dwEtl" :
{"jobType" : "LoadDimsFacts",
"isInitialLoad" : true,
"rerunEtlAfter" : "2021-04-28 16:33:00.119",
"fileEtlLog"
: {
"parquet"
: {},
"path"
: "C:\\ASCODE\\DATA\\NFL\\DW.ETL\\etlDwNFL.parquet\\"
},
"stgSources" :
[{ "moniker" : "PlayByPlay",
"description" : "Play by play s/sheet for all games",
"applicationSpecific" :
{"transactionName" : "play",
},"effectiveDateColumn" : "GameDate",
"isDefaultForEffectiveDate" : true,
"effectiveDateRule" : "ALL",
"timestampColumn" : "_SrcDt_RowTimestamp",
"fileSource" : { "path" : "C:\\ASCODE\\DATA\\NFL\\STAGING\\stgPbp.parquet", "parquet" : { } }
}, … ],"dimAuthority" :
{"packageName" : "com.dbtimes.nfldw.dimauth",
"dimensionNamePattern" : "(?<dimName>Dim[A-Z][a-zA-Z0-9]+)",
"fileDimensionsDestination" :
{"parquet" : {},
"path" : "C:\\DATA\\NFL\\DW\\DimNamePlaceholder.parquet",
"previousCopyPath" : "C:\\ASCODE\\DATA\\NFL\\DimAuthorityPrev\\DimNamePlaceholder.prev.parquet"
}
},"_comment__dimension_name " : "dimension name must match the dimensionNamePattern.It will be used to replace name placeholder with actual dimension name",
"dimensions" :
[ {"name" : "DimMyDimension",
"isLoad" : true,
"dimensionStgSources" :
[{ "moniker" : "someMoniker" },
… ],
"dimensionDimSources" :
[
{ "name"
: "DimDate",
"schema"
:
[
{ "colName"
: "Date" },
{ "colName" :
"DateDesc" },
{ "colName"
: "MonthName" }
]
}
],
"schema" :
[{ "colName" : "MyDimensionKey","colType" : "Long", "isKey" : true, "isSurrogateKey" : true, "unknownValue" : "0" },
{ "colName" : "MyDimensionNaturalKeyOne","colType" : "String", "isNaturalKey" : true, "unknownValue" : "Unknown" },
{ "colName" : "MyDimensionNaturalKeyTwo","colType" : "String", "isNaturalKey" : true, "unknownValue" : "Unknown" },
{ "colName" : "SomeStringDimensionAttr","colType" : "String", "isNaturalKey" : false, "unknownValue" : "Unknown" },
{ "colName" : "SomeIntDimensionAttr","colType" : "Integer", "unknownValue" : "-1" },
{ "colName" : "SomeDateDimensionAttr ","colType" : "Date", "unknownValue" : "1900-01-01" }
{ "colName" : "SomeType2Attr", "colType" : "String", "isTypeTwo" : true, "unknownValue" : "" },
] },
…
],
"dataMart" :
{
"packageName" :
"com.dbtimes.nfldw.datamartgames",
"factNamePattern" :
"(?<factName>Fact[A-Z][a-zA-Z0-9]+)",
"fileFactsDestination" :
{
"parquet" : {},
"path" : "C:\\DATA\\NFL\\DW\\FactNamePlaceholder.parquet",
"pathWithKeySetters" : "C:\\DATA\\NFL\\DW.ETL\\FactNamePlaceholder.WithKeySetters.parquet"
}
},
"facts" :
[
{ "name"
: "FactPlaysForMerge",
"isLoad" : false,
"processingMode" :
"MERGE_PARTITION",
"factStgSources" :
[
{ "moniker"
: "PlayByPlay" }
],
"schema" :
[
{ "colName"
: "SeasonYear",
"colType" : "Integer", "isForPartition"
: true },
{ "colName"
: "DataSourceMoniker", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceMoniker" },
{ "colName"
: "GameId", "colType" :
"String", "isMergeKey"
: true, "isForSettingForeignKey" : true, "foreignKey"
: "GameKey", "underlyingDim" :
"DimGame", "underlyingDimCol"
: "GameId" },
{ "colName" :
"PlayDate", "colType"
: "Date", "isMergeKey" :
true, "isForSettingForeignKey"
: true, "isEffDateForTypeTwo" : true, "foreignKey"
: "PlayDateKey",
"underlyingDim" : "DimDate", "underlyingDimCol"
: "Date" },
{ "colName"
: "OffenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "OffenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "DefenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DefenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "PenaltyTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PenaltyTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "TimeoutTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "TimeoutTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "Formation",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "Formation"
},
{ "colName"
: "PlayType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PlayType"
},
{ "colName"
: "PassType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PassType"
},
{ "colName"
: "PenaltyType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PenaltyType"
},
{ "colName"
: "ToGo", "colType" :
"Double", "isMeasure" :
true },
{ "colName"
: "PenaltyYards",
"colType" : "Double", "isMeasure" :
true }
]
},
…
]
}
}
Sample configuration file
{
"isDebug"
: false,
"sparkParams"
:
{
"spark.app.name"
: "Spark NFL DW"
},
"dwEtl"
:
{
"dwEtlConfigVersion": "1",
"jobType"
: "LoadDimsFacts",
"isInitialLoad"
: true,
"__optional
rerunEtlAfter" : "2021-04-28T16:33:00.119Z",
"fileEtlLog"
: {
"parquet": {},
"path": "/dw/data/etl/etldwnfl.parquet"
},
"__moniker"
: "moniker must be unique for each staging source, the
sources will be searched by their monikers",
"stgSources"
:
[
{ "moniker" :
"PlayByPlay",
"description" :
"Play by play s/sheet for all
games",
"applicationSpecific": {
"transactionName" : "play"
},
"effectiveDateColumn" : "GameDate",
"isDefaultForEffectiveDate" : true,
"timestampColumn" : "_SrcDt_RowTimestamp",
"effectiveDateRule" : "DISTINCT",
"fileSource" :
{ "path" : "/dw/data/staging/stgpbp.parquet", "parquet" :
{ } }
},
{ "moniker" :
"Teams",
"description" :
"Team names historically",
"fileSource" :
{ "path" : "/dw/data/staging/team.parquet", "parquet" :
{ } }
}
],
"dimAuthority"
:
{
"packageName" :
"com.dbtimes.nfldw.DataWarehouseNFL.DimAuthNFL",
"dimensionNamePattern" : "(?<dimName>Dim[A-Z][a-zA-Z0-9]+)",
"fileDimensionsDestination" :
{
"parquet" :
{},
"path" : "/dw/data/dm/___.parquet",
"previousCopyPath"
: "/dw/data/dmprev/___.prev.parquet"
}
},
"dimensions"
:
[
{
"name" : "DimDataSource",
"isLoad" : true,
"stgSourcesMonikers" :
[
],
"schema" :
[
{ "colName" :
"DataSourceKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"DataSourceMoniker","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"Unknown" },
{ "colName" :
"DataSourceDescription","colType"
: "String", "unknownValue" :
"Unknown" },
{ "colName" :
"TransactionName","colType"
: "String", "unknownValue" :
"Unknown" }
]
},
{
"name" : "DimDate",
"isLoad" : true,
"stgSourcesMonikers"
:
[
],
"schema" :
[
{ "colName" :
"DateKey","colType"
: "Integer", "isKey" : true, "isSurrogateKey" :
false, "unknownValue"
: "0" },
{ "colName" :
"Date","colType"
: "Date", "isNaturalKey" :
true, "unknownValue"
: "1900-01-01" },
{ "colName" :
"DateDesc","colType"
: "String", "unknownValue" :
"Unknown" },
{ "colName" :
"MonthName","colType"
: "String", "unknownValue" :
"Unknown" },
{ "colName" :
"Year","colType"
: "Integer", "unknownValue"
: "1900" },
{ "colName" :
"Quarter","colType"
: "Integer", "unknownValue" :
"0" },
{ "colName" :
"QuarterDesc","colType"
: "String", "unknownValue" :
"Unknown" },
{ "colName" :
"SeasonYear","colType"
: "Date", "unknownValue" :
"1900-01-01" },
{ "colName" :
"SeasonDescription","colType"
: "String", "unknownValue" :
"Unknown" }
]
},
{
"name" : "DimGame",
"isLoad" : true,
"stgSourcesMonikers" :
[
"PlayByPlay",
"Teams"
],
"schema" :
[
{ "colName" :
"GameKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"GameId","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"Unknown" },
{ "colName" :
"GameTeams","colType"
: "String", "unknownValue" :
"Unknown" }
]
},
{
"name" : "DimFieldPosition",
"isLoad" : true,
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"FieldPositionKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"Down","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"YardLine","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"RushDirection","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"YardLineFixed","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"YardLineDirection","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" }
]
},
{
"name" : "DimPlayType",
"isLoad" : true,
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"PlayTypeKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"Formation","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"PlayType","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"PassType","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"PenaltyType","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" }
]
},
{
"name" : "DimPlayTime",
"isLoad" : true,
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"PlayTimeKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"Quarter","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"Minute","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"Second","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" }
]
},
{
"name" : "DimPlayDetails",
"isLoad" : true,
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"PlayDetailsKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"IsRush","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsPass","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsIncomplete","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsTouchdown","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsSack","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsChallenge","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsChallengeReversed","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsMeasurement","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsInterception","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsFumble","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsPenalty","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsTwoPointConversion","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsTwoPointConversionSuccessful","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsPenaltyAccepted","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"IsNoPlay","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" }
]
},
{
"name" : "DimTeam",
"isLoad" : true,
"stgSourcesMonikers" :
[
"Teams"
],
"schema" :
[
{ "colName" :
"TeamKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"TeamOriginalName","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" },
{ "colName" :
"TeamName","colType"
: "String", "isNaturalKey"
: false },
{ "colName" :
"TeamAbbrevName","colType"
: "String" },
{ "colName" :
"TeamConference","colType"
: "String" },
{ "colName" :
"HistTeamName","colType"
: "String", "scdType" :
2 },
{ "colName" :
"HistTeamAbbrevName","colType"
: "String", "scdType" :
2 }
]
},
{
"name" : "DimTeamRole",
"__" : "This is static dimension. Its population is not needed in
incremental load",
"isLoad" : true,
"stgSourcesMonikers" :
[
],
"schema" :
[
{ "colName" :
"TeamRoleKey","colType"
: "Integer", "isKey" : true, "isSurrogateKey" :
false, "unknownValue"
: "0" },
{ "colName" :
"TeamRole","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"" }
]
}
],
"dataMart"
:
{
"packageName" :
"com.dbtimes.nfldw.DataWarehouseNFL.DataMartNFLGames",
"factNamePattern" : "(?<factName>(Fact|Bridge)[A-Z][a-zA-Z0-9]+)",
"fileFactsDestination" :
{
"parquet" :
{},
"path" : "/dw/data/dm/___.parquet",
"pathWithKeySetters" : "/dw/data/etl/___.withkeysetters.parquet"
}
},
"facts"
:
[
{ "name" : "FactPlays",
"isLoad" : true,
"processingMode" :
"REPLACE_PARTITION",
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"SeasonYear", "colType"
: "Integer", "isForPartition" :
true },
{ "colName" :
"DataSourceMoniker", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceMoniker" },
{ "colName" :
"GameId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "GameKey", "underlyingDim" :
"DimGame", "underlyingDimCol"
: "GameId" },
{ "colName" :
"PlayDate", "colType"
: "Date", "isForSettingForeignKey" : true, "isEffDateForTypeTwo"
: true, "foreignKey" :
"PlayDateKey", "underlyingDim"
: "DimDate", "underlyingDimCol" : "Date"
},
{ "colName" :
"OffenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "OffenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName" :
"DefenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DefenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName" :
"PenaltyTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PenaltyTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName" :
"TimeoutTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "TimeoutTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName" :
"Formation", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "Formation" },
{ "colName" :
"PlayType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PlayType" },
{ "colName" :
"PassType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PassType" },
{ "colName" :
"PenaltyType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PenaltyType" },
{ "colName" :
"Down", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "Down" },
{ "colName" :
"YardLine", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLine" },
{ "colName" :
"RushDirection", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "RushDirection" },
{ "colName" :
"YardLineFixed", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLineFixed" },
{ "colName" :
"YardLineDirection", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLineDirection" },
{ "colName" :
"Quarter", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Quarter" },
{ "colName" :
"Minute", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Minute" },
{ "colName" :
"Second", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Second" },
{ "colName" :
"IsRush", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsRush" },
{ "colName" :
"IsPass", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsPass" },
{ "colName" :
"IsIncomplete", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsIncomplete" },
{ "colName" :
"IsTouchdown", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsTouchdown" },
{ "colName" :
"IsSack", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsSack" },
{ "colName" :
"IsChallenge", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsChallenge" },
{ "colName" :
"IsChallengeReversed", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsChallengeReversed" },
{ "colName" :
"IsMeasurement", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsMeasurement" },
{ "colName" :
"IsInterception", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsInterception" },
{ "colName" :
"IsFumble", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsFumble" },
{ "colName" :
"IsPenalty", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsPenalty" },
{ "colName" :
"IsTwoPointConversion", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsTwoPointConversion" },
{ "colName" :
"IsTwoPointConversionSuccessful", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsTwoPointConversionSuccessful" },
{ "colName" :
"IsPenaltyAccepted", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayDetailsKey",
"underlyingDim" : "DimPlayDetails", "underlyingDimCol"
: "IsPenaltyAccepted" },
{ "colName"
: "IsNoPlay",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayDetailsKey", "underlyingDim"
: "DimPlayDetails",
"underlyingDimCol" : "IsNoPlay"
},
{ "colName" :
"Score", "colType"
: "Double", "isMeasure" :
true },
{ "colName" :
"Yards", "colType"
: "Double", "isMeasure"
: true },
{ "colName" :
"ToGo", "colType"
: "Double", "isMeasure"
: true },
{ "colName" :
"PenaltyYards", "colType"
: "Double", "isMeasure"
: true }
]
},
{ "name" : "FactPlaysForMerge",
"isLoad" : false,
"processingMode" :
"MERGE_PARTITION",
"stgSourcesMonikers":
[
"PlayByPlay"
],
"schema" :
[
{ "colName"
: "SeasonYear",
"colType" : "Integer", "isForPartition"
: true },
{ "colName"
: "DataSourceMoniker", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceMoniker" },
{ "colName"
: "GameId", "colType" :
"String", "isMergeKey"
: true, "isForSettingForeignKey" : true, "foreignKey"
: "GameKey", "underlyingDim" :
"DimGame", "underlyingDimCol"
: "GameId" },
{ "colName" :
"PlayDate", "colType"
: "Date", "isMergeKey" :
true, "isForSettingForeignKey"
: true, "isEffDateForTypeTwo" : true, "foreignKey"
: "PlayDateKey",
"underlyingDim" : "DimDate", "underlyingDimCol"
: "Date" },
{ "colName"
: "OffenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "OffenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "DefenseTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DefenseTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "PenaltyTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PenaltyTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "TimeoutTeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "TimeoutTeamKey",
"underlyingDim" : "DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName"
: "Formation",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "Formation"
},
{ "colName"
: "PlayType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PlayType"
},
{ "colName"
: "PassType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PassType"
},
{ "colName"
: "PenaltyType",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"PlayTypeKey", "underlyingDim"
: "DimPlayType",
"underlyingDimCol" : "PenaltyType"
},
{ "colName"
: "ToGo", "colType" :
"Double", "isMeasure" :
true },
{ "colName"
: "PenaltyYards",
"colType" : "Double", "isMeasure" :
true }
]
},
{ "name"
: "BridgeTeamsPlays",
"isLoad" : true,
"processingMode" :
"REPLACE",
"stgSourcesMonikers" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"DataSourceMoniker", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceMoniker" },
{ "colName" :
"GameId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "GameKey", "underlyingDim" :
"DimGame", "underlyingDimCol"
: "GameId" },
{ "colName" :
"PlayDate", "colType"
: "Date", "isForSettingForeignKey" : true, "isEffDateForTypeTwo"
: true, "foreignKey" :
"PlayDateKey", "underlyingDim"
: "DimDate", "underlyingDimCol" : "Date"
},
{ "colName" :
"TeamAbbrevName", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "TeamKey", "underlyingDim" :
"DimTeam", "underlyingDimCol"
: "HistTeamAbbrevName" },
{ "colName" :
"TeamRole","colType"
: "String", "isForSettingForeignKey"
: true, "foreignKey" :
"TeamRoleKey", "underlyingDim"
: "DimTeamRole",
"underlyingDimCol" : "TeamRole"
},
{ "colName" :
"Formation", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "Formation" },
{ "colName" :
"PlayType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PlayType" },
{ "colName" :
"PassType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PassType" },
{ "colName" :
"PenaltyType", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTypeKey",
"underlyingDim" : "DimPlayType", "underlyingDimCol"
: "PenaltyType" },
{ "colName" :
"Down", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "Down" },
{ "colName" :
"YardLine", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLine" },
{ "colName"
: "RushDirection",
"colType" : "String", "isForSettingForeignKey"
: true, "foreignKey" :
"FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "RushDirection" },
{ "colName" :
"YardLineFixed", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLineFixed" },
{ "colName" :
"YardLineDirection", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "FieldPositionKey", "underlyingDim"
: "DimFieldPosition", "underlyingDimCol"
: "YardLineDirection" },
{ "colName" :
"Quarter", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Quarter" },
{ "colName" :
"Minute", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Minute" },
{ "colName" :
"Second", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "PlayTimeKey",
"underlyingDim" : "DimPlayTime", "underlyingDimCol"
: "Second" }
]
}
]
}
}
Configuration file notes
Comment fields
"__moniker" in a sample configuration file is a comment. Any field that starts with two underscores is a comment and is ignored. A value for the comment field can be of any type.
Fact table schema
Schema for fact table "FactPlaysForMerge" in the sample configuration will result in fact table with following columns
· DataSourceKey (FK)
· GameKey (FK)
· PlayDateKey (FK)
· OffenseTeamKey (FK)
· DefenseTeamKey (FK)
· PenaltyTeamKey (FK)
· TimeoutTeamKey (FK)
· PlayTypeKey (FK) - In this example there are four columns in the schema needed to set foreign key PlayTypeKey.
· ToGo
· PenaltyYards
The fact able will be partitioned on SeasonYear. The fact directory structure
would look like this
To query partitioned Fact tables in, for example
Apache Drill, use syntax
select * from dfs.`fact-file-path\FactPlay.parquet`
where dir0=’SeasonYear=2016’ limit 5;
Object ModelObject
package com.dbtimes.dw.etl
object ModelObject
Methods
validateConfig
validateConfig(appConfig: Config): Unit - validates configuration against schema. Schema version to validate against is defined in "dwEtlConfigVersion" field in the configuration. All configuration errors must be fixed before proceeding with the ETL.
Arguments:
appConfig: com.typesafe.config.Config – configuration that has "dwEtl" as one of the fields. Configuration can include other fields – they will be ignored.
Returns:
UnitThrows 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
ModelObject.validateConfig(appConfig: Config)
runEtl
runEtl(appConfig: Config): Unit - execute the ETL process. The parameter holds the configuration that describes the dimensional model. Only one invocation of runEtl is allowed per application execution.
Arguments:
appConfig: com.typesafe.config.Config – configuration that has "dwEtl" as one of the fields. Configuration can include other fields – they will be ignored.
Returns:
UnitThrows an exception if validation failed
Example:
import org.apache.spark.sql.SparkSession
import com.typesafe.config.{Config, ConfigFactory}
import com.dbtimes.dw.etl.ModelObject
import com.typesafe.config.{ConfigFactory, Config}
def main(args: Array[String]) {
val
configFileName = if (args.length >= 1) args(0) else throw new RuntimeException("""ERROR:
Need to pass a parameter - configuration file name """)
// val appConfig = ConfigFactory.load(configFileName)
val appConfig = Utils.loadConfigAndApplyOverrides( configFileName, args.drop(1) )
// validate configuration
ModelObject.validateConfig(appConfig: Config)
val
sparkSessionBuilder = DwNFL.configureSparkSession(appConfig)
val
spark = sparkSessionBuilder.getOrCreate()
// ETL process to build or update dimensions and fact tables
ModelObject.runEtl(appConfig )
spark.stop()
}
private def configureSparkSession(appConfig: Config): SparkSession.Builder = {
val
sparkConfig = Utils.getConfigObjectFields( if ( appConfig.hasPath("sparkParams" )
) Some( appConfig.getConfig("sparkParams") ) else None )
val
conf = new SparkConf( )
.setAll(sparkConfig)
SparkSession.builder().config(conf)
}
Class Dim
package com.dbtimes.dw.etl
class Dim
In the most common scenario there are only two steps required to implement a new dimension:
1. Create a new class derived from Dim
a. The name of the class must match the dimension name from configuration file
2. Override loadDim method in dimension class.
Additional customization can be done by overriding appropriate methods of a Dim class.
Methods
dimKey
dimKey: String – dimension key name, can be used in optional override setForeignKeyOnFactTable for custom logic to set dimension key.
Arguments:
None
Returns:
String - dimension key and unknown value for dimension key, respectively
Example:
See
code example for dimKeyUnknownValue
dimKeyUnknownValue
dimKeyUnknownValue: String – unknown value for dimension key, populated from configuration dimension schema attribute "unknownValue" for dimension key. It can be used together with dimKey in optional override setForeignKeyOnFactTable for custom logic to set dimension key.
Arguments:
None
Returns:
String - dimension key and unknown value for dimension key, respectively
Example:
In
this SQL that is part of setForeignKeyOnFactTable
override implementation,
dimKeyUnknownValue is used to to set key value if there is no row in the
dimention to match fields in the fact table
// There is a
view created with the same name as dimension
// We use NULL safe join as if both columns are NULLs this
is a good key
// The code
assumes that there is only one key-value pair in the mapDimColsToFactCols
val sqlToSetKey =
s"""
|SELECT fact.*, IFNULL( dim.$dimKey, $dimKeyUnknownValue ) AS teamkey
|FROM $viewFactTable AS fact
|
LEFT OUTER JOIN $dimName AS dim ON
fact.${mapDimColsToFactCols.head._2} <=> dim.${mapDimColsToFactCols.head._1}
| AND fact.${effDateColAsOption.get} BETWEEN dim.StartDate AND dim.EndDate
|""".stripMargin
isInitialLoad
isInitialLoad: Boolean - This flag is populated from configuration path "dwEtl.isInitialLoad". Flag indicates that dimension created from scratch – initial load - or is being updated with new data. The value can be used in loadDim implementation to skip certain sources on initial load.
Arguments:
None
Returns:
Boolean – true on initial laod, and false otherwise.
datesToProcessView
datesToProcessView: String - The name of the view with effective dates. The view holds dates for the current run of the ETL process. The view has one column of DATE type with the name effDateColumnNameInDatesToProcess
Arguments:
None
Returns:
String – the view name
Example:
See code example for effDateColumnNameInDatesToProcess.
effDateColumnNameInDatesToProcess
effDateColumnNameInDatesToProcess: String - The name of the date column in the view datesToProcessView.
Arguments:
None
Returns:
String –the name of the date column in datesToProcessView
Example:
In
this example the date dimension is loaded with all dates at once by selecting
dates from datesToProcessView.
package com.dbtimes.nfldw.dimauth
import com.dbtimes.dw.etl.Dim
import com.dbtimes.nfldw.DwNFL
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.SparkSession
class DimDate(dimName: String) extends Dim(dimName) {
override
def loadDim(stgSrcViewWithMonikerName: String): Option[DataFrame] = {
val
sqlStgSource
=
s"""| SELECT
| CAST( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyyMMdd') AS INT) AS DateKey,
| $effDateColumnNameInDatesToProcess AS Date,
| DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy-MM-dd') AS DateDesc,
| DATE_FORMAT($effDateColumnNameInDatesToProcess, 'MMM') AS MonthName,
| CAST( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy') AS INT) AS Year,
| QUARTER($effDateColumnNameInDatesToProcess) AS Quarter,
| CONCAT( 'Q', CAST( QUARTER($effDateColumnNameInDatesToProcess) AS STRING ) ) AS QuarterDesc,
| CAST( IF( MONTH( $effDateColumnNameInDatesToProcess ) >= 8,
| CONCAT( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy'), '-01-01' ),
| CONCAT( CAST( YEAR($effDateColumnNameInDatesToProcess) - 1 AS STRING), '-01-01' ) ) AS DATE ) AS SeasonYear,
| IF( MONTH( $effDateColumnNameInDatesToProcess ) >= 8,
| CONCAT( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy'), '-', CAST( YEAR($effDateColumnNameInDatesToProcess) + 1 AS STRING ) ),
| CONCAT( CAST( YEAR($effDateColumnNameInDatesToProcess) - 1 AS STRING), '-', DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy') ) ) AS SeasonDescription
| FROM $datesToProcessView
""".stripMargin
val
spark = SparkSession.builder().getOrCreate()
val
dfStgSource = spark.sql(sqlStgSource)
Some(dfStgSource)
}
} getLastProcessedStgSourceTimestamp
getLastProcessedStgSourceTimestamp(stgSourceMoniker: String): Option[SqlTimestamp] - The method returns the maximum timestamp for the staging data source that was processed in the previous ETL run for the current dimension. It is used as optimization to load data into dimensional model object incrementally.
Arguments:
stgSourceMoniker: String – staging source moniker
Returns:
Option[SqlTimestamp] – the timestamp of the last loaded row in the previous ETL process for the given source and given dimension
Example:
See
example for isStgSourceChangedSinceLastLoad.
isStgSourceChangedSinceLastLoad
isStgSourceChangedSinceLastLoad(stgSourceMoniker: String): Boolean - The method returns false if the data source has not changed since was processed in the previous ETL run for the current dimension, and true otherwise. The method is used to load data into dimensional model object incrementally.
Arguments:
stgSourceMoniker: String – staging source moniker
Returns:
String – the view name and th name of the date column in that view, respectively
Example:
In
this example isStgSourceChangedSinceLastLoad is used to skip the load if the source
data did not change and getLastProcessedStgSourceTimestamp is used to limit the amount of data to
process.
override def loadDim(stgSrcViewWithMonikerName:
String): Option[DataFrame] = {
if
(
!isStgSourceChangedSinceLastLoad(stgSrcViewWithMonikerName) ) {
None
}
else
{
val
srcLastTimestamp = getLastProcessedStgSourceTimestamp( stgSrcViewWithMonikerName
)
val
srcLastTimestampAsString = if( srcLastTimestamp.isDefined) srcLastTimestamp.get.toString else ""
val
sqlStgSource = if (stgSrcViewWithMonikerName == "PlayByPlay")
{
s"""|SELECT
| ...
|FROM $stgSrcViewWithMonikerName
|WHERE _SrcDt_EffectiveDate =
'2099-0202'
|
${if (srcLastTimestampAsString.isEmpty) "" else "AND _SrcDt_RowTimestamp > '" + srcLastTimestampAsString + "'" } """.stripMargin
}
else
throw new RuntimeException("""Dim
Etl ERROR: Unknown stg source while loading dimension """)
val spark =
SparkSession.builder().getOrCreate()
val
dfStgSource = spark.sql(sqlStgSource)
Some(dfStgSource)
}
}
Required Overrides
loadDim
Each dimension implementation must override loadDim method to load dimension’s data from each staging source. There are two signatures available for this method – one with effective date and another without - and concrete dimension implementation must override one of them.
This method is called in the loop for each staging source defined in "stgSources" array. In addition for the second override signature, the method will be called for each effective date
The temporary view with the name of stgSrcViewWithMonikerName is available for SQL, so the
logic can be implemented to have a staging source specific code.
signature 1:
protected def loadDim(stgSrcViewWithMonikerName: String): Option[DataFrame]
Override this method to load new data in bulk or for all effective dates.
parameter:
stgSrcViewWithMonikerName: String – the value of staging source moniker
signature 2:
protected def loadDim(effDateYYYY_MM_DD: String, stgSrcViewWithMonikerName: String): Option[DataFrame]
Override this method to load new data for each effective date. This version of loadDim must be used to load a dimension with Type 2 columns.
parameters:
effDateYYYY_MM_DD: String – effective date in the format YYYY-MM-DD, e.g. 2020-11-11
stgSrcViewWithMonikerName: String – the value of staging source moniker
returns:
Option[DataFrame] – DataFrame with data from respective data source. The field names must match dimension schema and the field types must be compatible with field types defined in the schema. One exception is surrogate dimension key – it must not be included in the result.
Example:
In this example are shown two versions of loadDim implementation Date dimension. Only one of those implementations must be present in actual implementation.
effDateColumnNameInDatesToProcess and datesToProcessView are methods descrivbed above.
override def loadDim(stgSrcViewWithMonikerName:
String): Option[DataFrame] = {
val
sqlStgSource
=
s"""|
SELECT
| CAST( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyyMMdd') AS INT) AS DateKey,
| $effDateColumnNameInDatesToProcess AS Date,
| DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy-MM-dd') AS DateDesc,
| DATE_FORMAT($effDateColumnNameInDatesToProcess, 'MMM') AS MonthName,
| CAST( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy') AS INT) AS Year,
| QUARTER($effDateColumnNameInDatesToProcess) AS Quarter,
| CONCAT( 'Q', CAST( QUARTER($effDateColumnNameInDatesToProcess) AS STRING ) ) AS QuarterDesc,
| CAST( IF( MONTH( $effDateColumnNameInDatesToProcess ) >= 8,
| CONCAT( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy'), '-01-01' ),
| CONCAT( CAST( YEAR($effDateColumnNameInDatesToProcess) - 1 AS STRING), '-01-01' ) ) AS DATE ) AS SeasonYear,
| IF(
MONTH( $effDateColumnNameInDatesToProcess ) >= 8,
| CONCAT( DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy'), '-', CAST( YEAR($effDateColumnNameInDatesToProcess) + 1 AS STRING ) ),
| CONCAT( CAST( YEAR($effDateColumnNameInDatesToProcess) - 1 AS STRING), '-', DATE_FORMAT($effDateColumnNameInDatesToProcess, 'yyyy') ) ) AS SeasonDescription
| FROM $datesToProcessView
""".stripMargin
val
spark = SparkSession.builder().getOrCreate()
val
dfStgSource = spark.sql(sqlStgSource)
Some(dfStgSource)
}
override def loadDim(effDateYYYY_MM_DD: String, stgSrcViewWithMonikerName:
String): Option[DataFrame] = {
val
sqlStgSource
=
s"""|
SELECT
| CAST( DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyyMMdd') AS INT) AS DateKey,
| CAST( '$effDateYYYY_MM_DD' AS DATE) AS Date,
| DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyy-MM-dd') AS DateDesc,
| DATE_FORMAT('$effDateYYYY_MM_DD', 'MMM') AS MonthName,
| CAST( DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyy') AS INT) AS Year,
| QUARTER('$effDateYYYY_MM_DD') AS Quarter,
| CONCAT( 'Q', CAST( QUARTER('$effDateYYYY_MM_DD') AS STRING ) ) AS QuarterDesc,
| CAST( IF( MONTH( '$effDateYYYY_MM_DD' ) >= 8,
| CONCAT( DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyy'), '-01-01' ),
| CONCAT( CAST( YEAR('$effDateYYYY_MM_DD') - 1 AS STRING), '-01-01' ) ) AS DATE ) AS SeasonYear,
| IF( MONTH( '$effDateYYYY_MM_DD' ) >= 8,
| CONCAT( DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyy'), '-', CAST( YEAR('$effDateYYYY_MM_DD') + 1 AS STRING ) ),
| CONCAT( CAST( YEAR('$effDateYYYY_MM_DD') - 1 AS STRING), '-', DATE_FORMAT('$effDateYYYY_MM_DD', 'yyyy') ) ) AS SeasonDescription
""".stripMargin
val
spark = SparkSession.builder().getOrCreate()
val
dfStgSource = spark.sql(sqlStgSource)
if
(DwNFL.getIsDebug) {
dfStgSource.show(10);
dfStgSource.printSchema()
}
Some(dfStgSource)
}
Optional Overrides
enrichDim
If implemented, it will be called after all the sources have been loaded using loadDim(stgSrcViewWithMonikerName: String) or all sources loaded for one effective date using loadDim(effDateYYYY_MM_DD: String, stgSrcViewWithMonikerName: String). Implement this override to perform transformation on the source data before it is added to the dimension. The returned DataFrame has to have the same schema as the one returned from loadDim method.
Can be used, for example, to calculate/recalculate some fields from dimension sources that can only be done once all sources are loaded.
signature:
protected
def enrichDim(dfSrcDim:
DataFrame):
DataFrame
parameter:
dfSrcDim: DataFrame – dimension DataFrame with data for all sources or all sources for one effective date depending on one or two parametrs loadDim override
returns:
Option[DataFrame] – DataFrame with the same structure as DataFrame passed as parameter.
setForeignKeyOnFactTable
This method must be overridden if the logic of setting the key is anything other than straight equi-join on all natural keys. The new key will have to be appended to the existing fact Data Frame. For some dimensions the key must be set in a cascading fashion based on some precedence.The steps for this in the override are following:· in step 1 create new dataFrame with the key either set using some first rule or NULL· in the next step create a new DF using the df from step 1 as input where the new key is NULL and replace the last column with the non-NULL key based on the logic for this step. The resulting dataFrame may still have some keys NULL
· continue with more steps like above· in the last step repeat the same logic using the df from the previous step as input, only in this case set the key to Unknown.
· union all dataFrames from all steps where the key is not NULL· in the final dataFrame returned should have one more column with a new key with all keys non-NULL
signature:
protected def setForeignKeyOnFactTable(fact: DataFrame, mapDimColsToFactCols: Map[String, String], effDateColAsOption: Option[String]): DataFrame
parameters:
fact: DataFrame - the data frame to set the key on.
mapDimColsToFactCols: Map[String, String]-
this map includes only columns explicitly defined in fact schema. That is this
map does not include the Start/End dates for type 2 dimension, the CompoundKey
column, or the key itself. For this Map
the order in not important as the columns can be listed in the join in any
order, so do not need ListMap for it.
effDateColAsOption: Option[String] - effective date will be used for type 2 dimensions. The name of the new column the the one from the schema
returns:
Option[DataFrame] – DataFrame with one more column – a dimension key - than the DataFrame passed as parameter.
preProcess
It is called before sources are loaded and can be used to create dimension specific views that are later used in the dimension load process.
signature:
protected def preProcess(stgSrcViewsWithMonikerNames: List[String], datesToProcess: List[Date]): Unit
parameters:
stgSrcViewsWithMonikerNames: List[String] - a list of all staging sources.
datesToProcess: List[Date] - list of dates that should be processes in this load.
returns:
Unit
postProcess
It is called after dimension processing is complete right before the dimension is persisted into a parquet file. One possible use is to do some logging or for debugging. signature:
protected def postProcess(stgSrcViewsWithMonikerNames: List[String], datesToProcess: List[Date], dfModelObject: DataFrame): DataFrame
parameters:
stgSrcViewsWithMonikerNames: List[String] - a list of all staging sources.
datesToProcess: List[Date] - list of dates that should be processes in this load.
dfModelObject: DataFrame – current version of a dimension
returns:
DataFrame – new version of a dimension or the same as was passed as a parameter
getCustomDatesToProcess
It is called after effective dates are determined from the sources based on configuration. In this method these dates can be modified or completely replaced based on custom logic. signature:
protected def getCustomDatesToProcess(dfDatesToProcessBasedOnConfig: DataFrame): Option[DataFrame]
parameter:
dfDatesToProcessBasedOnConfig: DataFrame – current
version of dates to process
returns:
Option[DataFrame] – new version with dates to proess. Return None if dates are not overwritten
Example:
override def getCustomDatesToProcess(dfDatesToProcessBasedOnConfig:
DataFrame): Option[DataFrame] = {
// For DimTeans we need one date per season
dfDatesToProcessBasedOnConfig.createOrReplaceTempView("DatesToProcessBasedOnConfig")
if
(DwNFL.getIsDebug) {
dfDatesToProcessBasedOnConfig.show(50);
dfDatesToProcessBasedOnConfig.printSchema()
}
val
sqlDimDatesToProcess
=
s"""|SELECT
MAX( datesAll.$effDateColumnNameInDatesToProcess ) AS SeasonDate
|FROM Teams
| INNER JOIN DatesToProcessBasedOnConfig AS
datesAll ON Teams.SeasonYear = udfSeasonYear( datesAll.$effDateColumnNameInDatesToProcess )
|GROUP BY Teams.SeasonYear
|
""".stripMargin
val
spark = SparkSession.builder().getOrCreate()
val
dfDimDatesToProcess = spark.sql(sqlDimDatesToProcess)
if
(DwNFL.getIsDebug) {
dfDimDatesToProcess.show(5);
dfDimDatesToProcess.printSchema()
}
Some(dfDimDatesToProcess)
}
Class Fact
package com.dbtimes.dw.etl
class Fact
In the most common scenario there are only two steps required to implement a new fact table:
1. Create a new class derived from Fact
2. Override loadFact method
Additional customization can be done by overriding appropriate methods of a Fact class.
Methods
isInitialLoad
Same as Dim.isInitialLoad
datesToProcessView
Same as Dim.datesToProcessView
effDateColumnNameInDatesToProcess
Same as Dim.effDateColumnNameInDatesToProcess
getLastProcessedStgSourceTimestamp
Same as Dim.getLastProcessedStgSourceTimestamp
isStgSourceChangedSinceLastLoad
Same as Dim.isStgSourceChangedSinceLastLoad
Required Overrides
loadFact
Each fact or fact table implementation must override loadFact method to load fact table’s data from each staging source. There are two signatures available for this method – one with effective date and another without - and concrete fact table implementation must override one of them.
This method is called in the loop for each staging source defined in "stgSources" array. In addition, for the second override signature, the method will be called for each effective date
A
temporary view with the name of stgSrcViewWithMonikerName is available for SQL, so the
logic can be implemented to have a staging source specific code.
signature 1:
protected def loadFact(stgSrcViewWithMonikerName: String): Option[DataFrame]
Override this method to load new data in bulk or for all effective dates.
parameter:
stgSrcViewWithMonikerName: String – the value of staging source moniker
signature 2:
protected def loadFact(effDateYYYY_MM_DD: String, stgSrcViewWithMonikerName: String): Option[DataFrame]
Override this method to load new data for each effective date. This version of loadFact must be used to load a dimension with Type 2 columns.
parameters:
effDateYYYY_MM_DD: String – effective date in the format YYYY-MM-DD, e.g. 2020-11-11
stgSrcViewWithMonikerName: String – the value of staging source moniker
returns:
Option[DataFrame] – DataFrame with data from respective data source. The field names must match fact schema and the field types must be compatible with field types defined in the schema.
Example:
class FactPlays(factName: String) extends FactsNFL(factName)
{
override def loadFact(stgSrcViewWithMonikerName:
String): Option[DataFrame] = {
if (
!isStgSourceChangedSinceLastLoad(stgSrcViewWithMonikerName) ) {
dwEtlLog.info(s"--
Skipping loading fact $factName from source $stgSrcViewWithMonikerName as is has not changed
since last load")
None
}
else {
val dfFact = if (stgSrcViewWithMonikerName
== "PlayByPlay") {
val sqlStgSource =
s"""
|SELECT
|
src.SeasonYear AS SeasonYear,
| "$stgSrcViewWithMonikerName" AS DataSourceMoniker,
| CAST(
src.GameId AS STRING ) AS
GameId,
| CAST(
src.GameDate AS DATE ) AS
PlayDate,
| CAST(
src.PenaltyYards AS DOUBLE ) AS PenaltyYards
|FROM $stgSrcViewWithMonikerName AS src
| INNER JOIN $viewWithDatesToProcess AS dates ON src.SeasonYear = udfSeasonYear( dates.$effDateColumnNameInDatesToProcess )
|
| """.stripMargin
dwEtlLog.info(s"--
Loading fact $factName from source $stgSrcViewWithMonikerName using sql:\n" + sqlStgSource)
val dfStgSource = spark.sql(sqlStgSource)
dfStgSource.dropDuplicates()
}
else {
throw new RuntimeException(s"""Etl
ERROR: Unknown stg source while loading fact $factName""")
}
if (DwNFL.getIsDebug)
{
dfFact.printSchema()
dfFact.show(40)
}
Some(dfFact)
}
}
}
Optional Overrides
preProcess
Same as Dim.preProcess only for fact table.
postProcess
Same as Dim.postProcess only for fact table.
getCustomDatesToProcess
Same as Dim.getCustomDatesToProcess.
Sample
data
Dimension DimFieldPosition
Fact table FactPlays
