Yay!

We spent some time discussing this with @GGoncalves-WMF and arrived at an idea we were both happy with:

• Models should run in both incremental and non-incremental fashion
• For each model to be processed, airflow generates 3 tasks in a task-group:
  1. Decision task querying the metastore on whether one the model tables exist or not (goes to either 2 or 3 below)
  2. If at least one the model tables don't exist, run the non-incremental task associated with the model
  3. If all of the model tables exist, run the incremental task associated with the model

I'm in agreement with the almost every comment @xcollazo has made above.
The only point I'd like to discuss is data deletion vs iceberg maintenance only.
It's pretty sure there will be cases where we'll want data-deletion in addition to iceberg maintenance, and the other way around too.
Also, the data-deletion case brings in the time-column to be used for deletion (named partition-column parsing in the above).
I can't think of a better solution than what has been described above, with adaptations as Xabriel commented.

I have reduced the replication factor for the netflow dataset to 1 instead of 2. This should help while we (hopefully) fix the problem with an-druid1003.
Also, it seems that @brouberol 's idea of adding idle nodes to the cluster is worth it!

I like that! We could even think of an API layer to dowload the files directly from CephS3, bypassing the copy altogether (but this is very much scope creep :)

In T297231#12374446, @xcollazo wrote:

and insert them into an iceberg table.

Can we please estimate how big this table will be? Could this perhaps be the very first Ceph backed Iceberg table? 😉

One of the key decision regarding moving this forward is whether or not we wish the new data lake to

• Propose only Iceberg tables (we can also do files-only datasets, without table definition)
• Propose Hive tables in addition to Iceberg tables (classical folder with files, folder-defined partitioning)

If I don't mistake, the repo you pointed don't have the refinery code you reference.
If we decide to put the code you reference in the repo, we make a decision to move toward a monorepo.
Mono-repo versus multi-repo is still very much under discussion in the DE team, and I wouldn't commit to make the change you're asking for as of now.
But, my assumption is that other folks would give you the green light...
Sorry for not being helpful here.

Had a chat with @Antoine_Quhen and he had great suggestions:

• Add some job configuration information in the end-table (to detect changes)
• Add an application status (we have it at job level, but better to also have it at app level), to easily monitor when an app has failed
• Add a task-identifier for mapped-tasks in Airflow, otherwise their airflow_id will be the same for all mapped tasks

Good progress to report here.
Reading existing EventLogs is a way better idea than generating events: the logs contain exactly what we need, and we it prevents us to generate yet another stream of the same data. The only downside is if at some point we wish to interact with the metrics in realtime for alerts for instance. For now, the batch computation over the existing logs is good enough. Note: if we need to generate events, the code will not need to change much as the Spark internal mechanism to retrieve metrics-events from the EventLog is the same we would use in real jobs.
Currently three PR live:

• Add the spark code to read the EventLogs, aggregate them at stage level, and insert them into an iceberg table.
• Add the spark.wmf.metrics.airflow_id parameter to all our airflow saprk jobs. This parameter will be our way to retrieve metrics for jobs that are "the same" across time.
• Add the table creation script for the metrics.

Still missing:

• New airflow job to add previous day jobs metrics into the metrics table

One mopre note: I have paused the airflow job to prevent alerts to be sent while testing. Let's not forget to unpause it after we merge the patch.

I have checked the job, and the problem is deeper than just memory resources.
The job was suffering from data-skew and Spark couldn't optimize it due to the way the query was written: doing a left outer join and filtering for null in the right table can't be optimized in Spark, while using LEFT ANTI JOIN does exactly that and can be optimized.
I'm sending a PR with that change, which in addition doesn't require resource bumping.

Bringing this to @Ahoelzl and @Ottomata for attention / prioritization.

