Ottomata has uploaded a new change for review. https://gerrit.wikimedia.org/r/225485
Change subject: POC for Realtime Trending Pageviews ...................................................................... POC for Realtime Trending Pageviews This job was originally developed at https://gerrit.wikimedia.org/r/#/c/201474/ It was modified to use spark 1.3.0 Kafka Direct Stream API To run spark streaming job: ``` spark-submit \ --master yarn \ --num-executors 12 \ --class org.wikimedia.analytics.refinery.job.RealtimeTrendingPages \ refinery-job/target/refinery-job-0.0.15-SNAPSHOT.jar \ 10 600 10 otto-pagecounts-spark-streaming-$(date +%s) webrequest_text \ > /tmp/topPageViews.out ``` To run Visualization: ``` cd realtime_trending_viz npm install nodejs index.js ``` To view visualization from stat1002: ``` ssh -N bast1001.wikimedia.org -L 3000:stat1002.eqiad.wmnet:3000 ``` Then browse to http://localhost:3000 Change-Id: I62b50f052cdc66144854a16696389a439b2c31e2 --- A realtime_trending_viz/index.html A realtime_trending_viz/index.js A realtime_trending_viz/static/cloud.js A realtime_trending_viz/static/d3.layout.cloud.js M refinery-job/pom.xml A refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RealtimeTrendingPages.scala 6 files changed, 946 insertions(+), 0 deletions(-) git pull ssh://gerrit.wikimedia.org:29418/analytics/refinery/source refs/changes/85/225485/1 diff --git a/realtime_trending_viz/index.html b/realtime_trending_viz/index.html new file mode 100644 index 0000000..03f3c95 --- /dev/null +++ b/realtime_trending_viz/index.html @@ -0,0 +1,127 @@ +<!doctype html> +<html> + <head> + <title>Wikimedia Pageviews</title> + <link rel='stylesheet' type='text/css' + href='//cdnjs.cloudflare.com/ajax/libs/rickshaw/1.5.1/rickshaw.min.css'> + + <style> + * { margin: 0; padding: 0; } + html, body { width: 100%; height: 100%; } + .top, .bottom { width: 100%; height: 100%; left: 0; right: 0; position: fixed; } + .top { height: 52%; top: 8%; } + .bottom { height: 40%; bottom: 0; } + + .header { position: fixed; top: 10px; left: 20px; } + .header span:nth-child(1) { font-size: 18pt; } + .header span:nth-child(2) { color: #777; } + </style> + </head> + <body> + <section class="header"> + <span data-bind="text: hoverText, style: {color: articleColor}"></span> + <span data-bind="text: hoverInfo"></span> + </section> + <section class="top"> + <div class="cloud"></div> + </section> + <section class="bottom"> + <div class="total"></div> + </section> + + <script src='/socket.io/socket.io.js'></script> + <!--<script src="https://cdn.socket.io/socket.io-1.2.0.js";></script>--> + <script src='//cdnjs.cloudflare.com/ajax/libs/d3/3.5.5/d3.min.js'></script> + <script src='/static/d3.layout.cloud.js'></script> + <script src='//cdnjs.cloudflare.com/ajax/libs/knockout/3.2.0/knockout-min.js'></script> + <script src='//cdnjs.cloudflare.com/ajax/libs/jquery/2.1.3/jquery.min.js'></script> + <script src='//cdnjs.cloudflare.com/ajax/libs/rickshaw/1.5.1/rickshaw.min.js'></script> + <script src='/static/cloud.js'></script> + + <script> + var topTitle = 'Top 10 Articles'; + var topSubtitle = 'all wikis, last 10 minutes'; + var topTitleColor = '#2ca02c'; + var vm = { + hoverText: ko.observable(topTitle), + hoverInfo: ko.observable(topSubtitle), + articleColor: ko.observable(topTitleColor) + }; + ko.applyBindings(vm); + + var socket = io(); + + socket.on('total', function(newTotal){ + graph.series.addData({ one: +newTotal }); + graph.render(); + }); + + // Total count of all pageviews graph + var graph = new Rickshaw.Graph( { + element: $('.total')[0], + width: $('.bottom').innerWidth(), + height: $('.bottom').innerHeight(), + renderer: 'line', + series: new Rickshaw.Series.FixedDuration([{ + name: 'Total Pageviews' + }], undefined, { + timeInterval: 1000, + maxDataPoints: 100 + }) + } ); + graph.series[0].color = topTitleColor; + + var xAxis = new Rickshaw.Graph.Axis.Time({ graph: graph }); + var yAxis = new Rickshaw.Graph.Axis.Y({ + graph: graph, + orientation: 'right', + tickFormat: Rickshaw.Fixtures.Number.formatKMBT + }); + + var hoverDetail = new Rickshaw.Graph.HoverDetail( { + graph: graph + } ); + + graph.render(); + xAxis.render(); + yAxis.render(); + + + // render the top 10 cloud + var top10cloud = cloudLayout( + '.cloud', + 12, 60, + $('.top').innerWidth(), $('.top').innerHeight(), + function (d) { + vm.hoverText(d.text); + vm.articleColor(top10cloud.fillScale(d.article)); + vm.hoverInfo(d.count + ' views, last 10 minutes'); + }, + function (d) { + vm.hoverText(topTitle); + vm.articleColor(topTitleColor); + vm.hoverInfo(topSubtitle); + } + ); + + socket.on('tops', function(articles){ + var words = articles.split(';').map(function (article){ + var a = article.split('='); + return {article: decodeURIComponent(a[0]).replace(/_/g,' '), count: a[1]}; + }); + var counts = words.map(function(a){ return +a.count; }); + top10cloud.fontSizeScale.domain(d3.extent(counts)); + + top10cloud.stop().words(words).start(); + }); + + // really cool: + function downloadSVG() { + d3.select(this).attr("href", "data:image/svg+xml;charset=utf-8;base64," + btoa(unescape(encodeURIComponent( + svg.attr("version", "1.1") + .attr("xmlns", "http://www.w3.org/2000/svg";) + .node().parentNode.innerHTML)))); + } + </script> + </body> +</html> diff --git a/realtime_trending_viz/index.js b/realtime_trending_viz/index.js new file mode 100644 index 0000000..1fd87aa --- /dev/null +++ b/realtime_trending_viz/index.js @@ -0,0 +1,25 @@ +var express = require('express'); +var path = require('path'); +var app = express(); +var http = require('http').Server(app); +var io = require('socket.io')(http); + +app.use('/static', express.static(path.join(__dirname, 'static'))); + +app.get('/', function(req, res){ + console.log(req.route); + res.sendFile(path.join(__dirname, 'index.html')); +}); + +app.get('/update/:type/:msg', function(req, res){ + console.log(new Date(), req.params); + io.emit(req.params.type, req.params.msg); + res.end(); +}); + +//io.on('connection', function(socket){ +//}); + +http.listen(3000, function(){ + console.log('listening on *:3000'); +}); diff --git a/realtime_trending_viz/static/cloud.js b/realtime_trending_viz/static/cloud.js new file mode 100644 index 0000000..f8da65d --- /dev/null +++ b/realtime_trending_viz/static/cloud.js @@ -0,0 +1,108 @@ +function cloudLayout(selector, smallFont, largeFont, w, h, hoverinCallback, hoveroutCallback) { + var fill = d3.scale.category20b(); + + var fontSize = d3.scale.linear().range([smallFont, largeFont]).clamp(true), + container = d3.select(selector); + + function progress() { return; } + function getArticle(d) { return d.article; } + + var a = 0; + function draw(data, bounds) { + + var wordsJoin = data.map(getArticle).sort().join(''); + var oldWords = vis.selectAll('a').data(); + var oldWordsJoin = oldWords.map(getArticle).sort().join(''); + + var newWords = wordsJoin !== oldWordsJoin; + + var scale = bounds ? Math.min( + w / Math.abs(bounds[1].x - w / 2), + w / Math.abs(bounds[0].x - w / 2), + h / Math.abs(bounds[1].y - h / 2), + h / Math.abs(bounds[0].y - h / 2)) / 2 : 1; + + var text = vis.selectAll("a") + .data(data, function(d) { return d.text; }); + + var animateText = text.transition().duration(1000); + + if (newWords) { + animateText.attr("transform", function(d) { return "translate(" + [d.x, d.y] + ")rotate(" + d.rotate + ")"; }) + + text.enter().append('a') + .attr('xlink:href', function(d) { return d.article; }) + .attr('target', '_blank') + .attr("text-anchor", "middle") + .attr("transform", function(d) { return "translate(" + [d.x, d.y] + ")rotate(" + d.rotate + ")"; }) + .style("font-size", "1px") + .transition() + .duration(1000) + .style("font-size", function(d) { return d.size + "px"; }); + + text.style("font-family", function(d) { return d.font; }) + .style("fill", function(d) { return fill(d.article); }) + .append('text').text(function(d) { return d.text; }) + .on('mouseenter', hoverinCallback) + .on('mouseleave', hoveroutCallback); + + var exitGroup = background.append("g") + .attr("transform", vis.attr("transform")); + + var exitGroupNode = exitGroup.node(); + + text.exit().each(function() { + exitGroupNode.appendChild(this); + }); + exitGroup.transition() + .duration(1000) + .style("opacity", 1e-6) + .remove(); + vis.transition() + .delay(1000) + .duration(750) + .attr("transform", "translate(" + [w >> 1, h >> 1] + ")scale(" + scale + ")"); + } + + animateText.style("font-size", function(d) { return d.size + "px"; }); + } + + var layout = d3.layout.cloud() + .timeInterval(10) + .rotate(function (d) { return 0; }) + .size([w, h]) + .fontSize(function(d) { return fontSize(+d.count); }) + .text(function(d) { + var link = d.article, + wiki = link.indexOf('/wiki/'), + text = link.substr(wiki >= 0 ? wiki + 6 : 0); + return text; + }) + .on('word', progress) + .on('end', draw) + .font('Impact') + .spiral('archimedean'); + + layout.fontSizeScale = fontSize; + layout.fillScale = fill; + + var svg = container.append('svg') + .attr('width', w) + .attr('height', h), + background = svg.append('g'), + vis = svg.append('g') + .attr('transform', 'translate(' + [w >> 1, h >> 1] + ')'); + + return layout; +} + + +function downloadSVG() { + d3.select(this).attr('href', 'data:image/svg+xml;charset=utf-8;base64,' + btoa(unescape(encodeURIComponent( + svg.attr('version', '1.1') + .attr('xmlns', 'http://www.w3.org/2000/svg') + .node().parentNode.innerHTML)))); +} + + +//var tops = setInterval(function (){$.get('http://localhost:3000/update/tops/http%3A%2F%2Fde.wikipedia.org%2Fwiki%2FHelge_Schneider%3D' + (1017 + Math.random()*100) + '%3Bhttp%3A%2F%2Fes.wikipedia.org%2Fwiki%2FEj%C3%A9rcito_Nacional_de_Venezuela%3D' + (958 + Math.random()*100) + '%3Bhttp%3A%2F%2Fes.wikipedia.org%2Fwiki%2FCayo_Mario%3D' + (927 + Math.random()*100) + '%3Bhttp%3A%2F%2Fes.wikipedia.org%2Fwiki%2FMichelle_Bachelet%3D' + (862 + Math.random()*100) + '%3Bhttp%3A%2F%2Fen.wikipedia.org%2Fwiki%2FList_of_Avengers_Assemble_episodes%3D' + (720 + Math.random()*100) + '%3Bhttp%3A%2F%2Fen.wikipedia.org%2Fwiki%2FAcademy_Awards%3D' + (575 + Math.random()*100) + '%3Bhttp%3A%2F%2Fen.wikipedia.org%2Fwiki%2F87th_Academy_Awards%3D' + (472 + Math.random()*100) + '%3Bhttp%3A%2F%2Fen.m.wikipedia.org%2Fwiki%2FAngelsberg%3D' + (394 + Math.random()*100) + '%3Bhttp%3A%2F%2Fja.wikipedia.org%2Fwiki%2F%E5%AE%89%E8%97%A4%E7%99%BE%E7%A6%8F%3D' + (386 + Math.random()*100) + '%3Bhttp%3A%2F%2Fes.wikipedia.org%2Fwiki%2FLlanos%3D' + (374 + Math.random()*100));}, 5000); diff --git a/realtime_trending_viz/static/d3.layout.cloud.js b/realtime_trending_viz/static/d3.layout.cloud.js new file mode 100644 index 0000000..f035837 --- /dev/null +++ b/realtime_trending_viz/static/d3.layout.cloud.js @@ -0,0 +1,401 @@ +// Word cloud layout by Jason Davies, http://www.jasondavies.com/word-cloud/ +// Algorithm due to Jonathan Feinberg, http://static.mrfeinberg.com/bv_ch03.pdf +(function() { + function cloud() { + var size = [256, 256], + text = cloudText, + font = cloudFont, + fontSize = cloudFontSize, + fontStyle = cloudFontNormal, + fontWeight = cloudFontNormal, + rotate = cloudRotate, + padding = cloudPadding, + spiral = archimedeanSpiral, + words = [], + timeInterval = Infinity, + event = d3.dispatch("word", "end"), + timer = null, + cloud = {}; + + cloud.start = function() { + var board = zeroArray((size[0] >> 5) * size[1]), + bounds = null, + n = words.length, + i = -1, + tags = [], + data = words.map(function(d, i) { + d.text = text.call(this, d, i); + d.font = font.call(this, d, i); + d.style = fontStyle.call(this, d, i); + d.weight = fontWeight.call(this, d, i); + d.rotate = rotate.call(this, d, i); + d.size = ~~fontSize.call(this, d, i); + d.padding = padding.call(this, d, i); + return d; + }).sort(function(a, b) { return b.size - a.size; }); + + if (timer) clearInterval(timer); + timer = setInterval(step, 0); + step(); + + return cloud; + + function step() { + var start = +new Date, + d; + while (+new Date - start < timeInterval && ++i < n && timer) { + d = data[i]; + d.x = (size[0] * (Math.random() + .5)) >> 1; + d.y = (size[1] * (Math.random() + .5)) >> 1; + cloudSprite(d, data, i); + if (d.hasText && place(board, d, bounds)) { + tags.push(d); + event.word(d); + if (bounds) cloudBounds(bounds, d); + else bounds = [{x: d.x + d.x0, y: d.y + d.y0}, {x: d.x + d.x1, y: d.y + d.y1}]; + // Temporary hack + d.x -= size[0] >> 1; + d.y -= size[1] >> 1; + } + } + if (i >= n) { + cloud.stop(); + event.end(tags, bounds); + } + } + } + + cloud.stop = function() { + if (timer) { + clearInterval(timer); + timer = null; + } + return cloud; + }; + + cloud.timeInterval = function(x) { + if (!arguments.length) return timeInterval; + timeInterval = x == null ? Infinity : x; + return cloud; + }; + + function place(board, tag, bounds) { + var perimeter = [{x: 0, y: 0}, {x: size[0], y: size[1]}], + startX = tag.x, + startY = tag.y, + maxDelta = Math.sqrt(size[0] * size[0] + size[1] * size[1]), + s = spiral(size), + dt = Math.random() < .5 ? 1 : -1, + t = -dt, + dxdy, + dx, + dy; + + while (dxdy = s(t += dt)) { + dx = ~~dxdy[0]; + dy = ~~dxdy[1]; + + if (Math.min(dx, dy) > maxDelta) break; + + tag.x = startX + dx; + tag.y = startY + dy; + + if (tag.x + tag.x0 < 0 || tag.y + tag.y0 < 0 || + tag.x + tag.x1 > size[0] || tag.y + tag.y1 > size[1]) continue; + // TODO only check for collisions within current bounds. + if (!bounds || !cloudCollide(tag, board, size[0])) { + if (!bounds || collideRects(tag, bounds)) { + var sprite = tag.sprite, + w = tag.width >> 5, + sw = size[0] >> 5, + lx = tag.x - (w << 4), + sx = lx & 0x7f, + msx = 32 - sx, + h = tag.y1 - tag.y0, + x = (tag.y + tag.y0) * sw + (lx >> 5), + last; + for (var j = 0; j < h; j++) { + last = 0; + for (var i = 0; i <= w; i++) { + board[x + i] |= (last << msx) | (i < w ? (last = sprite[j * w + i]) >>> sx : 0); + } + x += sw; + } + delete tag.sprite; + return true; + } + } + } + return false; + } + + cloud.words = function(x) { + if (!arguments.length) return words; + words = x; + return cloud; + }; + + cloud.size = function(x) { + if (!arguments.length) return size; + size = [+x[0], +x[1]]; + return cloud; + }; + + cloud.font = function(x) { + if (!arguments.length) return font; + font = d3.functor(x); + return cloud; + }; + + cloud.fontStyle = function(x) { + if (!arguments.length) return fontStyle; + fontStyle = d3.functor(x); + return cloud; + }; + + cloud.fontWeight = function(x) { + if (!arguments.length) return fontWeight; + fontWeight = d3.functor(x); + return cloud; + }; + + cloud.rotate = function(x) { + if (!arguments.length) return rotate; + rotate = d3.functor(x); + return cloud; + }; + + cloud.text = function(x) { + if (!arguments.length) return text; + text = d3.functor(x); + return cloud; + }; + + cloud.spiral = function(x) { + if (!arguments.length) return spiral; + spiral = spirals[x + ""] || x; + return cloud; + }; + + cloud.fontSize = function(x) { + if (!arguments.length) return fontSize; + fontSize = d3.functor(x); + return cloud; + }; + + cloud.padding = function(x) { + if (!arguments.length) return padding; + padding = d3.functor(x); + return cloud; + }; + + return d3.rebind(cloud, event, "on"); + } + + function cloudText(d) { + return d.text; + } + + function cloudFont() { + return "serif"; + } + + function cloudFontNormal() { + return "normal"; + } + + function cloudFontSize(d) { + return Math.sqrt(d.value); + } + + function cloudRotate() { + return (~~(Math.random() * 6) - 3) * 30; + } + + function cloudPadding() { + return 1; + } + + // Fetches a monochrome sprite bitmap for the specified text. + // Load in batches for speed. + function cloudSprite(d, data, di) { + if (d.sprite) return; + c.clearRect(0, 0, (cw << 5) / ratio, ch / ratio); + var x = 0, + y = 0, + maxh = 0, + n = data.length; + --di; + while (++di < n) { + d = data[di]; + c.save(); + c.font = d.style + " " + d.weight + " " + ~~((d.size + 1) / ratio) + "px " + d.font; + var w = c.measureText(d.text + "m").width * ratio, + h = d.size << 1; + if (d.rotate) { + var sr = Math.sin(d.rotate * cloudRadians), + cr = Math.cos(d.rotate * cloudRadians), + wcr = w * cr, + wsr = w * sr, + hcr = h * cr, + hsr = h * sr; + w = (Math.max(Math.abs(wcr + hsr), Math.abs(wcr - hsr)) + 0x1f) >> 5 << 5; + h = ~~Math.max(Math.abs(wsr + hcr), Math.abs(wsr - hcr)); + } else { + w = (w + 0x1f) >> 5 << 5; + } + if (h > maxh) maxh = h; + if (x + w >= (cw << 5)) { + x = 0; + y += maxh; + maxh = 0; + } + if (y + h >= ch) break; + c.translate((x + (w >> 1)) / ratio, (y + (h >> 1)) / ratio); + if (d.rotate) c.rotate(d.rotate * cloudRadians); + c.fillText(d.text, 0, 0); + if (d.padding) c.lineWidth = 2 * d.padding, c.strokeText(d.text, 0, 0); + c.restore(); + d.width = w; + d.height = h; + d.xoff = x; + d.yoff = y; + d.x1 = w >> 1; + d.y1 = h >> 1; + d.x0 = -d.x1; + d.y0 = -d.y1; + d.hasText = true; + x += w; + } + var pixels = c.getImageData(0, 0, (cw << 5) / ratio, ch / ratio).data, + sprite = []; + while (--di >= 0) { + d = data[di]; + if (!d.hasText) continue; + var w = d.width, + w32 = w >> 5, + h = d.y1 - d.y0; + // Zero the buffer + for (var i = 0; i < h * w32; i++) sprite[i] = 0; + x = d.xoff; + if (x == null) return; + y = d.yoff; + var seen = 0, + seenRow = -1; + for (var j = 0; j < h; j++) { + for (var i = 0; i < w; i++) { + var k = w32 * j + (i >> 5), + m = pixels[((y + j) * (cw << 5) + (x + i)) << 2] ? 1 << (31 - (i % 32)) : 0; + sprite[k] |= m; + seen |= m; + } + if (seen) seenRow = j; + else { + d.y0++; + h--; + j--; + y++; + } + } + d.y1 = d.y0 + seenRow; + d.sprite = sprite.slice(0, (d.y1 - d.y0) * w32); + } + } + + // Use mask-based collision detection. + function cloudCollide(tag, board, sw) { + sw >>= 5; + var sprite = tag.sprite, + w = tag.width >> 5, + lx = tag.x - (w << 4), + sx = lx & 0x7f, + msx = 32 - sx, + h = tag.y1 - tag.y0, + x = (tag.y + tag.y0) * sw + (lx >> 5), + last; + for (var j = 0; j < h; j++) { + last = 0; + for (var i = 0; i <= w; i++) { + if (((last << msx) | (i < w ? (last = sprite[j * w + i]) >>> sx : 0)) + & board[x + i]) return true; + } + x += sw; + } + return false; + } + + function cloudBounds(bounds, d) { + var b0 = bounds[0], + b1 = bounds[1]; + if (d.x + d.x0 < b0.x) b0.x = d.x + d.x0; + if (d.y + d.y0 < b0.y) b0.y = d.y + d.y0; + if (d.x + d.x1 > b1.x) b1.x = d.x + d.x1; + if (d.y + d.y1 > b1.y) b1.y = d.y + d.y1; + } + + function collideRects(a, b) { + return a.x + a.x1 > b[0].x && a.x + a.x0 < b[1].x && a.y + a.y1 > b[0].y && a.y + a.y0 < b[1].y; + } + + function archimedeanSpiral(size) { + var e = size[0] / size[1]; + return function(t) { + return [e * (t *= .1) * Math.cos(t), t * Math.sin(t)]; + }; + } + + function rectangularSpiral(size) { + var dy = 4, + dx = dy * size[0] / size[1], + x = 0, + y = 0; + return function(t) { + var sign = t < 0 ? -1 : 1; + // See triangular numbers: T_n = n * (n + 1) / 2. + switch ((Math.sqrt(1 + 4 * sign * t) - sign) & 3) { + case 0: x += dx; break; + case 1: y += dy; break; + case 2: x -= dx; break; + default: y -= dy; break; + } + return [x, y]; + }; + } + + // TODO reuse arrays? + function zeroArray(n) { + var a = [], + i = -1; + while (++i < n) a[i] = 0; + return a; + } + + var cloudRadians = Math.PI / 180, + cw = 1 << 11 >> 5, + ch = 1 << 11, + canvas, + ratio = 1; + + if (typeof document !== "undefined") { + canvas = document.createElement("canvas"); + canvas.width = 1; + canvas.height = 1; + ratio = Math.sqrt(canvas.getContext("2d").getImageData(0, 0, 1, 1).data.length >> 2); + canvas.width = (cw << 5) / ratio; + canvas.height = ch / ratio; + } else { + // Attempt to use node-canvas. + canvas = new Canvas(cw << 5, ch); + } + + var c = canvas.getContext("2d"), + spirals = { + archimedean: archimedeanSpiral, + rectangular: rectangularSpiral + }; + c.fillStyle = c.strokeStyle = "red"; + c.textAlign = "center"; + + if (typeof module === "object" && module.exports) module.exports = cloud; + else (d3.layout || (d3.layout = {})).cloud = cloud; +})(); diff --git a/refinery-job/pom.xml b/refinery-job/pom.xml index 83a834c..6c8d803 100644 --- a/refinery-job/pom.xml +++ b/refinery-job/pom.xml @@ -78,6 +78,48 @@ <version>2.0.0</version> </dependency> + + + + + <dependency> + <groupId>org.apache.spark</groupId> + <artifactId>spark-streaming_2.10</artifactId> + <version>${spark.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>org.apache.spark</groupId> + <artifactId>spark-streaming-kafka_2.10</artifactId> + <version>${spark.version}</version> + </dependency> + + <dependency> + <groupId>eu.piotrbuda</groupId> + <artifactId>scalawebsocket_2.10</artifactId> + <version>0.1.1</version> + </dependency> + + <dependency> + <groupId>org.json4s</groupId> + <artifactId>json4s_2.10</artifactId> + <version>3.2.11</version> + </dependency> + + <dependency> + <groupId>org.json4s</groupId> + <artifactId>json4s-jackson_2.10</artifactId> + <version>3.2.11</version> + </dependency> + + <dependency> + <groupId>org.scalaj</groupId> + <artifactId>scalaj-http_2.10</artifactId> + <version>1.1.4</version> + </dependency> + + </dependencies> <build> diff --git a/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RealtimeTrendingPages.scala b/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RealtimeTrendingPages.scala new file mode 100644 index 0000000..478ed16 --- /dev/null +++ b/refinery-job/src/main/scala/org/wikimedia/analytics/refinery/job/RealtimeTrendingPages.scala @@ -0,0 +1,243 @@ +package org.wikimedia.analytics.refinery.job + +import kafka.serializer.StringDecoder + +import org.apache.spark._ +import org.apache.spark.SparkContext._ +import org.apache.spark.streaming._ +import org.apache.spark.streaming.dstream.DStream +import org.apache.spark.streaming.StreamingContext._ +import org.apache.spark.streaming.kafka._ +import org.json4s +import org.json4s.jackson.JsonMethods +import scalaj.http.Http +import org.wikimedia.analytics.refinery.core.PageviewDefinition +import scalawebsocket.WebSocket + +/** + * Consumes messages from one or more topics in Kafka and does X + * Usage: RealtimeTrendingPages <consumer-group> <batch-interval> <window-length> <sliding-interval> <num-partitions> + * <group> is the name of kafka consumer group + * + * Example: + * `$ bin/run-example \ + * org.wikimedia.analytics.refinery.spark.RealtimeTrendingPages \ + * my-consumer-group` 1 10 1 12 + * + */ +object RealtimeTrendingPages { + + // domain model for a Webrequest + case class Webrequest( + hostname: String, + sequence: Long, + dt: String, + time_firstbyte: Double, + ip: String, + cache_status: String, + http_status: String, + response_size: Long, + http_method: String, + uri_host: String, + uri_path: String, + uri_query: String, + content_type: String, + referer: String, + x_forwarded_for: String, + user_agent: String, + accept_language: String, + x_analytics: String, + range: String + ) + + + implicit val formats = json4s.DefaultFormats + + + val mainPages = List( + "Main_Page", + "Wikipedia:Portada", + "Заглавная_страница", + "Pagina_principale", + "Wikipedia:Hauptseite", + "メインページ", + "Hoofdpagina", + "Wikipedia:首页", + "Wikipédia:Página_principal", + "Wikipédia:Accueil_principal", + "위키백과:대문", + "Portal:Huvudsida", + "Syahan_nga_Pakli", + "Unang_Panid", + "Trang_Chính", + "Wikipedia:Strona_główna", + "Portada", + "Головна_сторінка", + "Halaman_Utama", + "صفحهٔ_اصلی", + "Portal:Forside", + "Wikipedia:Etusivu", + "الصفحة_الرئيسية", + "Hlavní_strana", + "Главна_страна", + "Kezdőlap" + ) + + def createWebrequestFromJson(json: String): Webrequest = { + try { + return JsonMethods.parse(json).extract[Webrequest] + } catch { + case _: org.json4s.MappingException => System.err.println(s"ERROR: Failed to parse json: '$json', return empty Webrequest") + return Webrequest("-", 0, "-", 0.0, "-", "-", "-", 0, "-", "-", "-", "-", "-", "-", "-", "-", "-", "-", "-") + } + } + + def isPageview(webrequest: Webrequest) = { + (webrequest.x_analytics.contains("ns=0") && + PageviewDefinition.getInstance().isPageview( + webrequest.uri_host, + webrequest.uri_path, + webrequest.uri_query, + webrequest.http_status, + webrequest.content_type, + webrequest.user_agent + ) + ) + } + + def getQueryParamValue(query: String, key: String): Option[String] = { + if (query.contains(key)) { + val startIndex = query.indexOf(key + "=") + (key + "=").length + var stopIndex = query.indexOf("&", startIndex) + if (stopIndex == -1) { + stopIndex = query.length + } + + return Some(query.substring(startIndex, stopIndex)) + } + + return None + } + + def stripMobile(uri_host: String) = { + val mobileSubdomain = ".m." + if (uri_host.contains(mobileSubdomain)) + uri_host.patch(uri_host.indexOf(mobileSubdomain), ".", mobileSubdomain.length) + else + uri_host + } + + def getNormalizedPageURI(webrequest: Webrequest): Option[String] = { + val title = PageviewDefinition.getInstance(). + getPageTitleFromUri(webrequest.uri_path, webrequest.uri_query) + // THIS filter LOGIC NEEDS TO BE IN A BETTER PLACE! + + if (title.isEmpty() || mainPages.contains(title)) + None + else + Some("http://"; + stripMobile(webrequest.uri_host) + "/wiki/" + title) + } + + + + def printTopPageViews(topPages: Array[(Long, String)]) = { + println("--------------------------------------------------") + topPages.foreach { case (v,k) => println(s"$k\t$v") } + } + + + def sendToWebsocketApp(destination: String, value: String) = { + val uri = s"http://stat1002.eqiad.wmnet:3000/update/$destination/$value"; + System.err.println("INFO: Sending request to " + uri) + try { + Http(uri).proxy("webproxy.eqiad.wmnet",8080).asString + } + catch { + case _: java.net.ConnectException => System.err.println("ERROR: Connection refused at " + uri) + } + } + + + + + + def viewCountsByPageURI( + webrequestStream: DStream[Webrequest], + windowLength: Long, + slidingInterval: Long + ): DStream[Tuple2[String,Long]] = { + + webrequestStream. + // Filter for pageviews only. + filter(isPageview). + // Normalize the uri_* fields into a page URI. + // This also filters out some unwanted pages (Main_Page, etc.) + map(getNormalizedPageURI). + // Remove any unwanted pages (These will be Nones instead of Somes) + flatMap(identity[Option[String]]). + // Count by the normalized page URIs over the last windowLength + // and emit counts every slidingInterval. + countByValueAndWindow(Seconds(windowLength), Seconds(slidingInterval)) + } + + + + def main(args: Array[String]) { + if (args.length < 5) { + System.err.println("Usage: RealtimeTrendingPages <batch-interval> <window-length> <sliding-interval> <consumer-group> <topics> <receiver-count>") + System.exit(1) + } + + val Array( + batchInterval, + windowLength, + slidingInterval, + consumerGroup, + topics + ) = args + + val brokers = "analytics1012.eqiad.wmnet:9092,analytics1018.eqiad.wmnet:9092,analytics1021.eqiad.wmnet:9092,analytics1022.eqiad.wmnet:9092" + + val sparkConf = new SparkConf().setAppName("RealtimeTrendingPages") + val ssc = new StreamingContext(sparkConf, Seconds(batchInterval.toLong)) + ssc.checkpoint("/tmp/RealtimeTrendingPages-spark-streaming-checkpoint") + + + val topicsSet = topics.split(",").toSet + val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers) + + val messageStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder]( + ssc, + kafkaParams, + topicsSet + ) + + val webrequestStream = messageStream. + // Extract the JSON message from the Kafka (Key, Value) message. + map(_._2). + // Create a webrequest object from the JSON message. + map(createWebrequestFromJson) + + viewCountsByPageURI(webrequestStream, windowLength.toLong, slidingInterval.toLong). + map(_.swap). + foreachRDD( + rdd => { + val topPages = rdd.top(10) + printTopPageViews(topPages) + + //Send to socketio node cool thang + val topPagesString = topPages.map(t => java.net.URLEncoder.encode(t._2, "UTF-8") + "=" + t._1).mkString(";") + sendToWebsocketApp("tops", topPagesString) + } + ) + + + + + // print them! + // pageURICounts.foreachRDD(rdd => printTopPageViews(rdd.top(10))) + + ssc.start() + ssc.awaitTermination() + } +} \ No newline at end of file -- To view, visit https://gerrit.wikimedia.org/r/225485 To unsubscribe, visit https://gerrit.wikimedia.org/r/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I62b50f052cdc66144854a16696389a439b2c31e2 Gerrit-PatchSet: 1 Gerrit-Project: analytics/refinery/source Gerrit-Branch: master Gerrit-Owner: Ottomata <[email protected]> _______________________________________________ MediaWiki-commits mailing list [email protected] https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
