Page MenuHomePhabricator

Add "wiki_id" to Page View Stream
Open, Needs TriagePublic

Description

The "webrequest_frontend_text" topic in Kafka we are using to produce the "page_view" stream doesn't have wiki_id, it only has a few fields that can be used to guess the "wiki_id".

These fields are in the current "page_view":

"uri_host": "en.wikipedia.org",
"pageview_info": {
		"page_title": "Hyundai_Department_Store",
		"project": "en.wikipedia",
		"language_variant": "default"
	}
"normalized_host": {
		"project": "en",
		"project_class": "wikipedia",
		"project_family": "wikipedia",
		"qualifiers": [],
		"tld": "org"
	}

We could use pageview_info.project probably as key, along with page_id, but the standard way to do it in our pipelines is to use "wiki_id", which we don't have here.

Ideally, we should find the way of adding wiki_id into the "page_view" stream, without adding a lot of I/O, as this is a high throughput topic.

There is one interesting endpoint that returns a matrix of wikis, with their url (similar to project) and dbname which seems to be equal wiki_id.

Matrix endpoint: https://meta.wikimedia.org/w/api.php?action=sitematrix&format=json

Other ideas we have considered, but would require more time:

  • Find the way of adding wiki_id into webrequest_frontend_text, probably inside the x-analytics field (This would be the recommended way)
  • Create a lookup table in Kafka with a log compacted topic. MediaWiki would require to produce events about changes in Wikis, so we can keep the lookup table updated. (This will require more work in different places, so it is discarded for now)

The simplest solution looks to simply call the Matrix endpoint at startup and load all the wikis (url - dbname) in a key value store that can be shipped to all Flink workers.

Details

Related Changes in Gerrit:
Related Changes in GitLab:
TitleReferenceAuthorSource BranchDest Branch
Page_view: Adds `wiki_id` from sitematrix.repos/data-engineering/mediawiki-event-enrichment!152javiermontonfeature/page-view-4-1-wiki-id-startupmain
Customize query in GitLab

Event Timeline

Find the way of adding wiki_id into webrequest_frontend_text, probably inside the x-analytics field (This would be the recommended way)

This might not be difficult. At least, from a code change perspective, I'd imagine it would be finding some global place in MW with which to add wiki_id to the X-Analytics headers. It might be more complex from a social perspective (will MW teams want to do this?)

Anyway, I'd recommend investigating this first.

Change #1297702 had a related patch set uploaded (by JavierMonton; author: JavierMonton):

[mediawiki/extensions/WikimediaEvents@master] extension: WikimediaEvents - add `wiki_id`

https://gerrit.wikimedia.org/r/1297702

Nice patch!

@mpopov @JAllemandou @phuedx I'm not sure who exactly to ask, but is there any reason we shouldn't add wiki_id in X-Analytics for all requests, not just pageview requests?

Nice patch!

@mpopov @JAllemandou @phuedx I'm not sure who exactly to ask, but is there any reason we shouldn't add wiki_id in X-Analytics for all requests, not just pageview requests?