IMO best approach would be to use an Iceberg table, and make it as easy as DELETE FROM ... WHERE ... to purge data.
I don't think you'd need to upgrade you job spark version in order to benefir from using an iceberg table, but in case, you could also upgrade your code to use spark 3.5 and Iceberg 1.6.
This setup is already present on the cluster (replace spark3-submit by spark35-submit) for you to test, and this is also already used in production (see https://github.com/wikimedia/analytics-refinery-source/tree/master/refinery-job-35 for our dedicated module).
Let me know if I'm not making sense :)

All "production" tables updated in the metastore/HDFS.
Still one patch to be merged, waiting for review.

Great analysis - thank you @xcollazo.

List of repos to review from codesearch (regex: "using iceberg")

• analytics/refinery (30 files) -- https://gerrit.wikimedia.org/r/c/analytics/refinery/+/1341854
• repos/data-engineering/refinery-and-airflow-sql (17 files) Not touched, this repo is a copy of refinery and will need to be updated.
• repos/data-engineering/airflow-dags (12 files) -- No change needed
• analytics/refinery/source (9 files) -- No change needed
• repos/data-engineering/amastilovic (10 files) Personal repo, not touched
• repos/data-engineering/mediawiki-content-pipelines (5 files) -- https://gitlab.wikimedia.org/repos/data-engineering/mediawiki-content-pipelines/-/merge_requests/115
• repos/data-engineering/refinery (4 files) -- no change needed
• repos/wikidata-platform/analytics (1 files) -- https://gitlab.wikimedia.org/repos/wikidata-platform/analytics/-/merge_requests/4
• repos/wikidata-platform/wikidata-query-rewriter (1 files) -- No change needed

Something I forgot to mention: changing the location to the fully qualified path work but doesn't change the already existing data and metadata files referenced in the various iceberg metadata files.
Newly created files use the fully qualified path (including metadata file), which means that for metadata, new snapshots rollover and old snapshots deletion will at some point get rid of no-fully-qualified paths. On the data side though, getting rid of the old files path means rewriting the entire table.
I plan on us doing the full-rewrite on demand if people really need it.

I have tested updating two tables location for their fully qualified versions, and it worked:

ALTER TABLE wmf_traffic.aqs_hourly SET LOCATION 'hdfs://analytics-hadoop/wmf/data/wmf_traffic/aqs/hourly';

Re-opening after having done some investigations: Andrew is right, as long as we change only the location to be fully-qualified instead of just a path, we should ok.
I tested that on a personal table with the command:

ALTER TABLE dbname.tablename SET LOCATION 'hdfs://analytics-hadoop/table/path'

As we're not going to put pageview_hourly in Turnilo, I'm boldly declining this task.

In T437638#12309808, @Ottomata wrote:

Hm, how urgent is this? We are working with RDS to redefine the meaning of 'access_method'. It would be nice if webrequest could use the same taxonomy.

T419019#12238068 & T419019#12297742

In T437472#12304529, @Ottomata wrote:

Judge components, not projects. The parts that matter are the Iceberg maintenance actions, the catalog client, and whatever expresses table policy.

Let's drop this. Judging projects matters. Community strength especially does, as we don't want to adopt something that will lose steam in a couple of years. Unification also matters.

Done!

Thanks @Antoine_Quhen and @xcollazo for your comments.

In T437079#12294644, @BTullis wrote:

Thanks, yes I agree that this was a bit wooly and vague. How does this version of the SeaTunnel description sound, instead?

The new version is great, thank you!

The only comment I have here is about the assumption that we want to go to from kafka to Iceberg directly.
Even if we decide to have an Iceberg only metastore, I think the new data platform should also be capable of reading datasets as folders, allowing us to ingest files in folders in the same was we do now, without having to handle the Iceberg metadata and schema management aspect. I like this approach for the separation of concern aspect, where we move data as it comes (simpler), and in the next transformation we handle schema management/validation and data changes (the same way we currently do with refine).

Thanks a lot @BTullis for having formalized all this :)
Here are some comments here about what's written in the task description:

• Trying to precise the sentance "This is the closest match in shape to what Sqoop does for us." in the SeaTunnel section: To my understanding, SeaTunnel could handle both the data-movement aspect (sqoop) and the multi-tables/schema jobs and transforms (sqoop management script) using configuration files. If we were to use spark, it would be a replacement for sqoop itself and we would need to add an equivalent of the sqoop management script.
• About Sling and dlt: My quick scan of their presentation docs don't highlight their scalability capabilities. While they would probably be ok for the many small tables we have, they wouldn't for the few big ones we also need to handle.

Actually I didn't realize but my patch changed the conf for both turnilo-next and turnilo. I just applied the helm change to turnilo, no need for a new patch.

Ok, applying the same new field to the main turnilo config. Thanks @CDanis .

My 2 cents:

• Replacing Gobblin should be a lot easier than replacing Sqoop as we use Gobblin without helpers, while we have a Sqoop management script doing a lot of custom work
• If we decide to use a streaming engine (Flink), we should use it in stream mode. Similarly, if we prefer the jobs to be handled batch, we should use a batch engine (Spark). While both Spark and Flink can do both Streaming and Batch, each of them has been devised with a primary use-case in mind, and I think we should keep them doing what they do best.
• I have found a new Apache project that seems to be doing what we wish in terms of data movement: https://seatunnel.apache.org/. Worth evaluating!

In T422443#12228359, @Zabe wrote:

@JAllemandou the DAG is failing, but only for commonswiki. Any idea why that is the case?

Hi @CDanis and @Krinkle ,
I have added a new field X-Is-Browser score (STR) to make filtering possible in Turnilo-next: https://w.wiki/T$ar
If you're happy with that, I'll roll it out to the main turnilo as well.

I asked Claude to help me with the categorization, and I think it did a great job: https://claude.ai/code/artifact/1e94957f-b4a2-4426-b1eb-84b8e04cd757
Ping @xcollazo and @Antoine_Quhen on this for review please :)

Summary from meeting today between @atsuko and myself:

• We want to test classes of DAGs that represent usage patterns
• @JAllemandou will try to define the DAG classes using an iterative approach (most used usage patterns first) and to find examples to be tested for those classes

Reverts are computed in mediawiki_history_incremental_v1 in an almost same way as they in mediawiki_history.
You can access the is_identity_reverted and revision_seconds_to_identity_revert field in mediawiki_history_incremental_v1, updated daily.

I think this task is a duplicate of T434160: Decouple airflow-dags repository from Archiva.
I let @amastilovic make the decision of merging as duplicate.

This is really great :) Thank you so much @APizzata-WMF for fixing this.

@Ottomata: Do you think I should wait for us to push a global OTEL style event format?

Starting this work, I'd like to talk about the approach and suggest a different one. Instead of sending metrics to Prometheus,which we we know has resource constraints in terms of metric cardinality, I'm wondering if using the event system, with a dedicated schema for our spark metrics could be a good idea.
Pros:

• no Prometheus overload, keep everything in the cluster

Cons:

• Less easy to visualize and alerting not yet tested (superset vs grafana)

I have spoken with @Ladsgroup . For our mediarequest and mediacount jobs to continue to work, we need to:

• Have them read rows from both upload and text webrequest sources
• Have their request uri_host be one of (upload.wikimedia.org, thumb.wikimedia.org)

We currently don't provide close-to-realtime pageviews, nor hourly pageviews in the pageview API.
We however provide hourly pageviews per page every hour as dump files: https://dumps.wikimedia.org/other/pageviews/2026/2026-07/
The files are a few hour late, but that's the best we have for now.
The files are not small, and you'll need to extract the trending ones yourself. I think you could use something like polars or DuckDB to help this :)