nyuuzyou/wikis huggingface dataset:
Data Platform/Systems/Druid/Alerts

We have a number of alerts set up in Icinga and Alertmanager that relate to Druid and its ingestion jobs.
This page exists as a set of instructions or runbooks to help identify what courses of action might be needed if one or more of these alerts is triggered.
== Druid Netflow Supervisor ==
This alert triggers if the realtime netflow ingestion job receives below a certain threshold of events, over a 30 minutes period.
The critical value is 0 and the warning value is 30.
There is a grafana dashboard showing the trend data.
== Druid webrequest_sampled_live Supervisor ==
This alert triggers if the realtime webrequest_sampled_live ingestion job receives below a certain threshold of events, over a 30 minutes period.
The pipeline is composed by multiple parts:
* Benthos webrequest data enrichment - on centrallog nodes we run a special Benthos instance that pulls messages/events from the Kafka Jumbo's webrequest_{text,upload} topics and enrich/modifies them with meaningful data (geo location, etc..). The end result is a new topic called webrequest_sampled.
* Druid indexation - the Druid analytics cluster ingests in real time the webrequest_sampled messages/events, indexing them in segments that can be queried from various sources (like Turnilo and Superset).
* Turnilo and Superset dashboards.
Note: the use case is similar to Turnilo's webrequest_sampled_128, but the main difference is that in this case we care only about the past 24h of data (and that it is indexed live as it is published to the Kafka topics, as opposed to wait for the DE's batch jobs etc..).
Since the pipeline is split into multiple parts, there may be multiple reasons why Druid is not indexing enough events:
* The Benthos instances on centrallog nodes may be down or misbehaving. Please check status and logs of benthos@webrequest_live.service on those nodes. Useful dashboards are the webrequest_sampled topic's metrics and the Benthos dashboard.
* Druid is not indexing Kafka events/messages correctly. In this case start from this grafana view and ping Data Engineering!
== Druid Segments Unavailable ==
This alert triggers for each data source if the cluster is reporting above a certain number of segments missing over a 15 minute period.
The critical value is 30 segments unavailable for each data source.
The warning value is 20 segments unavailable.
There is a Grafana dashboard showing the trend.
It may well be that Druid auto-heals this unavailability automatically, so chek the general Troubleshooting techniques before taking any action.
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Druid/Load test

