Wikimedia Developer Summit/2016/T114019/Minutes

Intro

 * Session name: Dumps 2.0 for realz (planning/architecture)
 * Meeting goal: Make headway on the question: "what should the xml/sql/other dumps infrastructure look like in order to meet current/future user needs, and how can we get there?"
 * Meeting style: Depending on participants, either Problem-solving or Strawman
 * Phabricator task link: https://phabricator.wikimedia.org/T114019

Topics for discussion

 * use cases for the dumps, known and desired
 * where we currently fall short or are expected to fall short in the future
 * an ideal architecture for dumps that would address the main issues would look like... what?

Next steps

 * Ariel: I want to walk out with a list of things we should investigate further, possible solutions, people interested in following up
 * Possible Ideas:
 * Event Bus for the incremental part of the dumps
 * Change Propagation system to coordinate the jobs
 * HTML dumps to use the html directly instead of asking mediawiki to render it
 * Ariel: would like to have a system where someone just sets up a little puppet configuration, and that's all that's needed to run a specific dump job
 * Mark: think there should be consideration about what is the actual format we want to dump. HTML and raw event dumps are different enough that transformations need to be thought of ahead of time (more so than XML to JSON)
 * Dan: idea: use heuristics to regenerate data only when it makes sense. For example, if most revisions are deleted within 5 days, don't do daily regeneration of revisions that are more than 5 days old.  Those would still change, but we can regenerate them monthly and still keep the level of vandalism in the dumps nomially low
 * Mark: if we had rolling regeneration and de-coupled the streams of raw data from the output, we could have flexibility to generate different outputs and they would all be as clean as possible
 * Andrew: should look at hadoop because this problem matches the map-reduce paradigm pretty well
 * Nuria: maybe celery's easier as an interrim solution, since Hadoop doesn't talk to mediawiki easily
 * (?) be able to view the state of a wiki page at that moment in time (not just wikitext but how the templates would expand, what category it's in, etc), what would be needed, are there use cases?
 * Post-meeting discussion: keep pages plus their revision content as single items in HDFS, incremental update means adding pages, removing pages? Regenerating means rolling these up into small downloadable files?

Jaime discussion notes

 * incremental dumps for sql tables. some of these tables we could keep track of the last row dumped and then dump everything inserted afterwards.
 * some tables are insert only, these in particular can be easily managed for incrementals


 * private data and dumping only certain columns based on values of flags in other columns (or even fields in other tables)
 * would it make sense for performance to not rely on MW managing the decision about which data is public? do we want to rely on the views we use for replication to labs?
 * want a more formalized process for changes to schema, where every item in every table is annotated by the devloper as private, public, or depends on certain flag values.
 * formalize the db schema change workflow. See: https://wikitech.wikimedia.org/wiki/Schema_changes This will not cover cases where the use of a field is changed; look at past changes for examples.


 * note that not all changes to dbs are done via mw but some are directly out of band, need to track those.
 * note tokudb + mysqldump of all tables = huge speed win. takes 12 hours. use this as a basis for filtering out private data? but there is still es. use the "labs view"
 * https://git.wikimedia.org/summary/operations%2Fsoftware%2Fredactatron : this is the production-side filtering, done with replication filters + triggers. Maintained by production DBAs (Jaime)
 * https://git.wikimedia.org/tree/operations%2Fsoftware.git/master/maintain-replicas: Maintained by labs (was Coren) Views restricting access to certain fields based on flags in other fields, or based on user privileges
 * if we had a db with only public data in it, how useful would that be? we could export it as full sql dbs and provide the data in that format. so bypass mwbzutils conversion tools!
 * sort revision content by primary key order and retrieve that way, might be much faster, up to ten times because of mysql db insert order - close physical proximity on the disk! and then a read of one item pulls in the rest etc
 * jaime mysql wrapper abstracting lb class, for use by monitoring and db admin, in development, eventually want to remove dependence on MW so this would be a separate library. currently opens new connection for every request (need this to be dependent on an option)

Hoo discussion notes

 * wants api for the user to find out which dumps are in progress, complete, one-click download of latest wiki etc.
 * desired: way to capture data from history that is not revision/page metadata and would only be present in a full dump of the specific date (e.g. category changes, we don't have a history of those)

Adam Wight discussion notes

 * I don't have anything architecturey yet, but from a risk management perspective, the migration path seems a bit unclear. I tried to imagine a low-investment way to experiment with Dumps 2.0, in.
 * Would symmetric nodes be better than a master scheduler node? How much complexity that would add?  This would give us extra fault tolerance and make it simple to run on additional clusters when needed.  One easy (but not high availability) division of labor would be to have a master scheduler assign an entire dump to a worker scheduler, which can then subcontract chunks to other workers in turn.  And pay them very little indeed... :S
 * We should use multiple brokers for high availability. Redis Cluster would give us that for free, otherwise it seems to be supported by Celery somehow.
 * Can someone who has survived a Celery integration please vouch that it's easy and fun? ?
 * I like JCrespo's idea of using WMF Labs views as our sanitization, mostly because that lets us reuse existing logic. Dumping from a paused Labs slave seems ideal, if the network and filesystem considerations aren't too horrible.  Presumably, the Labs sanitization happens on the main cluster, so we should be able to replicate a slave from there, to avoid crossing network segments?
 * We need some statistics about the time it takes to suppress revisions in oversight situations. Rather than risk mirroring these bad changes, we might want to delay incremental (and full) dumps by a day or two.  I guess that would look like, is that we would dump from an up-to-date, sanitized database, but daily incrementals for example would actually contain 2 days > age > 1 day old revisions.
 * The discussion about dump formats is really interesting, and IMO should take place independently of this framework rewrite. A lot of the dump consumption use cases sound badly broken, we should collect those stories and identify pain points.  I agree with apergos that a diversity of formats is great, but at the same time we should choose a canonical set and then put energy into improving that to cover 90% of potential uses, include indexes and stuff, and provide reference consumer libraries.
 * gnosygnu's is really onto something, historical dumps should only change when the formats are updated.
 * Just a tiny detail, I think it's important that our runner framework interfaces with chunk processing jobs at the command-line, to minimize the amount of glue we need. Jobs will be written in multiple languages, lets not mess with bindings.  On that note, the object store interface should be something simple to access from any language, maybe Redis?

Gabriel Wicke discussion notes
The main needs seem to be to speed up the dump process by doing incrementals and to allow third parties to incrementally update their dumps.

For HTML dumps we resolved this by using compressed sqdblite dbs. Lots of tools for sqlite, only issue is concurrency, but since updates take so little time you can do without.

Current revisions for the main namespace for en wikipedia are 200GB uncompressed, 27.4GB with xz compression, it takes less than two hours to walk through all page titles and revisions, retrieve information via the api and compare to current sqlite db contents. Note that we uncompress the db first, run this, then recompress, but even so it's not that expensive. The db is currently page-title based instead of page id, so page renames are a bit of a drag. For large dbs (with revision history) we would want to divide up the db into several smaller ones, probably by page id range. Storing revisions per page together would mean better compression too.

We were looking into brotli compression, not enough support for it yet. Have a look at benchmarks: https://quixdb.github.io/squash-benchmark/

So once you have a full dump, an incremental works like this: uncompress full db, get a working copy, get the changes for the page id range, apply those, recompress, atomically move to new location. If it fails you just get a new copy of the db and uncompress again, have many many shards so it's fast. This assumes you get a feed of relevant events. There is no authentication right now for the events feed til we go to kafka 0.9, you use a kafka consumer, there's a python library for that.

Eventbus stuff:
 * See https://grafana-admin.wikimedia.org/dashboard/db/eventbus for what's happening.
 * https://phabricator.wikimedia.org/T114443 this is the tracking bug for our eventbus deployment
 * https://phabricator.wikimedia.org/T116247 discussion about the schema. soon there will be user related events like user suppression.
 * https://github.com/wikimedia/mediawiki-event-schemas/tree/master/jsonschema/mediawiki and the schema itself.
 * MW extension for this .... https://github.com/wikimedia/mediawiki-extensions-EventBus

Misc notes:
 * if people download new dumps, how do we support people with xml based tools? write a converter
 * not all fields in xml dumps are yet in the sql dbs, but adding additional fields is relatively easy
 * most folks have import step from xml to db or something else. some people also stream it, most researchers don't stream
 * note that db format allows parallel access right away (but you do have to uncompress first)

Supporting OpenZIM production:
 * zim format is good for random access into a file, has lzma compression
 * a while ago they switched to restbase (the openzim / kiwix folks), they want the html dumps instead of hitting the restbase api for everything
 * they need hardware to really get their job done, maybe share our server for the html dumps? or just rent a server somewhere, they have been doing by themselves so far
 * funding is covered right now, they are mainly interested in comaintaining and being integrated into normal dump process somehow
 * there's a ticket open for network isolation about isolating services from each other, that might help, but this is very manual right now, mark and faidon are worried about doing this right now for that reason: https://phabricator.wikimedia.org/T121240

For sql table dumps, maybe leave as is, if we need incrementals, we could consider providing a script that reads sqlite & writes each row to mysql (or generally $DB)

We might allow users to apply changes from eventbus themselves;(we had talked about daily event logs for download with a tool to apply them). Most of those API requests (for recent changes) can be cached, so shouldn't cause terribly high load on our backend infrastructure, varnish would cache and serve these

Milimetric notes
The whole team participated in a discussion so these notes are a summary of that discussion.

EventBus discussion
Concern raised about a recent site outage related to MW failure to send events when kafka server was down. This turned out to be an hhvm bug in fsockopen implementation, so that when you try to connect to a dead host, it will not respect the timeout you pass it. Ticket: https://phabricator.wikimedia.org/T125084 So we really should not have normally the situation where MW app servers are backed up waiting to write events. However, we then have the problem of how not to miss events from MW.

For MW an event is 'committed' when sql transaction is complete, but then there is the possibility of failure to emit the event. One possibility: have a process comb through dbs periodically looking for events not emitted; this would require db schema change to have some field indicating that an action had its corresponding event emitted. Another option less likely; have a process reading the binlog and feeding the entries to a script that generates the appropriate events.

The binlog approach is potentially hard because we have to translate from that format to get actual clean events. But it might be a good way if used in addition to MW emitting events, because events produced from these two sources could be mutually compared and cleansed.

kafka mysql binlog libs:
 * https://github.com/mardambey/mypipe
 * https://github.com/zendesk/maxwell
 * https://github.com/pyr/recordbus

Question: how well do the binlog producers replicate events, how many events do they miss? What about if they're put under stress? kill -9, memory constraints, etc. existing event stream, improvements pending: https://phabricator.wikimedia.org/T124741

Users might want only a subset of the dumps, e.g. some wiki project, do we want to apply filtering to events somehow? Is the api just 'fast enough'? We always have to provide full history dumps though because right to fork. Depending on answers to the previous, look into store pre-computed partial dumps (would be trading storage and computation for bandwidth and potentially helping researchers); haivng a daily file (santized, reformatted) of events for users to grab and apply, and for processing incrementals, instead of having to real time process events; converter scripts that convert our canonical format to format the users want (maybe we can get them to write those scripts!)

Output formats: Aaron wanted line by line format so each revision is in its own object (record, whatever) atomic so it can be dealt with, json event level data would be good, you could partition it in folders and import it into hive. Note that we can't just do a straight up grouping of revisions by page id, because there are too many revisions in some pages to deal with. We do however want to connect the revisions by page somehow and order them.

A weird problem: sometimes events are received that are a month old or even older. Why? How far back? what happens when we receive these events, how would we process them?

Question: what is the nature of the events that we need to store? Do we really get revision events from over a year in the past? Knowing this will help us decide on an incremental file storage scheme. Without knowing this, we might have to store files containing revisions in two separate directory hierarchies, so we can access page-lev el information and time-based information (THE FOLLOWING IS NOT A SUGGESTION!! :)): :-) But if we do know the nature of the events, we may be able to be more efficient
 * <>/<>/<>_revision_metadata.json
 * <>/<>/<>_revision_metadata.json
 * <>/<>/<>/<>/revision_text.wiki