Data duplication and size is the only argument I can think of.
Using the canonical.wikis` table available from Hive is the way we usually do to retrieve project information when we have on of them. The table is small enough that it probably could be pre-loaded in the flink job.

The table is small enough that it probably could be pre-loaded in the flink job.

Yes, but then we'd need to manage updates for new wikis. If we had to fetch the values, I'd rather fetch it in flink job from the MW API and cache the values. New wikis will just cause a new fetch and cache.

But anyway, for this page view stream, we only need it on page view requests, so the X-Analytics patch as is will be fine. I was just wondering if it might be useful to have for other potential non-pageview uses of webrequest. Maybe not?

Even for pageview only, I would prefer not duplicate data that can easily be recomputed from other sources, for networking leanness.

I'm with @JAllemandou here. Also, why not just update the UDF that produces normalized_host so that it also includes wiki_id?

Also, why not just update the UDF that produces normalized_host so that it also includes wiki_id?

I think for the same reasons that joining with data from canonical.wikis is not good here. This is an online streaming job, so it would require restarts if the canonical wiki list changes.

However, fetching and caching the wiki_id from sitematrix API the first time a domain is encountered should be fine too. It will be one fetch on startup, and then a fetch each time a new domain is encountered.

Sorry @JMonton-WMF. We should have asked before I advised you to go the X-Analytics route.

Also, why not just update the UDF that produces normalized_host so that it also includes wiki_id?

But uh, perhaps we should do this anyway so that webrequest Hive table will also have this data!

I'm ok with any approach, but I'd like to understand properly the cons, pros and concerns.

Data duplication and size is the only argument I can think of.

This concern is about the size in HDFS I assume, right? Or is it in Kafka? or in other places? I guess the table wmf_raw.webrequest will get more data. I'm assuming that other tables, like the "page_view" processed table doesn't have issues with this, as the process will either ignore the new field or use it, but the result will be the same.

The new field is probably around 15 bytes (e.g.: ;wiki_id=enwiki ) and only on records that already have a page_id, which are around 7,5% ~ 8,5% of the webrequest records. In absolute numbers it's around 30 ~ 35M records per hour. 15 * 35M * 24(hours) / 1025 (KB) / 1024 (MB) / 1024 (GB) = ~12GB per day (Am I missing replication?). It will be 360GB more monthly data in wmf_raw.webrequest , right?

I see that a month of data in /wmf/data/wmf/webrequest/webrequest_source=text/year=2026/month=X is around 65TB, and they will be increased by 360GB more if we add this field, around 0,6%. (I'm not lookig at webrequest_source=upload

In Kafka the retention is 7 days, the topic has now 68702995122 records (51TB). Around 8% will get 15 extra bytes (68702995122 * 0.08 * 15 / 1024 / 1024 / 1024 = ~77GB. 77GB*3 replicas = 231GB). The topic will get 231GB extra on those 51TB, around 0,5%.

I understand the concerns about that, it is indeed a bunch of new data to be stored, specially in absolute numbers, but wanted to know if that's the issue or if I'm missing something, or doing the numbers wrong.

My biggest concern with doing HTTP calls in Flink is the I/O waiting. The topic has a throughput near 150k messages/second, we'll process them in parallel and completely stateless, which is very fast, but if we add a process that needs to do HTTP calls from time to time, it could cause the whole pipeline to lag.

  • If we load the mapping at startup it could be fine, only one call during startup, and the data is stored in the workers memory, but as Andrew pointed, it will fail if there are changes on the wikis as it will require a restart (maybe we could throw an exception, break the application, and let K8s restart it, so it will pick the new mapping? )
  • If do a call and cache it, it won't be a single call, it will be a call per worker, and we may have many small workers. At startup all workers will need to start doing calls and blocking all messages. And any time a worker dies (OOM, K8s relocating resources, or whatever, they will need to do the call again). Maybe it won't be a big deal, but I think we need to try it, maybe it's enough to not be able to catch-up with high throughput.

The code complexity concerns me a bit too, having a simple Flink pipeline doing source -> map -> sink is simple, but adding HTTP calls, cache, broadcasting a mapping to workers, and ensuring Flink workers doesn't wait for others workers (Flink's "map" vs "process" with unordered async) will make it a bit more convoluted.


In any case, I'd like to ask for advise, is the mapping clear?
We have this in the matrix, HTTP call:

{
          "url": "https://aa.wikipedia.org",
          "dbname": "aawiki",
          "code": "wiki",
          "sitename": "Wikipedia",
          "closed": ""
        }

And we have this in the event:

"uri_host": "en.wikipedia.org",
"pageview_info": {
		"page_title": "Hyundai_Department_Store",
		"project": "en.wikipedia",
		"language_variant": "default"
	}

Can we do this?

"http://" + uri_host == url
// or
"http://" + pageview_info.project + ".org" == url

So we can retrieve the dbname? or are those fields not always valid? e.g: Can uri_host be changed by the user? do we always have .org?

I understand the technical concern about loading API data in the flink app. I don't have a good solution for this.
About your mapping question, you should use "http://" + pageview_info.project + ".org" == url . The uri_host can still sometimes have modifiers subdomains (en.m.wikipedia.org).
And actually the best would be to use the normalized_host field, instead of pageview_info.


My concerns about adding a new field are two ways:

  • More data - my concern is less about data on HDFS or Kafka than it is about sending more data in the network. The webrequest-logs traffic flow is pretty big already, and I'd rather not make it bigger.
  • Data duplication - I'd prefer for us to have better sources of truth for small/canonical data tather than multiplying the same redundancy everywhere.

Change #1299435 had a related patch set uploaded (by JavierMonton; author: JavierMonton):

[mediawiki/extensions/EventBus@master] eventbus: mediawiki-user-change

https://gerrit.wikimedia.org/r/1299435

Change #1299437 had a related patch set uploaded (by JavierMonton; author: JavierMonton):

[operations/deployment-charts@master] topic: webrequest-page-view

https://gerrit.wikimedia.org/r/1299437

Change #1299437 merged by jenkins-bot:

[operations/deployment-charts@master] topic: webrequest-page-view

https://gerrit.wikimedia.org/r/1299437

I've done an implementation of the wiki_id on Flink:
https://gitlab.wikimedia.org/repos/data-engineering/mediawiki-event-enrichment/-/merge_requests/152

It's running at 1M messages/second with 20 Task Manager, so it doesn't seem to have issues with throughput.

The approach is the following:

  • On each Task Manager, at startup, they fetch the sitematrix, parse its content and keep a Java Map<String, String> in memory.
  • Each record does a search on the Map to find the wiki_id.
  • If a record fails to find the wiki_id, it produces a metric, increases a internal counter, and produces the event with wiki_id: null
    • This is to avoid blocking the pipeline for each failure to find the wiki_id. I assume that loosing a few events that doesn't have a wiki_id should be ok. They are either wrong records or newly added wikis.
  • If counter passes the configured threshold, the Task Manager will refresh the sitematrix on memory. This is done synchronously, so the queue may be blocked for 1 or 2 seconds.
  • There is also a cooldown timer (Java Duration) that is checked before refreshing the sitematrix. This is to avoid a situation where wrong records, or a missing wiki in the sitematrix could cause thousands of messages per second to trigger the refresh.

I thought about adding the sitematrix fetch only during the startup, in the Job Manager, and pass the Map to the workers, but that will fail if a new wiki is added to the sitematrix and will require a restart. Maybe it won't be a big deal to manually restart the application when new wikis are added, but I believe the other approach is more robust and self heal itself. In my opinion, I prefer to avoid manual work.