Databricks Configuration Framework
Project description
dbxconfig
Configuration framework for databricks pipelines. Define configuration and table dependencies in yaml config then get the table mappings config model:
Define your tables.
landing:
landing_dbx_patterns:
customer_details_1: null
customer_details_2: null
raw:
raw_dbx_patterns:
customers:
ids: id
depends_on:
- landing.landing_dbx_patterns.customer_details_1
- landing.landing_dbx_patterns.customer_details_2
base:
base_dbx_patterns:
customer_details_1:
ids: id
depends_on:
- raw.raw_dbx_patterns.customers
customer_details_2:
ids: id
depends_on:
- raw.raw_dbx_patterns.customers
Define you load configuration:
tables: ./test/Config/tables.yaml
landing:
trigger: customerdetailscomplete-{{filename_date_format}}*.flg
trigger_type: file
database: landing_dbx_patterns
table: "{{table}}"
container: datalake
root: "/mnt/{{container}}/data/landing/dbx_patterns/{{table}}/{{path_date_format}}"
filename: "{{table}}-{{filename_date_format}}*.csv"
filename_date_format: "%Y%m%d"
path_date_format: "%Y%m%d"
format: cloudFiles
spark_schema: ./test/Schema/{{table.lower()}}.yaml
options:
# autoloader
cloudFiles.format: csv
cloudFiles.schemaLocation: /mnt/{{container}}/checkpoint/{{checkpoint}}
cloudFiles.useIncrementalListing: auto
# schema
inferSchema: false
enforceSchema: true
columnNameOfCorruptRecord: _corrupt_record
# csv
header: false
mode: PERMISSIVE
encoding: windows-1252
delimiter: ","
escape: '"'
nullValue: ""
quote: '"'
emptyValue: ""
raw:
database: raw_dbx_patterns
table: "{{table}}"
container: datalake
root: /mnt/{{container}}/data/raw
path: "{{database}}/{{table}}"
options:
checkpointLocation: /mnt/{{container}}/checkpoint/{{database}}_{{table}}
mergeSchema: true
Import the config objects into you pipeline:
from dbxconfig import Config, Timeslice, StageType
# build path to configuration file
pattern = "auto_load_schema"
config_path = f"./Config/{pattern}.yaml"
# create a timeslice object for slice loading. Use * for all time (supports hrs, mins, seconds and sub-second).
timeslice = Timeslice(day="*", month="*", year="*")
# parse and create a config objects
config = Config(timeslice=timeslice, config_path=config_path)
# get the configuration for a table mapping to load.
table_mapping = config.get_table_mapping(
timeslice=timeslice,
stage=StageType.raw,
table="customers"
)
Development Setup
pip install -r requirements.txt
Unit Tests
To run the unit tests with a coverage report.
pip install -e .
pytest test/unit --junitxml=junit/test-results.xml --cov=dbxconfig --cov-report=xml --cov-report=html
Build
python setup.py sdist bdist_wheel
Publish
twine upload dist/*
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
dbxconfig-1.0.9.tar.gz
(10.4 kB
view hashes)
Built Distribution
dbxconfig-1.0.9-py3-none-any.whl
(13.1 kB
view hashes)
Close
Hashes for dbxconfig-1.0.9-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cd9f5a90404f60d57e7a6e4eca654005e8fd121e479a69ae759ac43d051a711d |
|
MD5 | 8f960fa304d94e72dd0752a5ef9b629b |
|
BLAKE2b-256 | a0e4c8d47eee15cf4c24c0cab0f4b8a46434dc12662d309f1f9e62f93d1f90f5 |