we want to have the text in one canonical location and then (like indices) have multiple "Pointers" to the text. (otherwise we run the very real risk of the text copies getting out of sync, and then we're screwed.) this lets us not have to do very expensive sorting operations. Time-stamped based? One thing about how revisions are applied is that imports of pages + their revisions from other wikis and subsequent merges can really screw the pooch for that (just a thing to keep in mind). Note that undelete of pages now finally means we keep the old page id at any rate, that change was recently merged.

Otto links for eventbus info

 * Schemas:
 * https://github.com/wikimedia/mediawiki-event-schemas


 * EventLogging:
 * https://github.com/wikimedia/eventlogging

service:
 * https://github.com/wikimedia/eventlogging/blob/master/bin/eventlogging-service
 * https://github.com/wikimedia/eventlogging/blob/master/eventlogging/service.py

Output format thoughts
Having an output format useful for distributed computing (Hadoop etc) would be good. So JSON records -- one line per revision (page partitioned and chronological) for import into Hadoop/Spark. The current dumps get converted to "RevDoc" format, then sorted into timestamp order, split up and imported. (Sorting is necessary because we don't produce revisions per page in a guaranteed order, see https://phabricator.wikimedia.org/T29112 and comments. RevDoc format information here: https://github.com/mediawiki-utilities/python-mwxml/blob/master/doc/revision_document-0.1.0.json Note that some information (namespaces, dbname) is lost during this process, and while the api could be queried for that info at the time of the import into Hadoop, things might have changed since the dump was run.

It would be nice to have modular output formats that can be maintained by the interested parties (e.g. Aaron like Revdocs, so he maintains revdocs output) so data gets into hadoop as described above (let's say) and then halfak's converted job takes over to create the desired output. So we need a modular way to add new output formats and a way to schedule these conversion jobs. E.g.: Halfak writes code to perform conversion, system then implements conversion with each new dump (or incremental).

We could also have a converter directly for events emitted to EventBus that can convert to RevDoc format without worrying about ordering.

Secondary data sets
Secondary data sets that could be generated from dumps,e.g. diffs between revisions, are used frequently for research. Collapsing differences between several revisions is cheap. Diffs (generally applying https://en.wikipedia.org/wiki/Longest_common_substring_problem ) N^3 at worst case -- (guess) Takes about 100 days to generate diffs for Enwiki's articles current revs vs prev revs, on one core. On hadoop, we have this down to 3 days -- Could get down to 24 hours possibly (note that includes compression time) using bz2 (sadly) because of ecosystem consistency considerations. Anyways a pipeline produces several intermediate sets of data when going from xml to (data I process in hadoop), the intermediate sets are useful, want to be able to do diff generation on these too.

Other issues
what do we do with page imports as events with old data? Do we want a known event order (token with each event in serial) so we can know that we are missing X from the stream and ask for the back events?

What about incremental dumps with sensitive content, i.e. deleting (rev oversight) old revisions that make it into the dumps?

Is it worth looking into hadoop as a service for cost reasons? We wouldn't be able to give them private tables/wikis/columns. What about bandwidth for downloaders? AWS backend would cover that, there's issue of cost of bw course. Current example: the Research Team is working with a vendor that provides Cloudera package services (Hadoop/Spark/etc.) on a Analytics Team-scale cluster for ~$5k per month. (public data only!)

Action items with owners

 * Create a project and get everyone on it for further discussion (Ariel) - DONE
 * identify other action items from discussion on project, coordinate further discussion (Ariel) - IN PROGRESS
 * DRAFT List here: Wikimedia Developer Summit 2016/T114019/Minutes/Questions

DON’T FORGET: When the meeting is over, copy any relevant notes (especially areas of agreement or disagreement, useful proposals, and action items) into the Phabricator task.

See https://www.mediawiki.org/wiki/Wikimedia_Developer_Summit_2016/Session_checklist for more details.

Lightweight VCS as intermediary
Update 2016-03-06: Brion is thinking along different lines, more akin to a relatively lightweight version-control system as an intermediary between the MediaWiki database on the one hand, and a local data set on the other.

Notes & questions mirrored from blog post:

The basic data model for the main content dumps hasn’t changed much in 10 years or so, when I switched us from raw blobs of SQL ‘INSERT’ statements to an XML data stream in order to abstract away upcoming storage schema changes… fields have been added over the years, and there have been some changes in how the dumps are generated to partially parallelize the process but the core giant-XML-stream model is scaling worse and worse as our data sets continue to grow.

One possibility is to switch away from the idea of producing a single data snapshot in a single or small set of downloadable files… perhaps to a model more like a software version control system, such as git.

Specific properties that I think will help: Some major open questions:
 * the master data set can change often, in small increments (so can fetch updates frequently)
 * updates along a branch are ordered, making updates easier to reason about
 * local data set can be incrementally updated from the master data set, no matter how long between updates (so no need to re-download entire .xml.bz2 every month)
 * network protocol for updates, and access to versioned storage within the data set can be abstracted behind a common tool or library (so you don’t have to write Yet Another Hack to seek within compressed stream or Yet Another Bash Script to wget the latest files)
 * Does this model sound useful for people actually using Wikipedia data dumps in various circumstances today?
 * What help might people need in preparing their existing tools for a switch to this kind of model?
 * Does it make sense to actually use an existing VCS such as git itself? Or are there good reasons to make something bespoke that’s better-optimized for the use case or easier to embed in more complex cross-platform tools?
 * When dealing with data objects removed from the wiki databases for copyright/privacy/legal issues, does this have implications for the data model and network protocol?
 * git’s crypto-hash-based versioning tree may be tricky here
 * Do we need a way to both handle the “fast-forward” updates of local to master and to be able to revert back locally (eg to compare current and old revisions)
 * Technical issues in updating the master VCS from the live wikis?
 * Push updates immediately from MediaWiki hook points, or use an indirect notify+pull?
 * Does RCStream expose enough events and data already for the latter, or something else needed to ‘push’?
 * Can update jobs for individual revisions be efficient enough or do we need more batching?