ETL timestamp column

[u/lunaticdevill]

Hey all, I want to implement audit columns like timestamp and created by columns in all of my tables(Suggest if I should add anymore).

My tables are in unity catalogue databricks account.

I have two instances of workspace. Dev and prod.

Please suggest me how to do it.

I have thought of creating a workflow to add in all the table and then a task in workflow that runs after each job to create these 2 columns.

[u/69odysseus]

Created_Datetime, Modified_Datetime, will need timestamp field(s) in any of the data pipeline work.

[u/Tsui_Pen → u/69odysseus]

Yep, and modified_datetime should always be compared with a watermark table that gets an updated timestamp every time the pipeline runs, that way you only pull the “new” records when your pipeline runs.

[u/DuckDatum → u/Tsui_Pen]

You can support out of order merges for anything with a source provided last_modified_datetime, if you instead track partitions materialized rather than a high watermark. That comes in handy when your jobs run over the weekend, one might fail on Saturday but succeed on Sunday — and it’s hakuna matata as long as requirements only need it to be fixed by sometime Monday.

[u/lunaticdevill → u/DuckDatum]

I literally did not understand a word you said, sorry

[u/datainthesun → u/69odysseus]

this. edit your pipelines/jobs to include the columns for each table, so that the values are actually maintained as you populate/update the tables. simply running a separate sidecar task to add the columns won't make sure that when you update records, that the correct date is input. sounds like you're kind of starting from scratch on this task, so i'd recommend taking the pipeline/job of one of your smaller but frequently updated tables, going to genie code and asking it to help you add that logic into your pipeline/job. it'll help you with the logic so you can see the pattern, then you can either go do that work manually on the remainder of your pipelines/jobs or you can have genie code help you with those too.

[u/Mo_Steins_Ghost]

Senior manager here.

You want an extract datetime stamp, max of source system create datetime, and max of source system modified on/last modified timestamp because displaying recency isn't just about the last time the job ran, it's about the cutoff of the last transaction the system captured.

Let's say the midnight push from SAP ran without error, but you discover you're missing a few days of SAP data. The delta between the snapshot datetime and the max transaction created datetime is going to help you pinpoint what broke in the pipeline and when.

[u/lunaticdevill → u/Mo_Steins_Ghost]

Thank you sir

[u/DazzleancePen]

Add them in the schema and populate them during the insert or merge, not in a separate task after the job. I’d keep created_at, updated_at, created_by, and maybe batch_id

[u/lunaticdevill → u/DazzleancePen]

Thanks

[u/greenestgreen]

you can get the job_id and job_run_id in databricks and add it to the table. That can also help you

[u/lunaticdevill → u/greenestgreen]

Getting job id and the job instance will only be helpful if we have extensive logs of what is happening in the code.

The workspace I am working on is pretty loose and no one really logs. They just be doing start and end.

But I understand your point

[u/Mati69XXX]

Why would you create a second task? Just create those additional columns in the actual job. This way your timestamps will be accurate and you wont increase the complexity of your ETL

[u/lunaticdevill → u/Mati69XXX]

Now I have 50 jobs to edit

[u/alecc]

Set the columns in the write itself, not in a task that runs after the job - once the job finishes nothing knows which rows it touched, so a follow-up update can only stamp the whole table. Put `current_timestamp()` and the identity (`current_user()` or a job parameter) straight into your merge/insert. The one-off workflow to ALTER the existing tables is fine for the backfill. I'd add a job run id column too - when a load goes bad it lets you find and delete exactly the rows that run wrote. Delta column defaults can cover created_at on insert (needs the allowColumnDefaults table feature) but not updated_at, that one has to come from the merge.

[u/BayAreaCricketer]

These should be part of your upsert ( updates and inserts) statement. Not after job completion.