I've performed a short study of our late events, which I detect according to the timestamp of the file created by Gobblin.

The test code: https://gitlab.wikimedia.org/-/snippets/149

I've isolated the optimization here https://phabricator.wikimedia.org/T371803

We have prepared some work to Refine raw events directly into Iceberg tables.

As discussed in the meeting, I'm OK with minimal changes to ESC: It could be nice to add the hourly_computing_scale and have a boolean ~needs_to_be_refined.

To check the output of the new Refine process, I've been working with diff from https://github.com/G-Research/spark-extension/ to verify that my output matches what was generated by the old Refine process. Here are the problems I've encountered and circumvented:

• The user_agent structure depends on the existing table schema. So when migrating, we need to recreate the table, not from EvolveHiveTable, but from the original schema with show create table. https://github.com/wikimedia/analytics-refinery-source/blob/master/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/refine/TransformFunctions.scala#L341
• The _schema column is not filled with the current Refine process. It is pleasantly filled with the new code. The workaround is to remove _schema from the comparison. But I can't find where it's filled in the new code.
• The geocoded_data is different between the two processes. The geocoded_data.subdivision_code has been recently added. Also, the geocoding library is updated, and some small differences in the generated data exist. The workaround is to remove geocoded_data in the comparison.
• The deduplicate transform function runs on a DF that was not previously sorted. As Spark picks only the first duplicated lines (in each group), it picks them randomly. https://github.com/wikimedia/analytics-refinery-source/blob/master/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/refine/TransformFunctions.scala#L112 The workaround is to remove in the comparison of the whole meta (keeping only id & stream) The duplications I've seen mostly involved identical lines differing only with meta.dt and meta.request_id.

Following the discussion on Slack, I've created the table and changed the dir owner to analytics-product. The long-term discussion about HDSF dir architecture and ownership is here: T367243

We have been working with @amastilovic on a Wikitech page to describe the reasoning behind choosing Airflow itself to store Dataset status.

Moreover, one more conf for a dag to execute in a depth-first manner is to add in its default_args (or in some tasks args) weight_rule="upstream".

I've encountered an issue with our current production Airflow setup where the scheduler is not executing tasks in a depth-first manner as expected. Upon investigation, I found that the depth-first execution feature isn't supported in our current Airflow version (2.7.3). This functionality was introduced in Airflow 2.8.0, as per this pull request: Apache Airflow PR #27827 (I've tried it locally with success).

https://docs.sqlfluff.com/en/stable/dialects.html#hive

5 datasets are being refined as a POC on the prod cluster. 2 on the test cluster.

The Airflow PR has been merged and should be released in Airflow 2.9 in April.

Done. The last version was done with Blubber:

• https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/602
• T351792

Some research:

• Each XML dumps snapshot may represent ~5.5TB (including ~1.8TB for wikidata and 1.4TB for enwiki)
• The Airflow sensor may take ~19days to turn green. It waits until the last dump has been processed (_IMPORTED flag). Most dumps are generated in a matter of days (~4 on average, maybe). Enwiki may take 7 days. And they all wait for the wikidata dump (~19 days).
• When the sensor turns green, a heavy Spark job is launched to convert all the compressed XML to parquet. ~5.5TB (compressed) is taking ~4.5 days to process.
• The perceived gaps are due to the non-parallelism of the dag + very long jobs. 1 heavy job is preventing the other ones from running. due to the retries (Thx for the pointer @JAllemandou ). Other symptoms, same problem here I think: https://phabricator.wikimedia.org/T342911

Linked with: https://phabricator.wikimedia.org/T351792

wmf.wikidata_item_page_link/snapshot=2024-01-15 looked fine from a row count perspective but 2024-01-22 snapshot was unexpectedly available and had zero rows:

select count(*) from wmf.wikidata_item_page_link where snapshot='2024-01-22';
...
+------+
| _c0  |
+------+
| 0    |
+------+

I've added the missing partitions in the source of wikidata_item_page_link and its missing snapshot is now generated.

https://docs.google.com/document/d/1upAje5lMawu4X6seRxI8Lx7YN-oHEzcEcm2fO6E5OH0/edit

I would like to add rewrite_manifests to the list of maintenance actions:

• rewrite_data_files
• expire snapshots
• rewrite_manifests https://iceberg.apache.org/docs/latest/spark-procedures/#rewrite_manifests

@BTullis , @brouberol , @Stevemunene I would like your feedback on this subject:

Following the Grafana dashboard review, I've performed some changes to it:

• I distributed the graphs into 4 sections: Failures, Durations, Counts, Scheduling
• I added missing parameterization of variables (e.g the list of operators changes when the instance is selected)
• I updated TTLs into statsd-exporter in order to reflect into Prometheus the rate of the metrics sent by Airflow (e.g. when Airflow emits a metric to say 1 task is in failure, it generates 1 call. and we would like to get this single point into Prometheus)
• I isolated task failures into its own graph

1/ Splitting the CI

In this ticket, I needed to find a way to detect when an Iceberg table has some data in it. This would replace the Hive partition sensor when migrating a table to Iceberg.
We chose to launch a Spark application running an SQL count. It's now implemented here:

• https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/564
• https://gerrit.wikimedia.org/r/c/analytics/refinery/+/983673

A drawback of this solution is that it generates a FAILED Spark application each time the sensor does not find any data in the interval. When monitoring our Spark applications, we want to avoid artificially growing the FAILED counts.

https://github.com/wikimedia/operations-puppet/blob/f7c3eb56a9417571792b7636367f3c13e850bc83/modules/profile/manifests/airflow.pp#L199

I think it's because the airflow-analytics jobs check should be run like the other commands: with a custom PYTHONPATH=/path/to/root/of/airflow-dags/ as an env variable.

I have the first version of the code in review.

In this puppet patch, we are adding configuration to send more Airflow metrics to Prometheus, and to customize them.

The workaround to our Gilab-CI pb is here: https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/537

The puppet configuration is now merged, and statsd_exporter is running on an-test-client1002. Analytics Prometheus is scrapping from it, as it should.