💨🥫 A Data Factory system for running data processing pipelines built on AirFlow and tailored to CKAN. Includes evolution of DataPusher and Xloader for loading data to DataStore.
Home Page: https://tech.datopian.com/flows/
License: Other
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
Using the same structure of the multiple node DAG assuming local files, create a DAG that handles resources on the cloud (CSV and JSON).
DAG tasks:
ckannext-aircan (connector) tasks:
- [x] 1. create endpoint to receive Airflow response after processing
- [x] 2. Handle Airflow response
- [ ] 3. If success, download processed json file from bucket
NOTE: This is not the strategy. We will send the processed JSON via API.
After this long task is complete, we still need to:
- [ ] Handle errors This will be in the next milestone
- [ ] Handle absence of a response from Airflow This will be in the next milestone
- [ ] Delete remote file (create a separate DAG for that) This will be in the next milestone
Refactor libraries to receives only the env data they need; not the entire config
Why not pass these arguments explicitly to the function rather than pull from config
Originally posted by @rufuspollock in #16
After #8 , the directory structure of the project has changed and the import on the DAGs have broken imports.
Relates to #30
Load data into CKAN DataStore using Airflow as the runner. Replacement for DataPusher and Xloader. Clean separation of components do you can reuse what you want (e.g. don't use Airflow but use your own runner)
datapusher_submitcurl ...) with ability to configure both ckan instance and src csv fileDeploy to GCP
dependencies folder on GCP bucket; this second method is easier for development purposes only)Auth and logging
CKAN integration
The logs for the last task are not appearing even though the DAG succeeds
Note The docs for DX are still under development and the steps here may need adjustments/clarification
Following https://gitlab.com/datopian/tech/devops/-/blob/master/dx-rfc-dx-approach.md and https://gitlab.com/datopian/tech/devops/-/issues/163, @Irio helped us to come up with the following list:
gcloud auth login
gcloud container clusters get-credentials
argocd proj create [project name] pointing to remote argo cd -- created project aircanaircan-devBefore deploying, use Terraform (repo dx-terraform) to create a new PSQL instance -- @Irio we need to add documentation about this step.
As an aircan user, I want to separate the data directory from the dags directory so that import of dags will be lightweight and data can be accessed outside the code.
Decided to delete the sample files from Dag.
Introducing a status API providing reporting on run status, error and logs.
When I run a DAG in AirCan I want to
so that I can report on this to users and empower them to resolve errors
/api/3/action/aircan_submit?dag_id=... that runs a DAG and returns the run ID/api/3/action/aircan_status?run_id=... which reports on status of a run e.g. PENDING | RUNNING | PAUSED | FAILED | SUCCESS and provides error informationFUTURE
Thinking of user using a CKAN instance. A run of a DAG is triggered by the CKAN instance.
The user knows the name of the DAG they'll trigger (atm specified in an .env var; it can change later)
They'd then access the following endpoint to get the status of the DAG_run
GET http://ckan:500/api/3/action/dag_run/<dag_id> # returns all recent runs of that DAG
GET http://ckan:500/api/3/action/dag_run/<dag_id>/<run_id>
They'd see a page with the execution dates for the dag_id
Response from Airflow:
[
{
"dag_id": "ckan_api_load_gcp",
"dag_run_url": "/admin/airflow/graph?dag_id=ckan_api_load_gcp&execution_date=2020-07-09+13%3A21%3A56%2B00%3A00",
"execution_date": "2020-07-09T13:21:56+00:00",
"id": 64,
"run_id": "manual__2020-07-09T13:21:56+00:00",
"start_date": "2020-07-09T13:21:56.963772+00:00",
"state": "success"
},
{
"dag_id": "ckan_api_load_gcp",
"dag_run_url": "/admin/airflow/graph?dag_id=ckan_api_load_gcp&execution_date=2020-07-16+13%3A07%3A02%2B00%3A00",
"execution_date": "2020-07-16T13:07:02+00:00",
"id": 65,
"run_id": "manual__2020-07-16T13:07:02+00:00",
"start_date": "2020-07-16T13:07:02.100794+00:00",
"state": "failed"
},
]
The flow we'd need
on CKAN you hit:
POST api/3/aircan_submit?dag_id=XXX&...
As the answer of this request, you must get back the run_id
What do you do with this run ID? [For now we can assume the client keeps that run id and it's up to them. Longer term we will have a "Run/Job" objects in CKAN] We'd need to persist it on a DB... Otherwise it'll be lost
Our customized response including access to GCP logs
Response:
{ airflow_response: {
"dag_id": "ckan_api_load_gcp",
"dag_run_url": "/admin/airflow/graph?dag_id=ckan_api_load_gcp&execution_date=2020-07-09+13%3A21%3A56%2B00%3A00",
"execution_date": "2020-07-09T13:21:56+00:00",
"id": 64,
"run_id": "manual__2020-07-09T13:21:56+00:00",
"start_date": "2020-07-09T13:21:56.963772+00:00",
"state": "success"
}
gcp_logs: {
logs for that particular run_id
}
},
{
"dag_id": "ckan_api_load_gcp",
"dag_run_url": "/admin/airflow/graph?dag_id=ckan_api_load_gcp&execution_date=2020-07-16+13%3A07%3A02%2B00%3A00",
"execution_date": "2020-07-16T13:07:02+00:00",
"id": 65,
"run_id": "manual__2020-07-16T13:07:02+00:00",
"start_date": "2020-07-16T13:07:02.100794+00:00",
"state": "failed"
},
They'd get the result of the Airflow API for DAG status
https://airflow.apache.org/docs/stable/rest-api-ref.html
Ideally combined with GCP logs
FAQs
Another path to consider (or support both): having an endpoint set up on airflow ready to receive a post from AirCan.
i.e. a task fails while running on a DAG. Aircan sends a notification by hitting an endpoint on CKAN.
kwargs = {
resoure: ...
}
dagid = dag_run(**kwargs)
running = True
while(running) {
status = dag_status(dagid)
{
running: ...
}
}
Questions Errors Handling
Logs
Other questions
What are the endpoints (on CKAN) that will trigger the DAG? right now we have datastore_create and aircan_submit. Are there any other triggering endpoints?
What is the best way to organize the docs? I fing the README on aircan is extensive and potentially with lots of non-useful information. i.e. are people going to use aircan on standalone?
When user uploads a file with tabular data to portal (CKAN), we want to guess columns names properly, so it would be passed as part of payload to a DAG.
schema_fields_array payload parameter(currently hard coded).Add tabular data file (currently CSV) guessing columns names mechanism.
Populate schema_fields_array payload parameter using guessing columns names mechanism above.
Write tests for checking functionality.
Add DataStore loading via API to library and build a DAG that loads a (configurable) file to (configurable) CKAN instance (either local and/or demo.ckan) using API only.
load_csv_via_api method (inserted on #17), and combined with create_datastore_table does the entire flow via APIapi_load (leave old datastore-load-via-db-copy aside)
params dictionary of the DAG as shown in #18**CKAN_SITE_URL), as well as API Key (with CKAN_SYSADMIN_API_KEY)load.py to have an agnostic configload_csv with API approach (similar to datapusher) #17resurce_id after create and pass it to load #17test trigger works) #18Logging and reporting is a crucial aspect of a data factory system like this.
When a Run is initiated by an Operator they want to see it is running and be notified of application and (meta)data errors as soon as possible, especially “halts” so that they can debug and re-run
If there are a lot of (data) errors I want to examine these in a system that allows me to analyse / view easily (i.e. don’t have my page crash as it tries to load 100k error messages)
I don’t want to receive 100k error emails …
When a scheduled Run happens as an Operator (Sysadmin), I want to be notified afterwards (with a report?) if something went wrong, so that I can do something about it …
When I need to report to my colleagues about the Harvesting system I want to get an overall report of how it is going, how many datasets are harvested etc so that I can tell them
Status info: this is Run is running, it is finished, it took this long …
(Raw) Log information …
Reports / Summaries e.g. 200 records processed, 5 errors, 2 warnings, 8 new dataset, 192 existing records updated
Build a DAG that loads a file to CKAN ( remote demo.ckan) using DB connection
...
/lib/ folderdelete_datastore_table <<
create_datastore_table <<
load_csv_to_postgres_via_copy <<
restore_indexes_and_set_datastore_activeWe want to allow authenticated users only on the airflow GCP webserver.
Discuss other authentication methods rather than basic auth. By setting the security configuration, it is possible to allow only specific users, or groups of users, etc. Much stronger than basic HTTP Auth. In this version, we'll use Google Auth, which requires a token
Our DAGs require parameters and it looks like Google Cloud Composer does not support them with its interface. We'll have to investigate possible ways to send parameters to our DAG (CKAN resource ID, input csv, output JSON)
Replace params with kwargs as suggested in https://guptakumartanuj.wordpress.com/2018/01/21/passing-and-accessing-run-time-arguments-to-dag-airflow-through-cli/
Bug with double quotes when calling a Jinja template - Fixed on #36
Change .cfg file to override params
Docs with commands explaining how to trigger a DAG on Composer - See #38
When I clone/use this repository, I want to be able to know the steps to take in order to install requirements and run the tests so I can contribute back to the code more easily and ensure that high code coverage is maintained during the whole life of this project.
make test command.Get latest changes from PR #11 and build on top of that.
Create Makefile with commands requirements and test.
Add .coveragerc configuration file.
Add .isort.cfg configuration file.
Add pytest.ini configuration file.
Add .flake8 configuration file.
Fix directory structure after merging #13. -- DONE: See PR #27
Set up GitHub Actions. -- DONE: See PR #24
Set up Code Climate and add Not doing for now..codeclimate.yml configuration file.
Set up Travis CI and add Switching to GitHub Actions instead..travis.yml configuration file.
This mostly involves replicating the structure of the repo for Giftless as it follows best practices and use similar tools, including the test suite with pytest.
Data Dictionary and other details in the resourceAssuming a local CKAN instance and a local Airflow instance:
Note: This task won't provide any integration with GCP logs (see #69 ); it only collects the output of Airflow API
Packing the code into a PyPI package is necessary to have Aircan running into a cloud provider (GCP).
aircan-lib deployed and possible to install via requirements.txtAfter #8 and all the changes on the new DAGs, it is necessary to update the README file to reflect the most recent status of the project.
When user uploads a file with tabular data to portal (CKAN), we want an Airflow DAG to be triggered so the data is uploaded to Datastore.
A new ckan extension exists similar to ckanext-datapusher that does the following (TODO: check behaviour of ckanext-datapusher in some detail):
curl -X POST \
-H 'Authorization: YOUR_CKAN_API_KEY' \
-d '{"resource_id": "RESOURCE_ID"}' \
http://CKAN_HOST/api/action/datapusher_submit
with successful response:
{
"help": "http://ckan:5000/api/3/action/help_show?name=datapusher_submit",
"success": true,
"result": {
"message": "Created <DagRun ckan_api_load_multiple_steps @ 2020-07-06 07:27:15+00:00: manual__2020-07-06T07:27:15+00:00, externally triggered: True>",
"execution_date": "2020-07-06T07:27:15+00:00",
"run_id": "manual__2020-07-06T07:27:15+00:00"
}
}
push notification mechanism to notify the loader (Airflow DAG) to upload data to DataStore, likeload_csv_to_postgres_via_copy, we'll probably need a refactor, since it returns a generic errorexcept Exception as e:
return {
'success': False,
'message': 'Error during deleting index: {}'.format(e)
}
local airflow with load code for local csv converting to json. Part of Epic #1
This the uber-epic for the complete evolution of CKAN DataStore load to AirCan.
npm test. Right now they are pointing to a temporary CKAN instance (not to DX) and they run the entire flow. No CI for this on github atmrun_id which you can pass in to the DAG and which it uses in logging etc when running it so we can reliably track logs etc. Also move airflow status info into logs (so we don't depend on AirFlow API).
aircan_status(run_id) function that can be turned into an API in CKAN (or elsewhere)FUTURE after this
graph TD
v1[v0.1 CSV load working, CI/CD setup with rich tests]
v2[v0.2 errors, logging and UI integration]
v3[UI integration]
v3[v0.3 expand the tasks and e.g. xlsx, google sheets loading]
v4[v0.4 harvesting ...]
v1 --> v2
v2 --> v3
v3 --> v4
graph TD
deploytotest[Deploy DAGs to test GCC]
deploydags[Deploy DAGs into this AirFlow<br/>starting with CKAN data load]
deploygcc[Deploy Airflow<br/>i.e. Google Cloud Composer]
nhsdag[NHS DAG for loading to bigquery]
nhs[NHS Done: instance updated<br/>with extension and working in production]
logging[Logging]
reporting[Reporting]
othersite["Other Site Done"]
start[Start] --> deploygcc
start --> logging
multinodedag --> deploytotest
subgraph General Dev of AirCan
errors[Error Handling]
aircanlib[AirCan lib refactoring]
multinodedag[Multi Node DAG]
logging --> reporting
end
subgraph Deploy into Datopian Cluster
deploytotest[Deploy DAGs to test GCC] --> deploydags
deploygcc --> deploydags
end
subgraph CKAN Integration
setschema[Set Schema from Resource]
endckan[End CKAN work]
setschema --> endckan
end
deploydags --> nhsdag
deploydags --> othersite
endckan --> nhs
subgraph NHS
nhsdag --> nhs
end
classDef done fill:#21bf73,stroke:#333,stroke-width:1px;
classDef nearlydone fill:lightgreen,stroke:#333,stroke-width:1px;
classDef inprogress fill:orange,stroke:#333,stroke-width:1px;
classDef next fill:lightblue,stroke:#333,stroke-width:1px;
class multinodedag done;
class versioning nearlydone;
class setschema,errors,deploydags,nhsdag,deploygcc inprogress;
With the new schema changes (https://github.com/datopian/ckanext-aircan/blob/0ba67147cc0fccfca5e32d5ff16ac16f49541ca9/ckanext/aircan_connector/action.py#L28)
the following error is coming.
Traceback (most recent call last)
File "/usr/local/lib/airflow/airflow/models/__init__.py", line 1491, in _run_raw_tas
result = task_copy.execute(context=context
File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 112, in execut
return_value = self.execute_callable(
File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 117, in execute_callabl
return self.python_callable(*self.op_args, **self.op_kwargs
File "/home/airflow/gcs/dags/aircan/dags/api_ckan_load_gcp.py", line 77, in task_create_datastore_tabl
schema = eval_schema.get('fields'
AttributeError: 'str' object has no attribute 'get
The output should be the same as the one in #10, but we would have a DAG with 4 nodes, all PythonOperators. It would be more Airflow-oriented and more suitable for retriggering retires and dealing with errors
Build a DAG that loads a file to CKAN ( remote demo.ckan) using API only
Check https://gitlab.com/datopian/tech/data-loader-next-gen/-/blob/master/data_loader/load.py
Copy over relevant parts (+ tests) into local /lib/ folder
Create appropriate airflow tasks + DAG using this lib methods: ~rufus debatable if this is worth it
Is there any retry policy? ~rufus on what conditions. Prob not for now
The task load_csv_to_postgres_via_copy considers these two fields when trying to fit the CSV into the table. Find a solution that allows a clean CSV load
Do a trial deployment of aircan to GCP cloud composer
Extras
"conf": {
resource: { // a f11s data resource object
"path": ckan_resource['url'],
"format": ckan_resource['format'],
"ckan_resource_id": ckan_resource['id'],
"schema": ckan_resource['schema']
}
"ckan_config": {
"api_key": ..
}
}~Note: Airflow does not support nested JSON; we'll have to send the info in a one-level json ~
Note2: When using JINJA templates, it doesn't. When fetching params from **context, it works, so we're adopting the original proposal
"conf": {
"path": ckan_resource['url'],
"format": ckan_resource['format'],
"ckan_resource_id": ckan_resource['id'],
"schema": ckan_resource['schema']
"ckan_api_key": ..
}
**context as it's only argIf the JSON file already exists, the create action fails, making the DAG fail. Create unique names (based on timestamps) on the JSON creation task
On the body of the request, send
"schema": {
"fields": [
{
"name": "FID",
"type": "integer",
"format": "default"
},
{
"name": "Mkt-RF",
"type": "number",
"format": "default"
}
]
}At this time our internal libraries handle errors and responses this way, for example return {"success": False, "response": response.json()}.
Ideally, we should create a proper class for exceptions, errors handling and responses, unifying the way the system behaves (in library level)
When I work on AirCan as a developer, I want to be able to run the test suite without making network requests and using services that can be mocked instead so that I can test more effectively without depending on external resources.
lib/load.py are mocked in the test suite and can be run independently from external resources.delete_datastore_table.create_datastore_table.load_csv_to_postgres_via_copy.delete_index.restore_indexes_and_set_datastore_active._generate_index_name.lib/api_ckan_load.py and examples/csv_to_json.py is also useful for now.load.py are using the CKAN API. Comment present:
This should work over API (not inside CKAN process) or do direct
lib/api_ckan_load.py: https://github.com/ckan/ckan/tree/master/ckan/testsmake test should run clean and all tests related to API integration should pass.After the refactor from lib to dependencies, some tests have imports outdated .
PR #51 adds some fixes; we still need to
make testWhen a user manually triggering an import of an existing file via API in GCP the result is always null.
Sample API call:
curl -X POST \
-H 'Authorization: YOUR_CKAN_API_KEY' \
-d '{"resource_id": "RESOURCE_ID"}' \
http://CKAN_HOST/api/action/datapusher_submit
In case we are running in GCP env (CKAN__AIRFLOW__CLOUD=GCP), the DAG is triggered and
the response is:
{
"help": "http://ckan:5000/api/3/action/help_show?name=datapusher_submit",
"success": true,
"result": null
}
On the other hand, running the same API call in local env (CKAN__AIRFLOW__CLOUD=LOCAL),
we are getting a proper response:
{
"help": "http://ckan:5000/api/3/action/help_show?name=datapusher_submit",
"success": true,
"result": {
"message": "Created <DagRun ckan_api_load_multiple_steps @ 2020-07-06 07:27:15+00:00: manual__2020-07-06T07:27:15+00:00, externally triggered: True>",
"execution_date": "2020-07-06T07:27:15+00:00",
"run_id": "manual__2020-07-06T07:27:15+00:00"
}
}
invoke_gcpAt this time they are sent via array in the DAG params (e.g. "schema_fields_array": "['field1', 'field2']") and everything is treated as text type.
What are good alternatives?
Questions:
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.