def clean_data(**kwargs):
global pg_hook
ti = kwargs["ti"]
new_id = ti.xcom_pull(task_ids="hello_task_01")
# Load data from the raw_data column, it's only 1 value
pg_command = """SELECT raw_data FROM dag_dag WHERE id = %s"""
data = pg_hook.get_records(pg_command, parameters=[new_id])[0][0]
# clean the data for the Meteor visualisation
data = {"number of labs": len(data)}
# Transform the dict into a string for PostgreSQL
data = json.dumps(data)
# Save the data
pg_command = """UPDATE dag_dag SET clean_data = %s WHERE id = %s"""
pg_hook.run(pg_command, parameters=[data, new_id])
return "Data prepared for the visualisation successfully."
# Setup the DAG
#
# schedule_interval uses the cron format
#
# * * * * * *
# | | | | | |
# | | | | | +-- Year (range: 1900-3000)
# | | | | +---- Day of the Week (range: 1-7, 1 standing for Monday)
# | | | +------ Month of the Year (range: 1-12)
# | | +-------- Day of the Month (range: 1-31)
# | +---------- Hour (range: 0-23)
# +------------ Minute (range: 0-59)
#
# See more: http://www.nncron.ru/help/EN/working/cron-format.htm
#
# Or datetime.timedelta
# See more: https://docs.python.org/2/library/datetime.html#datetime.timedelta
#
评论列表
文章目录
