大道至简，让简单的归于简单，纯碎的归于纯碎。这是我在简书上的第一篇文章，希望在这个平台上宾主双方彼此都会愉快。
闲着也是闲着，但缺着不能一直缺着，所以还是要补上这一篇，哪怕是有一点阻力。前面其它的系列文章可以参阅我的另一处文集。
仪表板是数据的简要表示，应用的场景不少，人们都喜欢这种形式，大屏上一刷，高大上的样子。要做好事情当然要踏踏实实地干，分析建模的苦功在水下，人们看不到也看不懂。但要讲好故事也要做好表面功夫，一个精美的仪表板往往能得到人们的认同，所以也不差那么几百行代码了。
R markdown通过 flexdashboard包支持仪表板，它是R markdown的一个扩充。Shiny也通过 shinydashboard包提供仪表板的支持，本篇先介绍flexdashboard的一个实例，因为篇幅的关系，将在下一篇中再介绍shinydashboard的实现，以便比较一下。
本篇不介绍flexdashboard包的具体用法，可参阅《 R Markdown: The Definitive Guide》一书的第5章 《Dashboards》。
本例通过从Github上部署的服务（项目）读取Rstudio CRAN上R软件包的下载日志，监控下载流量的变化，实时监控是仪表板的典型用法，就像驾驶位上的仪表盘一样。这里是延迟1周的备份下载日志，不过作为演示也足够了。 项目源码在github上，J.J.Allaire的作品，简要精当（实际上是Rstudio三巨头合作的示例，bubbles包与扫描日志的服务器端代码是CTO Joe Cheng，shinySignals包是首席科学家Hadley Wickham）。
更多的flexdashboard例子请看这里。
一、先看看运行效果，界面已经汉化了。
这个仪表板有两个维度的动态，一是数据每秒更新一次，二是调整左边的两个反应式输入变量，会动态的改变右边仪表板的显示。
1、开始的时候，下载流量小于50次/秒时，流量表的颜色是绿色的。
2、调整左边滑杆的值，大于30次/秒就提醒，流量表的颜色变成了橙色。
3、最近下载列表显示的行数，由左边的一个反应式数字输入来调整。
这是调整前的，显示50行。
这是调整后的，显示10行。
二、flexdashboard的仪表板组件。
这个例子很简单，但已经具备了典型仪表板的所有要素。一个flexdashboard仪表板可以有7种组件，如上所见。
1、基于 HTML 小部件的交互式 JavaScript 数据可视化图形。上面的泡泡图就是一个htmlwidget bubbles，生成了一个单旋臂星系图，更多的htmlwidget可以看 这里，提供了各种各样丰富的可视化工具。
2、R 图形，包括基础、栅栏和网格图形。上面例子中没有，R markdown代码块中R绘图的输出，前面的系列文章中很多了。
3、表格，如上面的下载百分比列表与最近下载列表（本例中是Shiny渲染输出）。
前面的3种组件是R markdown文档中通用的，后面的4中组件则是flexdashboard独有的。
4、数值框（展示重要数据），如上图中顶端蓝色的下载总数与下载的用户数两个指标。
5、仪表盘，如上图中的流量表。
6、文本注释，如上图中“数据每秒更新一次”的说明，它有特定的语法，用">"开头。
7、导航栏（提供与仪表板相关的更多链接），如上图中最右上角的“源码”链接。
左边的反应式输入变量，滑杆与数字，是Shiny的反应式组件，它们不是flexdashboard的组件。flexdashboard仪表板可以是一个动态的Shiny R markdown文档，如本例所见，这提供了更好的交互性和动态。
三、仪表板源码
源码有2个文件，一个仪表板R markdown文件和一个从服务器读取数据的R函数文件。
1、dashboard.Rmd
先看Rmd文件的YAML头，输出类型是flexdashboard::flex_dashboard，实质上是HTML页面。然后设置了它的样式主题是cosmo，宇宙。flexdashboard内置的主题有“default”, “bootstrap”, “cerulean”, “cosmo”, “darkly”, “flatly”, “journal”, “lumen”, “paper”, “readable”, “sandstone”, “simplex”, “spacelab”, “united”, “yeti”，如果想用其他更多的样式主题，可以到Bootswatch上看看，上面各主题的具体颜色配置，也可以到该站上看看。比如要设置成spacelab主题，用下面代码块中注释的部分代替即可。有关flexdashboard样式主题的设置，可以参阅这篇文章。这些样式的源码在这里，也可以参考定义自己的样式。
上面的第7种flexdashboard组件，导航栏，在YAML头中用navbar定义。
它的runtime是shiny，这是一个交互式的Shiny R markdown文档，要部署在Shiny Server上，当然笔记本上单机运行也可以。
---
title: "CRAN 下载监控"
output:
flexdashboard::flex_dashboard:
theme: cosmo
# theme:
# version: 4
# bootswatch: spacelab
orientation: rows
social: menu
navbar:
- { title: "源码", href: "https://github.com/rstudio/flexdashboard/tree/main/examples/04_shiny-cran-downloads" }
runtime: shiny
---
这里开始用markdown语法写文章来讲解R markdown源码，所以源码中代码块标志的三个反引号，中间一个加了个反斜杠，以免显示混乱，读者拷贝合并代码块时，要注意把额外的反斜杠去掉。
`\``{r setup, include=FALSE}
library(flexdashboard)
# 这个工具类R函数源码从 CRAN上读取R包下载日志的增量数据流，后面再介绍。
source("helpers.R")
# pkgStream是一个反应表达式，它代表了增量的R包下载日志数据流。
# 每秒更新一次并返回上次更新后的增量下载数据data frame。
# 通过invalidateLater()函数让反应表达式1秒后自动失效来自动从服务器更新数据。
# 参阅 https://mastering-shiny.org/reactivity-objects.html#timed-invalidation-adv
pkgStream <- packageStream()
# pkgData 也是一个反应表达式，它累积了之前所有pkgStream返回的数据，
# 然后抛弃了所有超过maxAgeSecs秒的数据，这里包含的是5分钟内的数据。
# 有需要的话，也可以把该参数变成反应式变量，动态调整。
maxAgeSecs <- 60 * 5
pkgData <- packageData(pkgStream, maxAgeSecs)
`\``
Sidebar {.sidebar}是兼容shinydashboard的写法，表示下面的Shiny反应式输入组件安排在左边，下面一排等号是分页符，相当于一级标题"#"，要把Sidebar作为一个单独的框架页面。具体markdown语法请参阅《Markdown syntax》一节。这里在R代码块中定义了两个Shiny反应式输入组件sliderInput()与numericInput()。
仪表板中的每个组件都可以包括标题和注释部分。三级标题 "###" 后面的文本为标题；">" 开头的文本是注释。
Sidebar {.sidebar}
=======================================================================
### cran.rstudio.com
此例的数据流是延迟1周的cran.rstudio.com下载日志，产生下载日志数据流的服务器代码在[jcheng5/cransim](https://github.com/jcheng5/cransim)。
`\``{r}
# 下载高流量颜色阀值
sliderInput("rateThreshold", "当流量超出时以不同颜色提醒:",
min = 0, max = 100, value = 50, step = 1
)
# 最近下载窗口显示数量
numericInput("maxrows", "最近下载窗口显示数量:", 50)
`\``
> 数据每秒钟更新一次， 从服务器读取增量的数据，即该时间间隔之间下载R包的流量数据。
仪表板这一页中，Row下面一行减号相当于二级标题"##Row"，用于在仪表板的布局中分行，它是个网格结构，具体布局可以参阅《Layout》一节，这里源码中用等号行分页减号行分行是为了在源码中便于阅读。在flexdashboard中，一级标题会作为整个仪表板的标题显示，在本例中就是顶部“仪表板”Tab；三级标题会作为仪表板组件的标题显示，二级标题是布局标题，它们不会显示。这一行的网格中会安放流量表、总下载数、下载用户数3个组件。
仪表板
=======================================================================
Row
-----------------------------------------------------------------------
### 下载数/秒 (过去 5 分钟)
`\``{r}
# downloadRate 是一个计算仪表板运行期间下载流量的反应表达式
# 记住pkgData()存放的是5分钟内R包下载日志的数据。
startTime <- as.numeric(Sys.time())
downloadRate <- reactive({
elapsed <- as.numeric(Sys.time()) - startTime
nrow(pkgData()) / min(maxAgeSecs, elapsed)
})
# 输出下载流量指标。因为是在Shiny反应式编程环境中，要用render()函数封装渲染。
# gauge()是flexdashboard的仪表htmlwidget，根据上面的反应式变量rateThreshold设置仪表的颜色。
renderGauge({
rate <- formatC(downloadRate(), digits = 1, format = "f")
gauge(rate, min = 0, max = 100, symbol="次", gaugeSectors(
#success = c(0, 33), warning = c(34, 66), danger = c(77, 100)
success = c(0, input$rateThreshold), warning = c(input$rateThreshold, 100)
))
})
`\``
### 总下载数 {.value-box}
`\``{r}
# dlCount 是一个反应表达式，
# 记录了从 pkgStream收到的所有数据行数，跨越了5分钟的时间窗口。
dlCount <- downloadCount(pkgStream)
# 输出总下载数
renderValueBox({
valueBox(dlCount(), icon = "fa-download")
})
`\``
### 下载的用户数 {.value-box}
`\``{r}
# usrCount 是一个反应表达式，
# 记录了仪表板运行期间下载过R包的单个用户计数。
usrCount <- userCount(pkgStream)
# 输出下载的用户数
renderValueBox({
valueBox(value = usrCount(), icon = "fa-users")
})
`\``
这一行安放2个flexdashboard组件，过去5分钟内下载最多的R包，它们所占百分比的列表和泡泡图。这是render()泡泡图，记住，反应式编程环境要用render()函数渲染输出。
Row
-----------------------------------------------------------------------
### 下载最多的R包 (过去 5 分钟) {data-width=700}
`\``{r}
# 泡泡图HTML widget bubbles, https://github.com/jcheng5/bubbles。
renderBubbles({
if (nrow(pkgData()) == 0)
return()
order <- unique(pkgData()$package)
df <- pkgData() %>%
group_by(package) %>%
tally() %>%
arrange(desc(n), tolower(package)) %>%
# 只显示前60，否则可视化效果不好。
head(60)
bubbles(df$n, df$package, key = df$package, color = rainbow(60, alpha=NULL)[sample(60)])
})
`\``
### 下载百分比 (过去 5 分钟) {data-width=340}
`\``{r}
renderTable({
df <- pkgData() %>%
group_by(package) %>%
tally() %>%
arrange(desc(n), tolower(package)) %>%
mutate(percentage = n / nrow(pkgData()) * 100) %>%
select("Package" = package, "Percent" = percentage) %>%
as.data.frame() %>%
head(30)
# 列名改为中文，只显示前30。
names(df)<- c("R包","百分比")
df
}, digits = 1)
`\``
等号行分页，这一页只有一个Shiny输出组件最近下载列表，不需要布局代码，所以没有二级标题。可以在仪表板顶端按“最近下载”Tab切换。
最近下载
=======================================================================
### 最近下载
`\``{r}
renderTable({
downloads <- tail(pkgData(), n = input$maxrows)
downloads <- downloads[,c("date", "time", "size", "r_version",
"r_arch", "r_os", "package")]
downloads[order(nrow(downloads):1),]
# 列名改为中文。
names(downloads)<-c("日期", "时间", "大小", "R版本",
"体系", "OS", "包名")
})
`\``
2、helpers.R
用到的两个包shinySignals、bubbles需要从Github安装。
library(shiny)
# devtools::install_github("hadley/shinySignals")
library(shinySignals)
library(dplyr)
# devtools::install_github("jcheng5/bubbles")
library(bubbles)
# 这是一个空的 data frame 原型，用于存放服务器返回的下载日志数据。
prototype <- data.frame(date = character(), time = character(),
size = numeric(), r_version = character(), r_arch = character(),
r_os = character(), package = character(), version = character(),
country = character(), ip_id = character(), received = numeric())
packageStream()连接服务器读取cran.rstudio.com的下载日志stream，返回一个data frame stream。返回的是一个反应表达式，通过Shiny invalidateLater()机制每1000毫秒更新一次。产生下载日志数据流的服务器代码在jcheng5/cransim，用go语言编写，扫描下载日志归档文件并返回相应的增量数据。
packageStream <- function(session = getDefaultReactiveDomain()) {
# Connect to data source
sock <- socketConnection("cransim.rstudio.com", 6789, blocking = FALSE, open = "r")
# Clean up when session is over
session$onSessionEnded(function() {
close(sock)
})
# 通过一个定时失效的反应表达式得到日志中新的行。
newLines <- reactive({
invalidateLater(1000, session)
readLines(sock)
})
# 将日志行数据转换为data frame，并以反应表达式返回。
reactive({
if (length(newLines()) == 0)
return()
read.csv(textConnection(newLines()), header=FALSE, stringsAsFactors=FALSE,
col.names = names(prototype)
) %>% mutate(received = as.numeric(Sys.time()))
})
}
把反应表达式pkgStream返回的行，累积起来，然后按时间窗口（5分钟）过滤。使用了上面packageStream()为数据添加的received时间标签。可以用?shinySignals::reducePast看该函数的文档，这应该是最难理解的一个函数了，会用即可。shinySignals包提供了Shiny反应机制反应讯号处理的扩展工具，文档见其Github项目。
packageData <- function(pkgStream, timeWindow) {
shinySignals::reducePast(pkgStream, function(memo, value) {
rbind(memo, value) %>%
filter(received > as.numeric(Sys.time()) - timeWindow)
}, prototype)
}
累计反应表达式pkgStream收到的行数。看懂了上面shinySignals::reducePast()的用法，这里也就看懂了。
downloadCount <- function(pkgStream) {
shinySignals::reducePast(pkgStream, function(memo, df) {
if (is.null(df))
return(memo)
memo + nrow(df)
}, 0)
}
累计的单独用户计数。这个算法有点复杂，本篇的重点不在根据IP和日期确定跨日的两条下载记录是否为同一个用户，可以先不管它，就当是个测试数据好了，有时间再仔细分析它的算法，所以把注解原文留下。
# Use a bloom filter to probabilistically track the number of unique
# users we have seen; using bloom filter means we will not have a
# perfectly accurate count, but the memory usage will be bounded.
userCount <- function(pkgStream) {
# These parameters estimate that with 5000 unique users added to
# the filter, we'll have a 1% chance of false positive on the next
# user to be queried.
bloomFilter <- BloomFilter$new(5000, 0.01)
total <- 0
reactive({
df <- pkgStream()
if (!is.null(df) && nrow(df) > 0) {
# ip_id is only unique on a per-day basis. To make them unique
# across days, include the date. And call unique() to make sure
# we don't double-count dupes in the current data frame.
ids <- paste(df$date, df$ip_id) %>% unique()
# Get indices of IDs we haven't seen before
newIds <- !sapply(ids, bloomFilter$has)
# Add the count of new IDs
total <<- total + length(newIds)
# Add the new IDs so we know for next time
sapply(ids[newIds], bloomFilter$set)
}
total
})
}
# Quick and dirty bloom filter. The hashing "functions" are based on choosing
# random sets of bytes out of a single MD5 hash. Seems to work well for normal
# values, but has not been extensively tested for weird situations like very
# small n or very large p.
library(digest)
library(bit)
BloomFilter <- setRefClass("BloomFilter",
fields = list(
.m = "integer",
.bits = "ANY",
.k = "integer",
.bytesNeeded = "integer",
.bytesToTake = "matrix"
),
methods = list(
# @param n - Set size
# @param p - Desired false positive probability (e.g. 0.01 for 1%)
initialize = function(n = 10000, p = 0.001) {
m = (as.numeric(n) * log(1 / p)) / (log(2)^2)
.m <<- as.integer(m)
.bits <<- bit(.m)
.k <<- max(1L, as.integer(round((as.numeric(.m)/n) * log(2))))
# This is how many *bytes* of data we need for *each* of the k indices we need to
# generate
.bytesNeeded <<- as.integer(ceiling(log2(.m) / 8))
.bytesToTake <<- sapply(rep_len(.bytesNeeded, .k), function(byteCount) {
# 16 is number of bytes an md5 hash has
sample.int(16, byteCount, replace = FALSE)
})
},
.hash = function(x) {
hash <- digest(x, "md5", serialize = FALSE, raw = TRUE)
sapply(1:.k, function(i) {
val <- rawToInt(hash[.bytesToTake[,i]])
# Scale down to fit into the desired range
as.integer(val * (as.numeric(.m) / 2^(.bytesNeeded*8)))
})
},
has = function(x) {
all(.bits[.hash(x)])
},
set = function(x) {
.bits[.hash(x)] <<- TRUE
}
)
)
rawToInt <- function(bytes) {
Reduce(function(left, right) {
bitwShiftL(left, 8) + right
}, as.integer(bytes), 0L)
}