This page documents a load test performed on the Analytics' Pivot instance and the Druid cluster on 2016-09-29.
== Experiment ==
3 data sets have been explored in Pivot (https://pivot.wikimedia.org/): Edit History Test (small, simplewiki only, 2001->2016), Pageviews Daily (medium, all wikis, 2015-06->2016-09) and Pageviews Hourly (big, all wikis, 2016-06->2016-09). For each of them a grid of tests have been performed, playing with 2 lengths of time ranges (short, long) and 3 types of queries (totals, time-series, table). Note that totals queries are the simplest ones computationally, while time-series queries have a medium complexity, and table queries are the most complex, because they all possess 2 breakdowns/splits (each of them with limit=10).
The script test/pivot_url_generator.js in https://gerrit.wikimedia.org/r/analytics/pivot has been used to generate 100 URLs for each square of the experiment grids. Then a sample of those URLs have been requested to https://pivot.wikimedia.org and measurements made. Caveat: The requests per second measurements are approximations, because of issues with executing POST request with Siege. This issues con be solved by modifying the URL generator script to generate plywood queries (the ones that pivot executes asynchronously).
== Results ==
=== Edit History Test ===
 	Totals query	Time-series query	Table query
Short time range (<20% of data set) 	30+ req/s 	30+ req/s 	30+ req/s
Long time range (40%-80% of data set)	30+ req/s 	30+ req/s 	30+ req/s
Pivot machine: No alterations to CPU or memory consumption.
Druid cluster: Minor CPU peaks (<10% of 32 cores) when executing bursts of requests.
=== Pageviews Daily ===
 	Totals query	Time-series query	Table query
Short time range (<20% of data set) 	30+ req/s 	30 req/s 	30 req/s
Long time range (40%-80% of data set)	30+ req/s 	30 req/s 	30 req/s
Pivot machine: No alterations to CPU or memory consumption.
Druid cluster: CPU peaks (10%-20% of 32 cores) when executing bursts of requests.
=== Pageviews Hourly ===
 	Totals query	Time-series query	Table query
Short time range (<20% of data set) 	30 req/s 	30 req/s 	10 req/s
Long time range (40%-80% of data set)	30 req/s 	20 req/s 	10 req/s
Pivot machine: No alterations to CPU or memory consumption.
Druid cluster: CPU peaks (30% of 32 cores) when executing bursts of requests.
== Conclusions ==
* Pivot is not a bottleneck in any case.
* Small and medium datasets (Edit History Test, Pageview Daily) work fine for WMF internal consumption.
* Large datasets mostly work, but in some cases, with more than 10 queries per second Druid may have pretty high response times, and Pivot may display timeout messages.
* The most demanding queries for Druid are those that have high-cardinality breakdowns. Such queries on medium or large datasets may break Druid. Time-series breakdowns are OK, because Druid is optimized for them.
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Edit data loading

This page describes the first step of the edit history reconstruction pipeline: the loading of MediaWiki data into Hadoop. This is done using a couple of scripts stored in the analytics-refinery-source repository. After that, next steps in the pipeline will process that data within the cluster and generate the desired output.
=== Sqooping data ===
The main script uses apache sqoop to import some MediaWiki tables from publicly available databases replica and production replicas into Analytics' Hadoop cluster. You can find it in the analytics-refinery github. The MediaWiki tables imported from the public replicas are archive, category, categorylinks, change_tag, change_tag_def, content, content_models, externallinks, image, imagelinks, ipblocks, ipblocks_restrictions, iwlinks, langlinks, logging, page, pagelinks, page_props, page_restrictions, redirect, revision, slots, slot_roles, templatelinks, user, user_groups, user_properties, wbc_entity_usage. Tables imported from the production replicas are: actor, comment, watchlist. Finally some special case sqoop jobs are used to get the cu_changes and discussiontools_subscription tables from the production replicas.
In addition to that, another table is created in Hadoop: namespace_mapping. It contains localized namespaces for every wiki (see namespace mapping script below). This sqooping process is to be done at the beginning of every month, and it is done since the beginning of time every time. It is designed in a non-incremental fashion to adapt to the fact that MediaWiki data (revision, archive and logging) can suffer alterations in log records created in the past.
==== Sqooping wikis in groups ====
As in the other steps of the pipeline, there have been performance challenges associated with the size and nature of the data in this script as well. Sqoop would crash if trying to import all wikis at once, but also a job for each one of the ~800 wikis would be too slow and error prone. The solution we went for is to group the wikis in clusters that can be processed by sqoop in parallel. The groups have been determined by studying the sizes of all wikis. You can see a diagram of the partitions in github.
==== Namespace mapping script ====
An important feature of this process is also the generation of the namespaces mapping table. This table holds a relation between namespace names and namespace ids for all wikis. Note that many wikis have (or have had at some point in time) their own localized versions of the namespace names, like "Benutzer" (German) instead of "User". This table translates all versions (localized and standard) of namespace names into their namespace ids, which will help further steps in the pipeline to normalize and reconstruct the editing data. You can find the namespace mapping script in github.
Category:Data pipelines
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Edit serving layer

This is the last step of the edit history reconstruction pipeline. It is composed of a back end storing service optimized for fast analytical queries, and a front end UI that is connected to it and permits simple slice and dice operations on the data.
=== Publicly available ===
* AQS API — This API uses AQS as a proxying layer to druid, a column-oriented datastore (see below).
=== Internal to WMF ===
==== Back end ====
At the moment we're using druid, a column-oriented datastore, that has been giving nice results performance-wise for recent data (last 2 years). The Druid cluster has currently 3 boxes druid100[1-6].eqiad.wmnet. It supports a considerable amount of non-trivial queries per second, see this load test. A nice advantage of Druid is that it has an awesome dedicated UI called Pivot that works out of the box.
We are also considering and testing other datastores, namelly ClickHouse and Presto, which could provide good performance and allow more flexibility than druid. They would however have no connector to Turnilo. We could try to write our own.
==== Front end ====
We have two tools conneted to the Druid datastore:
* Turnilo, a simple but powerful slice-and-dice UI for data analysis (a rewritten version of the now non-opensource Pivot). If you have ldap access, you can try it at turnilo.wikimedia.org.
* Superset, a less simple but more featurefull tool for BI and dashboarding using druid. If you have ldap access, you can try it at superset.wikimedia.org.
Category:Data pipelines
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Exporting from HDFS to Swift

When generating regular datasets via jobs scheduled in Oozie, you may want to export those datasets for production use outside of the Analytics Cluster. To do so, you (as of 2019-08) have two options:
# On a stat box, pull down files from HDFS to /srv/published/datasets. Data in /srv/published/datasets eventually is accessible to the world at https://analytics.wikimedia.org/datasets/. This is not schedule-able via Oozie itself. See also Analytics/Ad_hoc_datasets.
# Use analytics/refinery's oozie/util/swift/upload workflow to upload a dataset directory from HDFS to the production Swift object store. Once in Swift, the dataset can be downloaded by a service or job outside of the Analytics Cluster.
= Oozie Swift Upload =
To schedule Oozie jobs to upload your regular datasets to Swift, you should include the oozie/util/swift/upload sub workflow as an action in your main workflow.xml. As an example, see Discovery Analytics' esbulk job swift_upload action and related configuration properties.
Swift access credentials are rendered on a per request basis, and usually in environment variable file format for use with the swift python client. See https://phabricator.wikimedia.org/T296945 for an example of such a request.
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Hadoop/Test

The Analytics team runs a Hadoop test cluster in production. This seems controversial since we could use Cloud virtual machines instead, but due to the complexity of the Hadoop set up (Kerberos auth, TLS encryption, Kafka topics to sample to generate data, etc..) it would be very resource demanding to run a test cluster in there.
The purpose of the cluster is to test any kind of infrastructure change before production, from a small config to big and invasive changes like upgrading the Hadoop distribution or changing the security auth schema.
== Hosts and their roles ==
an-test-master1001.eqiad.wmnet - Hadoop master (HDFS Namenode, Yarn ResourceManager, Mapreduce History server).
an-test-master1002.eqiad.wmnet - Hadoop master standby ((HDFS Namenode, Yarn ResourceManager).
an-test-coord1001.eqiad.wmnet - Multi-purpose node (Hive Server/Metastore, Oozie, Presto, systemd timers for refine and druid load).
an-test-worker100[1,2,3].eqiad.wmnet - Hadoop worker nodes (HDFS Datanode, Yarn Nodemanagerm HDFS Journalnode).
an-test-druid1001.eqiad.wmnet - Single node Druid test cluster (all Druid daemons running on it).
an-test-ui1001.eqiad.wmnet - Yarn and Hue UIs.
an-test-client1002.eqiad.wmnet - analytics client
== UI quicklinks ==
Hue: ssh -L 8080:an-test-ui1001.eqiad.wmnet:80 an-test-ui1001.eqiad.wmnet
Yarn: ssh -L 8080:an-test-ui1001.eqiad.wmnet:80 an-test-ui1001.eqiad.wmnet and then set Host: yarn.wikimedia.org in the browser's custom header settings
Druid Coordinator/Overlord: ssh -L 8081:localhost:8081 an-test-druid1001.eqiad.wmnet
== Oozie coordinator testing ==
Refinery source and Refinery are not kept in sync with every "prod" deployment to avoid adding workload to whoever is in Ops week, but anybody that wants to update them can follow the deployment rules stated in Analytics/Systems/Cluster/Deploy/Refinery#Deploying to Hadoop test (plus the usual rules for deployment).
A possible multi-user testing scenario is the following:
* Refinery and source are deployed to HDFS as they are, namely without any ad-hoc changes applied.
* Every user maintains a /user/$username/oozie directory on hdfs, uploading custom oozie files and settings.
* When launching the coordinator/bundle, the user should fix their properties file to point to the /user/$username/oozie directory
The most common example is probably to avoid spamming analytics-alerts@ from the test cluster:
cd refinery/oozie
for filename in `grep -l -r analytics-alerts`; do sed -e 's/analytics-alerts@/ltoscano@/g' -i $filename; done
# then upload to /user/elukey/hdfs in this case
== Hadoop upgrade - jobs to test ==
Before rolling a new hadoop version to the production server we should have tested the following jobs on the upgrade test cluster:
* Gobblin data import (subset of webrequest and events)
* Webrequest Load (on smaller data)
* Events Refine (on smaller data)
* Druid load (webrequest_sampled_128 for instance)
* Wikitext-current (single small XML dump imported manually)
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Hadoop Event Ingestion Lifecycle

High level overview of how an event makes it into the analytics-hadoop cluster. 
== Event Ingestion Lifecycle ==
(There is a recording of a talk describing this system (as well as EventLogging))
Producer code (EventBus, EventLogging, Flink, etc. etc.)
* EventGate (optional)
* Kafka
* HDFS raw (via Gobblin MR job)
* Hive event database (via Refine Spark job)
* Hive event_sanitized database (via RefineSanitized Spark job) (optional)
* unsanitized Hive event table data deleted (after 90 days)
== Canary Events ==
=== Why? ===
Most event streams in WMF Kafka are multi-datacenter. You can learn more about how we run multi-DC Kafka clusters.
An Event Platform stream can be made up of many individual topics in Kafka. In the multi DC case, usually a single stream is made up of its prefixed Kafka topics. E.g. my_stream would be made up of eqiad.my_stream and codfw.my_stream Kafka topics. 
In Hive (via Refine) we ingest these Kafka topics into tables in the event database named after the stream. The datacenter Kafka topic prefixes are added as a datacenter Hive partition key.
Downstream jobs that use these Hive event tables need to know when an hourly partition has completed. 
==== Problem ====
Problem: If there is no data in one of a stream's composite Kafka topics for a given hour, how can we differentiate between the following scenarios?
# There is a problem somewhere with in the ingestion pipeline, and data failed to be ingested.
# There really was no data produced to the topic during this hour.
==== Solution ====
We solve this problem by producing artificial 'canary' events into every stream. As long as we successfully produce at least one canary event into all of a stream's composite topics every hour, we can be sure that there should be at least one event in every topic every hour. If there are no events for a topic in an hour, then something is broken. Downstream jobs can hope that the hourly Hive partition will be ingested later, and wait until it is.
=== ProduceCanaryEvents ===
The ProduceCanaryEvents job runs as a is scheduled by Airflow several times an hour. It …
* discovers all event streams via EventStreamConfig (that have canary_events_enabled)
* uses the stream's schema examples to create an artificial canary event for the stream (with meta.domain set to 'canary'), 
* POSTs these events to the destination_event_service (eventgate) in each datacenter.
This means that all streams will have artificial canary events in them, and consumers must intentionally filter these events out. (e.g. where meta.domain != 'canary')
=== See also ===
* ✅ - Enable canary events for all streams
* - Send a critical alert to data-engineering if produce_canary_events isn't running correctly canary events are now scheduled by airflow. Alerting is handled by airflow.
* ✅ - Automate ingestion and refinement into Hive of event data from Kafka using stream configs and canary/heartbeat events
* ✅ - Refine event pipeline at this time refines data in hourly partitions without knowing if the partition is complete
Category:Data pipelines
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Hive

Apache Hive is an abstraction built on top of MapReduce that allows SQL to be used on various file formats stored in HDFS. WMF's first use case was to enable querying of unsampled webrequest logs.
As of January 2025, we are running Hive 2.3.6.
__TOC__
== Deprecation ==
Querying on Hive (also called Hive-on-MapReduce) has been officially deprecated since 2015. As of January 2025, there are general plans, but no specific schedules, for deprecating and then removing Wmfdata's Hive module (T384541) and removing the Hive CLI (T379385).
=== What to use instead ===
For most or all queries, Spark should be a drop-in replacement for Hive. You can also use Presto, but it has some syntax differences with Hive.
There is currently no drop-in replacement for Wmfdata's hive.load_csv, but one will be added soon.
== Use ==
The Hive command-line interface is available on all the analytics clients. Here's how to use it:
 
$ kinit
$ hive
(hive takes a few seconds to start. You should see a prompt like hive (default)> when it’s ready.)
Hive can also be easily accessed through Python using wmfdata-python
#!/usr/bin/env python3
import wmfdata as wmf
# Returns a Pandas dataframe
wmf.hive.run("SHOW TABLES FROM event")
== Access ==
In order to access Hive, you need an account with production shell access in the analytics-privatedata-users user group. For more instructions, see Data Platform/Data access.
Some of the data in Hive, like the webrequest logs, are private data so only analytics-privatedata-users can access it. If you are requesting access to Hive, you probably want to be in this group.
== Querying ==
* Data Platform/Systems/Hive/Queries (includes a FAQ about common tasks and problems)
* Data Platform/Systems/Hive/QueryUsingUDF
While hive supports SQL, there are some differences: see the Hive Language Manual for more info.
== Maintained tables ==
(see also Data Platform/Data Lake)
* Webrequest (raw and refined)
* pageview_hourly
* projectview_hourly
* pagecounts_all_sites
* mediacounts
* mobile_apps_session_metrics
* EventLogging data, in the event database (details)
=== Notes ===
* The wmf_raw and wmf databases contain Hive tables maintained by Analytics. You can create your own tables in Hive, but please be sure to create them in a different database, preferably one named after your shell username.
* Hive has the ability to map tables on top of almost any data structure. Since webrequest logs are JSON, the Hive tables must be told to use a JSON SerDe to be able to serialize/deserialize to/from JSON. We use the JsonSerDe included with Hive-HCatalog.
* The HCatalog .jar will be automatically added to a Hive client's auxpath. You shouldn't need to think about it.
== Troubleshooting ==
See the FAQ
== Subpages of ==
== References ==
* Early Hive at WMF Researcher Analysis (2013)
* Hive Compression Test
* Hadoop SequenceFile
* Introduction to Hive Partitioning
Category:Hive
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Hive/Alerts

This page provides troubleshooting guidelines and runbooks to use in the event of alerts from our Monitoring systems,
== Hive Metastore Heap Usage ==
A critical alert is triggered when 90% of the available heap has been used on average over the past hour.
A warning alert is triggered when 80% of the available heap has been used on average over the past hour.
Examine the trends via the Hive dashboard in Grafana.
An alert might indicate an issue with a currently running process, or it might indicate that we need to increase the resources available to the metastore.
Category:Hive
Category:Data platform
Category:Data platform systems
Data Platform/Systems/Hive/Avro

This describes how to do evolution of avro schemas in hive. 
== Intro ==
Despite of what you might read here: https://cwiki.apache.org/confluence/display/Hive/AvroSerDe we have found that avro in hive has several gotchas. Thus we are documenting what has worked for us.
== Create tables with avro schema ==
We have found that easies is to use the avro.schema.literal. property.
Example of how to create a table (if schema is too large you might need to remove all whitespace)
CREATE TABLE nuria_avro
 ROW FORMAT SERDE
 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
 STORED as INPUTFORMAT
 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
 OUTPUTFORMAT
 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
 TBLPROPERTIES (
'avro.schema.literal'='
{
 "type": "record",
 "namespace": "org.wikimedia.analytics.schemas",
 "name": "testTable",
 "fields": [
 {"name": "userName", "type": "string", "default":""},
 {"name": "favouriteNumber", "type": "int", "default":1},
 {"name": "interests", "type": {"type": "array", "items": "string"}, "default":[]}
 ]
}');
== Update schema ==
Updating schema requires updating the TBLPROPERTY. Please note that optional fields in avro have to be defined a certain way
using unions, see specifications for details: 
ALTER TABLE nuria_avro SET TBLPROPERTIES ('avro.schema.literal'='
{
 "type": "record",
 "namespace": "org.wikimedia.analytics.schemas",
 "name": "testTable",
 "fields": [
 {"name": "userName", "type": "string", "default":""},
 {"name":"lastName", "type":["string", "null"], "default":""},
 {"name": "favouriteNumber", "type": "int", "default":1},
 {"name": "interests", "type": {"type": "array", "items": "string"}, "default":[]}
 ]
}');
== External Tables ==
When interacting with data that is not created within hive, an external table needs to be created to point at that data.
CREATE EXTERNAL TABLE `avro_test`
PARTITIONED BY ( 
 `year` string, 
 `month` string, 
 `day` string, 
 `hour` string)
ROW FORMAT SERDE 
 'org.apache.hadoop.hive.serde2.avro.AvroSerDe' 
STORED AS INPUTFORMAT 
 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' 
OUTPUTFORMAT 
 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
LOCATION
 'hdfs://analytics-hadoop/wmf/data/raw/mediawiki/mediawiki_testTable/hourly'
TBLPROPERTIES (
 'avro.schema.literal'='{
 "type": "record",
 "namespace": "org.wikimedia.analytics.schemas",
 "name": "testTable",
 "fields": [
 {"name": "userName", "type": "string", "default":""},
 {"name":"lastName", "type":["string", "null"], "default":""},
 {"name": "favouriteNumber", "type": "int", "default":1}
 ]
 }'
)
== Mediawiki and Avro ==
Analytics/Cluster/MediaWiki_Avro_Logging
Category:Hive
Category:Data platform
Category:Data platform systems