Bearloga has uploaded a new change for review. https://gerrit.wikimedia.org/r/235532
Change subject: Initial commit and patch to remove file duplication. ...................................................................... Initial commit and patch to remove file duplication. Brings config.R, common.R, and main.sh into the root dir. Change-Id: I160f0728417bf2a00dc6f5723bf7760496ebac06 --- A .gitignore A README.md A common.R A config.R A main.sh A search/api.R A search/app.R A search/core.py A search/desktop.R A search/mobile.R A wdqs/basic_usage.R 11 files changed, 485 insertions(+), 0 deletions(-) git pull ssh://gerrit.wikimedia.org:29418/wikimedia/discovery/golden refs/changes/32/235532/1 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..45225f3 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +*.RData diff --git a/README.md b/README.md new file mode 100644 index 0000000..85b899b --- /dev/null +++ b/README.md @@ -0,0 +1,12 @@ +Golden Retriever Scripts +======================== + +This repository contains aggregation/acquisition scripts for extracting data from the MySQL/Hive databases. + +- **search** contains scripts for getting usage data on search features, incl. APIs. The generated datasets are accessible at http://datasets.wikimedia.org/aggregate-datasets/search/ +- **wdqs** contains scripts for getting usage data on Wikidata Query Service. The generated datasets are accessible at http://datasets.wikimedia.org/aggregate-datasets/wdqs/ + +## Contact + +- [Oliver Keyes](https://meta.wikimedia.org/wiki/User:Okeyes_(WMF)) +- [Mikhail Popov](https://meta.wikimedia.org/wiki/User:MPopov_(WMF)) diff --git a/common.R b/common.R new file mode 100644 index 0000000..5a39578 --- /dev/null +++ b/common.R @@ -0,0 +1,31 @@ +source("config.R") + +# Dependencies +library(lubridate) +library(olivr) +suppressPackageStartupMessages(library(data.table)) + +# Query building function +query_func <- function(fields, table, ts_field, date = NULL, conditionals){ + + # Ensure we have a date and deconstruct it into a MW-friendly format + if(is.null(date)){ + date <- Sys.Date()-1 + } + date <- gsub(x = date, pattern = "-", replacement = "") + + # Build the query proper (this will work for EL schemas where the field is always 'timestamp') + query <- paste(fields, "FROM", table, "WHERE LEFT(timestamp,8) =", date, "AND", conditionals) + + results <- data.table::as.data.table(olivr::mysql_read(query, "log")) + return(results) +} + +# Conditional write; if the file exists, add x to the end. If it doesn't, write an entirely new file. +conditional_write <- function(x, file){ + if(file.exists(file)){ + write.table(x, file, append = TRUE, sep = "\t", row.names = FALSE, col.names = FALSE) + } else { + write.table(x, file, append = FALSE, sep = "\t", row.names = FALSE) + } +} diff --git a/config.R b/config.R new file mode 100644 index 0000000..f1b524e --- /dev/null +++ b/config.R @@ -0,0 +1,9 @@ +# Config variables and setup: +options(scipen = 500, q = "no") + +# base_path : This is set on a per-script level +# (before sourcing common.R) + +if(!file.exists(base_path)) { + dir.create(path = base_path) +} diff --git a/main.sh b/main.sh new file mode 100644 index 0000000..997892b --- /dev/null +++ b/main.sh @@ -0,0 +1,8 @@ +R CMD BATCH search/desktop.R && +R CMD BATCH search/mobile.R && +R CMD BATCH search/app.R && +R CMD BATCH search/api.R && +R CMD BATCH search/failures.R && +R CMD BATCH wdqs/basic_usage.R +python core.py && +rm -rf .RData diff --git a/search/api.R b/search/api.R new file mode 100644 index 0000000..8904a0e --- /dev/null +++ b/search/api.R @@ -0,0 +1,51 @@ +# Per-file config: +base_path <- "/a/aggregate-datasets/search/" + +source("../common.R") + +# Central function +main <- function(date = NULL){ + + # Date handling + if(is.null(date)){ + date <- Sys.Date() - 1 + } + subquery <- paste0(" WHERE year = ", lubridate::year(date), + " AND month = ", lubridate::month(date), + " AND day = ", lubridate::day(date), " ") + + # Write query and dump to file + query <- paste0("ADD JAR /srv/deployment/analytics/refinery/artifacts/refinery-hive.jar; + CREATE TEMPORARY FUNCTION search_classify AS + 'org.wikimedia.analytics.refinery.hive.SearchClassifierUDF'; + USE wmf; + SELECT year, month, day, search_classify(uri_path, uri_query) AS event_type, + COUNT(*) AS search_events + FROM webrequest + ", subquery, + "AND webrequest_source IN('text','mobile') AND http_status = '200' + GROUP BY year, month, day, search_classify(uri_path, uri_query);") + query_dump <- tempfile() + cat(query, file = query_dump) + + # Query + results_dump <- tempfile() + system(paste0("export HADOOP_HEAPSIZE=1024 && hive -f ", query_dump, " > ", results_dump)) + results <- read.delim(results_dump, sep = "\t", quote = "", as.is = TRUE, header = TRUE) + file.remove(query_dump, results_dump) + + # Filter and reformat + results <- results[complete.cases(results),] + results <- results[results$event_type %in% c("language","cirrus","prefix","geo","open"),] + output <- data.frame(timestamp = as.Date(paste(results$year, results$month, results$day, sep = "-")), + event_type = results$event_type, + events = results$search_events, + stringsAsFactors = FALSE) + + # Write out + conditional_write(output, file.path(base_path, "search_api_aggregates.tsv")) +} + +#Run and kill +main() +q(save = "no") diff --git a/search/app.R b/search/app.R new file mode 100644 index 0000000..7a7c231 --- /dev/null +++ b/search/app.R @@ -0,0 +1,49 @@ +# Per-file config: +base_path <- "/a/aggregate-datasets/search/" + +source("../common.R") + +# Retrieves data for the mobile web stuff we care about, drops it in the aggregate-datasets directory. Should be run on stat1002, /not/ on the datavis machine. + +main <- function(date = NULL, table = "MobileWikiAppSearch_10641988"){ + + # Retrieve data using the query builder in ./common.R + data <- query_func(fields = " + SELECT timestamp, + CASE event_action WHEN 'click' THEN 'clickthroughs' + WHEN 'start' THEN 'search sessions' + WHEN 'results' THEN 'Result pages opened' END AS action, + event_timeToDisplayResults AS load_time, + userAgent", + date = date, + table = table, + conditionals = "event_action IN ('click','start','results')") + data$timestamp <- as.Date(olivr::from_mediawiki(data$timestamp)) + data$platform[grepl(x = data$userAgent, pattern = "Android", fixed = TRUE)] <- "Android" + data$platform[is.na(data$platform)] <- "iOS" + data <- data[,userAgent := NULL,] + + # Generate aggregates and save + app_results <- data[,j = list(events = .N), by = c("timestamp","action", "platform")] + conditional_write(app_results, file.path(base_path, "app_event_counts.tsv")) + + # Produce load time data + load_times <- data[data$action == "Result pages opened",{ + output <- numeric(3) + quantiles <- quantile(load_time,probs=seq(0,1,0.01)) + + output[1] <- round(median(load_time)) + output[2] <- quantiles[95] + output[3] <- quantiles[99] + + output <- data.frame(t(output)) + names(output) <- c("Median","95th percentile","99th Percentile") + output + }, by = c("timestamp","platform")] + conditional_write(load_times, file.path(base_path, "app_load_times.tsv")) + +} + +# Run and kill +main() +q(save = "no") diff --git a/search/core.py b/search/core.py new file mode 100644 index 0000000..610f5c2 --- /dev/null +++ b/search/core.py @@ -0,0 +1,157 @@ +import re +import gzip +import datetime +import os.path +import csv +from collections import Counter, OrderedDict +from sys import exit +from floccus import check +from floccus import misc +from floccus import get + +#Regexes for parsing +is_valid_regex = re.compile("^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}") +has_zero_results_regex = re.compile("Found 0 total results") +execution_id_regex = re.compile("by executor (\d{9,})$") + +#File paths for output +output_daily = "/home/ironholds/zero_results/" +aggregate_filepath = "/a/aggregate-datasets/search/cirrus_query_aggregates.tsv" +breakdown_filepath = "/a/aggregate-datasets/search/cirrus_query_breakdowns.tsv" +suggest_filepath = "/a/aggregate-datasets/search/cirrus_suggestion_breakdown.tsv" + +class BoundedRelatedStatCollector(object): + ''' + Collects related log lines that occur within a provided timedelta of + another log line with the same group_by value. + ''' + def __init__(self, callback, bounds=None): + self.data = OrderedDict() + self.callback = callback + self.bounds = bounds if bounds else datetime.timedelta(seconds=120) + self.visited = 0 + + def push(self, group_by, line, timestamp): + if group_by == None: + self.callback([line]) + return + + if group_by in self.data: + values, maxTimestamp = self.data[group_by] + # delete the key so it moves to the end of the list + del self.data[group_by] + values.append(line) + if timestamp > maxTimestamp: + maxTimestamp = timestamp + else: + values = [line] + maxTimestamp = timestamp + + self.data[group_by] = (values, maxTimestamp) + + self.visited += 1 + if self.visited % 1000 == 0: + flush_up_to = maxTimestamp - self.bounds; + self.flush(flush_up_to) + + def flush(self, max_timestamp=None): + for group_by in self.data: + values, timestamp = self.data[group_by] + if max_timestamp != None and timestamp > max_timestamp: + return + self.callback(values) + del self.data[group_by] + +#Check if a line is even valid +def extract_timestamp(row): + match = is_valid_regex.match(row) + if match: + try: + return datetime.datetime.strptime(match.group(), '%Y-%m-%d %I:%M:%S') + except ValueError: + return None + else: + return None + +def extract_execution_id(row): + match = execution_id_regex.search(row) + if match: + return match.group(1) + else: + return None + +def daily_write(date, zero_results): + with open((output_daily + date + ".tsv"), "ab") as tsv_file: + write_obj = csv.writer(tsv_file, delimiter = "\t") + for line in zero_results: + start = re.sub("(\\t|\\n|\")", "", line[0]) + write_obj.writerow([start, str(line[1])]) + +#For each line in the file, if it's valid, increment the query count. +#If it has zero results, log the query to the query list and increment +#the zero count. +def parse_file(filename): + stats = { + 'queries': 0, + 'zero_result_count': 0, + 'prefix_queries': 0, + 'prefix_zero': 0, + 'full_queries': 0, + 'full_zero': 0, + 'suggested_queries': 0, + 'suggested_zero': 0, + 'zero_result_queries': list(), + } + def count_query(lines): + # Just assume whichever logline showed up last is the one we want + line = lines[-1] + if check.check_prefix_search(line): + stats['queries'] += 1 + stats['prefix_queries'] += 1 + if check.check_zero(line): + stats['prefix_zero'] += 1 + stats['zero_result_queries'].append(get.get_query(line)) + elif check.check_full_search(line): + stats['queries'] += 1 + stats['full_queries'] += 1 + if check.check_zero(line): + stats['full_zero'] += 1 + stats['zero_result_queries'].append(get.get_query(line)) + if check.check_suggestion(line): + stats['suggested_queries'] += 1 + if check.check_zero(line): + stats['suggested_zero'] += 1 + + collector = BoundedRelatedStatCollector(count_query) + connection = gzip.open(filename) + for line in connection: + timestamp = extract_timestamp(line) + if timestamp is not None: + execution_id = extract_execution_id(line) + collector.push(execution_id, line, timestamp) + + connection.close() + collector.flush() + + zero_result_queries = Counter(stats['zero_result_queries']).most_common(100) + high_level_stats = Counter({"Search Queries": stats['queries'], + "Zero Result Queries": stats['prefix_zero'] + stats['full_zero'] + }) + breakdown_stats = Counter({ + "Full-Text Search": float(stats['full_zero'])/stats['full_queries'], + "Prefix Search": float(stats['prefix_zero'])/stats['prefix_queries'] + }) + + suggestion_stats = Counter({ + "Searches with Suggestions": float(stats['suggested_zero'])/stats['suggested_queries'] + }) + return(high_level_stats, breakdown_stats, suggestion_stats, zero_result_queries) + +#Run and write out +filepath, date = misc.get_filepath() +high_level, breakdown, suggests, zero_results = parse_file(filepath) +misc.write_counter(high_level, date, aggregate_filepath) +misc.write_counter(breakdown, date, breakdown_filepath) +misc.write_counter(suggests, date, suggest_filepath) +daily_write(date, zero_results) +exit() diff --git a/search/desktop.R b/search/desktop.R new file mode 100644 index 0000000..fe62996 --- /dev/null +++ b/search/desktop.R @@ -0,0 +1,49 @@ +# Per-file config: +base_path <- "/a/aggregate-datasets/search/" + +source("../common.R") + +# Retrieves data for the desktop stuff we care about, drops it in the aggregate-datasets directory. Should be run on stat1002, /not/ on the datavis machine. + +main <- function(date = NULL, table = "Search_12057910"){ + + # Get data and format + data <- query_func(fields = " + SELECT timestamp, + CASE event_action WHEN 'click-result' THEN 'clickthroughs' + WHEN 'session-start' THEN 'search sessions' + WHEN 'impression-results' THEN 'Result pages opened' + WHEN 'submit-form' THEN 'Form submissions' END AS action, + event_clickIndex AS click_index, + event_numberOfResults AS result_count, + event_resultSetType as result_type, + event_timeOffsetSinceStart AS time_offset, + event_timeToDisplayResults AS load_time + ", + date = date, + table = table, + conditionals = "event_action IN ('click-result','session-start','impression-results', 'submit-form')") + data$timestamp <- as.Date(olivr::from_mediawiki(data$timestamp)) + + # Generate aggregates and save + results <- data[,j = list(events = .N), by = c("timestamp","action")] + conditional_write(results, file.path(base_path, "desktop_event_counts.tsv")) + + # Generate load time data and save that + load_times <- data[data$action == "Result pages opened",{ + output <- numeric(3) + quantiles <- quantile(load_time,probs=seq(0,1,0.01)) + output[1] <- round(median(load_time)) + output[2] <- quantiles[95] + output[3] <- quantiles[99] + + output <- data.frame(t(output)) + names(output) <- c("Median","95th percentile","99th Percentile") + output + }, by = "timestamp"] + conditional_write(load_times, file.path(base_path, "desktop_load_times.tsv")) + return(invisible()) +} + +main() +q(save = "no") diff --git a/search/mobile.R b/search/mobile.R new file mode 100644 index 0000000..61dc9a4 --- /dev/null +++ b/search/mobile.R @@ -0,0 +1,51 @@ +# Per-file config: +base_path <- "/a/aggregate-datasets/search/" + +source("../common.R") + +# Retrieves data for the mobile web stuff we care about, drops it in the public-datasets directory. Should be run on stat1002, /not/ on the datavis machine. + +main <- function(date = NULL, table = "MobileWebSearch_12054448"){ + + # Get data and format the timestamps + data <- query_func(fields = " + SELECT timestamp, + CASE event_action WHEN 'click-result' THEN 'clickthroughs' + WHEN 'session-start' THEN 'search sessions' + WHEN 'impression-results' THEN 'Result pages opened' END AS action, + event_clickIndex AS click_index, + event_numberOfResults AS result_count, + event_resultSetType as result_type, + event_timeOffsetSinceStart AS time_offset, + event_timeToDisplayResults AS load_time, + event_platformVersion AS version", + date = date, + table = table, + conditionals = "event_action IN ('click-result','session-start','impression-results')") + data$timestamp <- as.Date(olivr::from_mediawiki(data$timestamp)) + + # Convert it into event aggregates and write out + mobile_results <- data[,j = list(events = .N), by = c("timestamp","action")] + conditional_write(mobile_results, file.path(base_path, "mobile_event_counts.tsv")) + + # Process load times and write out + load_times <- data[data$action == "Result pages opened",{ + output <- numeric(3) + quantiles <- quantile(load_time,probs=seq(0,1,0.01)) + + output[1] <- round(median(load_time)) + output[2] <- quantiles[95] + output[3] <- quantiles[99] + + output <- data.frame(t(output)) + names(output) <- c("Median","95th percentile","99th Percentile") + output + }, by = "timestamp"] + conditional_write(load_times, file.path(base_path, "mobile_load_times.tsv")) + return(invisible()) +} + +main() +q(save = "no") + +# dates <- seq(as.Date("2015-06-11"), as.Date("2015-06-17"), by = "date") diff --git a/wdqs/basic_usage.R b/wdqs/basic_usage.R new file mode 100644 index 0000000..d474f66 --- /dev/null +++ b/wdqs/basic_usage.R @@ -0,0 +1,67 @@ +# Per-file config: +base_path <- "/a/aggregate-datasets/wdqs/" + +source("../common.R") + +# Retrieves data for the WDQS stuff we care about, drops it in the aggregate-datasets directory. Should be run on stat1002, /not/ on the datavis machine. + +# Create a script that would produce raw data on usage of +# - query.wikidata.org +# - SPARQL endpoint: query.wikidata.org/bigdata/namespace/wdq/sparql + +# Central function +main <- function(date = NULL) { + + # Date handling + if(is.null(date)) { + date <- Sys.Date() - 1 + } + subquery <- paste0(" WHERE year = ", lubridate::year(date), + " AND month = ", lubridate::month(date), + " AND day = ", lubridate::day(date), " ") + + # Write query and dump to file + query <- paste0("USE wmf; + SELECT year, month, day, + FIND_IN_SET(uri_path, '/bigdata/namespace/wdq/sparql,/,/index.php') AS uri_path, + IF(INSTR(uri_query, 'query') > 0, 'query', 'other') AS uri_query, + IF(INSTR(content_type, 'sparql-results') > 0, 'sparql results', 'other') AS content_type, + COUNT(*) AS n + FROM webrequest", + subquery, + "AND webrequest_source = 'misc' AND FIND_IN_SET(http_status, '200,301,302,303') > 0 + GROUP BY year, month, day, + FIND_IN_SET(uri_path, '/bigdata/namespace/wdq/sparql,/,/index.php'), + IF(INSTR(uri_query, 'query') > 0, 'query', 'other'), + IF(INSTR(content_type, 'sparql-results') > 0, 'sparql results', 'other');") + query_dump <- tempfile() + cat(query, file = query_dump) + + # Query + results_dump <- tempfile() + system(paste0("export HADOOP_HEAPSIZE=1024 && hive -f ", query_dump, " > ", results_dump)) + results <- read.delim(results_dump, sep = "\t", quote = "", as.is = TRUE, header = TRUE) + file.remove(query_dump, results_dump) + + results$uri_path <- factor(results$uri_path, 0:3, c("other", "/bigdata/namespace/wdq/sparql", "/", "/index.php")) + + output <- data.frame(timestamp = as.Date(paste(results$year, results$month, results$day, sep = "-")), + path = results$uri_path, + query = results$uri_query, + content = results$content_type, + events = results$n, + stringsAsFactors = FALSE) + + # Write out + conditional_write(output, file.path(base_path, "wdqs_aggregates.tsv")) + +} + +# Backlog (start date: 2015-07-28): +# backlog <- function(days) { +# for (i in days:1) try(main(Sys.Date() - i), silent = TRUE) +# }; backlog(30) # as of 2015-08-27 + +# Run and kill +main() +q(save = "no") -- To view, visit https://gerrit.wikimedia.org/r/235532 To unsubscribe, visit https://gerrit.wikimedia.org/r/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I160f0728417bf2a00dc6f5723bf7760496ebac06 Gerrit-PatchSet: 1 Gerrit-Project: wikimedia/discovery/golden Gerrit-Branch: master Gerrit-Owner: Bearloga <[email protected]> _______________________________________________ MediaWiki-commits mailing list [email protected] https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
