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 the ETL process for a dimensional model. The dimensional model design should follow the Kimball methodology (see The Data Warehouse Toolkit by Kimball/Ross).
The library implements the ETL process for the dimensional model based on the configuration and on specific code needed for each dimension and fact table.
The librarys Dim and Fact classes generically handle various aspects of dimension design and lifecycle.
Dim class:
- Loading a dimension from multiple sources
- Historical load of data on initial load for both Type 1 and Type 2 dimensions
- Re-loading a dimension from a specific point in time
- Type 1 and Type 2 changes, and by extension Type 3 and Type 4. A Type 2 dimension allows for Type 1 columns, so it is 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 changes) or to catch up when several days are skipped for any reason.
- Loading a 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
- Loading a fact table from multiple sources
- Loading multiple dates in bulk or one at a time
- Full and incremental fact table load
- Support for different strategies of incremental fact table processing
- Creation of a table with key-setting columns to troubleshoot Unknown keys and key setting in general
Configuration file
Configuration schema
Actual - dwetl-config-schema.json
As a document - Data Warehouse ETL
Schema
Configuration file
at a glance
This configuration shows a compressed version, leaving out some options. Refer to the Configuration schema for the complete definition.
· DW ETL configuration is described inside "dwEtl" object
· Configuration file can have additional attributes if needed at the root level in addition to " dwEtl" 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"},
"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" :
[{ "id" : "PlayByPlay",
"description" : "Play by play s/sheet for all games",
"applicationSpecific" :
{"transactionName" : "play",
},"effectiveDateColumn" : "GameDate",
"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" :
[{ "id" : "someId" },
],
"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", "scdType": 2, "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"
:
[
{ "id" : "PlayByPlay" }
],
"schema"
:
[
{ "colName" :
"SeasonYear", "colType"
: "Integer", "isForPartition" :
true },
{ "colName" :
"DataSourceId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceId" },
{ "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"
:
{
"jobType"
: "LoadDimsFacts",
"isInitialLoad"
: true,
"__optional
rerunEtlAfter" : "2021-04-28T16:33:00.119Z",
"fileEtlLog"
: {
"parquet": {},
"path": "/dw/data/etl/etldwnfl.parquet"
},
"__id"
: "id must be unique for each staging source, the sources
will be searched by their ids",
"stgSources"
:
[
{ "id" : "PlayByPlay",
"description" :
"Play by play s/sheet for all
games",
"applicationSpecific": {
"transactionName" : "play"
},
"effectiveDateColumn" : "GameDate",
"timestampColumn" : "_SrcDt_RowTimestamp",
"effectiveDateRule" : "DISTINCT",
"fileSource" :
{ "path" : "/dw/data/staging/stgpbp.parquet", "parquet" :
{ } }
},
{ "id" : "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,
"stgSourcesIds" :
[
],
"schema" :
[
{ "colName" :
"DataSourceKey","colType"
: "Long", "isKey" : true, "isSurrogateKey" :
true, "unknownValue"
: "0" },
{ "colName" :
"DataSourceId","colType"
: "String", "isNaturalKey"
: true, "unknownValue" :
"Unknown" },
{ "colName" :
"DataSourceDescription","colType"
: "String", "unknownValue" :
"Unknown" },
{ "colName" :
"TransactionName","colType"
: "String", "unknownValue" :
"Unknown" }
]
},
{
"name" : "DimDate",
"isLoad" : true,
"stgSourcesIds" :
[
],
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
"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,
"stgSourcesIds" :
[
],
"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",
"stgSourcesIds" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"SeasonYear", "colType"
: "Integer", "isForPartition" :
true },
{ "colName" :
"DataSourceId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceId" },
{ "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",
"stgSourcesIds":
[
"PlayByPlay"
],
"schema"
:
[
{ "colName" :
"SeasonYear", "colType"
: "Integer", "isForPartition" :
true },
{ "colName" :
"DataSourceId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceId" },
{ "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",
"stgSourcesIds" :
[
"PlayByPlay"
],
"schema" :
[
{ "colName" :
"DataSourceId", "colType"
: "String", "isForSettingForeignKey" : true, "foreignKey"
: "DataSourceKey",
"underlyingDim" : "DimDataSource", "underlyingDimCol"
: "DataSourceId" },
{ "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
"__id" in the 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 a fact table with the 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 the foreign key PlayTypeKey.
· ToGo
· PenaltyYards
The fact table will be partitioned by SeasonYear. The fact directory structure
would look like this
To query partitioned fact tables in, for
example, Apache Drill, use the following 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 the configuration against the schema.
Arguments:
appConfig: com.typesafe.config.Config the configuration that has "dwEtl" 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
ModelObject.validateConfig(appConfig:
Config)
runEtl
runEtl(appConfig: Config): Unit executes 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 the configuration that has "dwEtl" as one of its fields. The configuration can include other fields; they will be ignored.
Returns:
UnitThrows an exception if validation fails.
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, only two steps are 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 the configuration file
2. Override the loadDim method in the dimension class.
Additional customization can be done by overriding appropriate methods of the Dim class.
Methods
dimKey
dimKey: String the dimension key name; it can be used in the optional override setForeignKeyOnFactTable for custom logic to set the dimension key.
Arguments:
None
Returns:
String the dimension key name
Example:
See
code example for dimKeyUnknownValue
dimKeyUnknownValue
dimKeyUnknownValue: String the unknown value for the dimension key, populated from the dimension schema attribute "unknownValue" in the configuration. It can be used together with dimKey in the optional override setForeignKeyOnFactTable for custom logic to set the dimension key.
Arguments:
None
Returns:
String - the unknown value for the dimension key
Example:
In
this SQL, which is part of the
setForeignKeyOnFactTable override
implementation, dimKeyUnknownValue is used to set the key value if there is no row
in the dimension that matches the fields in the fact table
// There is a
view created with the same name as the dimension
// We use a NULL-safe join because if both columns are NULL, 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". The flag indicates whether the dimension is being created from scratch (initial load) or is being updated with new data. The value can be used in a loadDim implementation to skip certain sources on initial load.
Arguments:
None
Returns:
Boolean true on initial load, 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(stgSrcView: 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(stgSourceId: 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 an optimization to load data into the dimensional model object incrementally.
Arguments:
stgSourceId: String staging source id
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(stgSourceId: String): Boolean - The method returns false if the data source has not changed since it 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:
stgSourceId: String staging source id
Returns:
Boolean false if the data source has not changed since it was processed in the previous ETL run for the current dimension, and true otherwise
Example:
In
this example, isStgSourceChangedSinceLastLoad is used to skip the load if the source
data has not changed, and getLastProcessedStgSourceTimestamp is used to limit the amount of data to
process.
override def loadDim(stgSrcView: String): Option[DataFrame] = {
if
(
!isStgSourceChangedSinceLastLoad(stgSrcView) ) {
None
}
else
{
val
srcLastTimestamp = getLastProcessedStgSourceTimestamp( stgSrcView )
val
srcLastTimestampAsString = if( srcLastTimestamp.isDefined) srcLastTimestamp.get.toString else ""
val
sqlStgSource = if (stgSrcView == "PlayByPlay") {
s"""|SELECT
| ...
|FROM $stgSrcView
|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 the loadDim method to load the dimensions data from each staging source. There are two signatures available for this method one with an effective date and one without and a concrete dimension implementation must override one of them.
This method is called in a loop for each staging source defined in the "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 stgSrcView is available for SQL, so the logic can be
implemented to have staging-source-specific code.
signature
1:
protected def loadDim(stgSrcView: String): Option[DataFrame]
Override this method to load new data in bulk or for all effective dates.
parameter: stgSrcView: String the view with the same name as staging source id
signature 2:
protected def loadDim(effDateYYYY_MM_DD: String, stgSrcView: 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
stgSrcView: String the view with the same name as staging source id
returns: Option[DataFrame] DataFrame with data from respective data source. The field names must match the dimension schema, and the field types must be compatible with the field types defined in the schema. One exception is the surrogate dimension key it must not be included in the result.
Example:
This example shows two versions of loadDim implementation for the Date dimension. Only one of them must be present in an actual implementation.
effDateColumnNameInDatesToProcess and datesToProcessView are methods described above.
override def loadDim(stgSrcView: 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, stgSrcView: 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(stgSrcView: String) or all sources loaded for one effective date using loadDim(effDateYYYY_MM_DD: String, stgSrcView: String). Implement this override to perform transformations 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 the loadDim method.
Can be used, for example, to calculate or recalculate some fields from dimension sources, which 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 whether the loadDim override takes one or two parameters
returns:
DataFrame a DataFrame with the same structure as DataFrame passed as a parameter.
setForeignKeyOnFactTable
This method must be overridden if the logic of setting the key is anything other than a straight equi-join on all natural keys. The new key will have to be appended to the existing fact DataFrame. For some dimensions the key must be set in a cascading fashion based on some precedence.The steps for this in the override are the following:· in step 1, create a new DataFrame with the key either set using some first rule or NULL;· in the next step, create a new DataFrame using the DataFrame 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 NULL keys;· continue with more steps like the above;· in the last step, repeat the same logic using the DataFrame 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;· 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 DataFrame to set the key on.
mapDimColsToFactCols: Map[String,
String]-
this map includes only columns explicitly defined in the fact schema. That is,
this map does not include the Start/End dates for a Type 2 dimension, the
CompoundKey column, or the key itself.
For this Map, the order is not important, as the columns can be listed
in the join in any order, so a ListMap is not needed.
effDateColAsOption: Option[String] - the effective date will be used for Type 2 dimensions. The name of the new column is the one from the schema
returns: DataFrame DataFrame with one more column a dimension key - than the DataFrame passed as a 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(stgSrcViews: List[String], datesToProcess: List[Date]): Unit
parameters:
stgSrcViews: List[String] - a list of views for all staging sources. Each view has
the same name as the staging source id
datesToProcess: List[Date] - a list of dates that should be processed 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 for logging or debugging. signature:
protected def postProcess(stgSrcViews: List[String], datesToProcess: List[Date], dfModelObject: DataFrame): DataFrame
parameters:
stgSrcViews: List[String] - a list of views for all staging sources. Each view
has the same name as the staging source id
datesToProcess: List[Date] - a list of dates that should be processed 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 process. 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, only two steps are required to implement a new fact table:
1. Create a new class derived from Fact
2. Override the loadFact method
Additional customization can be done by overriding appropriate methods of the 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 the loadFact method to load the fact tables data from each staging source. There are two signatures available for this method one with an effective date and one without and a concrete fact table implementation must override one of them.
This method is called in a loop for each staging source defined in the "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 stgSrcView is available for SQL, so the logic can be
implemented to have staging-source-specific code.
signature 1:
protected def loadFact(stgSrcView: String): Option[DataFrame]
Override this method to load new data in bulk or for all effective dates.
parameter: stgSrcView: String the view with the same name as staging source id
signature 2:
protected def loadFact(effDateYYYY_MM_DD: String, stgSrcView: String): Option[DataFrame]
Override this method to load new data for each effective date. This version of loadFact must be used to load a fact table that uses dimensions with Type 2 columns.
parameters:effDateYYYY_MM_DD: String effective date in the format YYYY-MM-DD, e.g. 2020-11-11
stgSrcView: String the view with the same name as staging source id
returns: Option[DataFrame] DataFrame with data from respective data source. The field names must match the fact schema, and the field types must be compatible with the field types defined in the schema.
Example:
class FactPlays(factName: String) extends FactsNFL(factName)
{
override def loadFact(stgSrcView: String): Option[DataFrame] = {
if (
!isStgSourceChangedSinceLastLoad(stgSrcView) ) {
dwEtlLog.info(s"--
Skipping loading fact $factName from source $stgSrcView as is has not changed
since last load")
None
}
else {
val dfFact = if (stgSrcView == "PlayByPlay") {
val sqlStgSource =
s"""
|SELECT
| src.SeasonYear AS SeasonYear,
|
"$stgSrcView" AS DataSourceId,
| CAST( src.GameId AS STRING
) AS GameId,
| CAST( src.GameDate AS DATE
) AS PlayDate,
| CAST( src.PenaltyYards AS DOUBLE ) AS PenaltyYards
|FROM $stgSrcView AS src
| INNER JOIN $viewWithDatesToProcess AS dates ON
src.SeasonYear = udfSeasonYear( dates.$effDateColumnNameInDatesToProcess )
|
| """.stripMargin
dwEtlLog.info(s"--
Loading fact $factName from source $stgSrcView 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, but for a fact table.
postProcess
Same as Dim.postProcess, but for a fact table.
getCustomDatesToProcess
Same as Dim.getCustomDatesToProcess.
Sample
data
Dimension DimFieldPosition
Fact table FactPlays
