How it Works:
For each new automated dataset, we build ETL pipelines from modules that define how the data is accessed, extracted and loaded, as well as a schema which defines the structure of the data and any pre-processing rules that need to be applied. This way, instead of writing the code for each part from scratch, we can simply select the modules we need, provide them a few parameters, and schedule our new pipeline to run when necessary.Connectors:
As implied by their name, connectors define how the pipeline will connect to the source data. This could be for remote files accessed over HTML, SFTP or FTP, or even something as simple as a local file residing on the same machine as the running pipeline.Extractors:
Extractors define how our pipeline will extract the connected data into a format that can eventually be loaded into the Data Center. They are built to work with various source file types. These could be something as simple as a CSV file, something a bit more complex like an Excel file, or even something as involved as performing OCR (optical character recognition) on PDF files. Currently we only have modules to support CSV and Excel files, but as publishers provide different source file types, we’ll be able to adapt by creating new extractors. We use Python library called marshmallow as part of our extraction process to handle serialization/deserialization and validate our extracted data.Schemas:
In order for marshmallow to know how to validate our data, we provide a marshmallow schema for each pipeline. The schema defines what data type each field should be as well what pre-processing and validation methods, if any, need to be run during extraction.Loaders:
Finally, the loaders take the extracted and validated data and load it into the specified dataset in the Data Center. For now, our loaders all work by using the Data Center’s API.Benefits:
As mentioned earlier, one of the key benefits of developing this new system was the ability to increase our code reuse and cut down on code rewriting. We also developed the new processes following a test-driven development paradigm, writing unit tests for each small part of the process, which has significantly reduced the time needed to test individual jobs as they’re added. Combined, these two aspects of the new system have led to big decreases in per-job implementation time; a single job that used to take hours to code and test now only takes several minutes to an hour.In addition to saving time, the new processes also makes it easier to validate data and ensure that it conforms to certain standards before uploading it to the dataset. This makes up for a shortcoming in CKAN’s default data-loading processes in which it attempts to “guess” the datatypes of fields in a dataset. This “guessing” is not always accurate and can lead to situations where CKAN halts the loading of data mid-stream because of type mismatches. We’re now able to avoid this altogether by explicitly declaring datatypes in our schemas, and ensuring that each record conforms to that schema (fixing it if necessary) before uploading.Going Forward:
As of writing this, four of the eight datasets that have automated loading processes are handled by the new pipeline and we have plans to migrate the rest. We also intend to run every upcoming automated dataset through the new process as well, allowing us to handle an increasing number datasets and increase the rate at which we can onboard new publishers. We’ll also look into other ETL tools such as FME and Kettle, but our code seems to be working well for us, and we haven’t yet needed to have non-developers involved in running our ETL processes.In the meantime, feel free to browse the code, fork it, and even make a pull request if you’ve added something you think would be helpful. We look forward to continuing the development of our ETL pipeline, and hope that it might also prove useful to others.
