我是這樣用Django和D3開發911數據看板的
最近我和幾個警局部門合作創建了一個顯示 911（也被稱作 Call for Service）數據的智能看板，該看板允許用戶對這些數據進行進一步挖掘。當我開始這個項目的時候，已經有一個用 dc.js 寫的原型，dc.js 是用于構建動態看板的 JavaScript 框架，所有的數據都在前端（data on the frontend），原型的數據記錄來源于佛羅里達州的 Tampa 市。我需要基于此原型，增強其處理數據的能力 —— 上百萬條數據記錄。
Tampa/dc.js 智能看板原型
我們面臨的問題是要讓這個智能看板適用于其他城市。Tampa市警局的看板是純前端實現，沒有后臺服務器。所有的數據在一個 CSV 文件中，運行時所有數據會讀進瀏覽器的內存當中。這種方式顯然不適用于大型數據，也很難進行設置和升級。
我決定用 Django 和 D3 來創建新的看板。同時，我選取了一些用起來順手工具。它們包括：
-
Django REST Framework
-
django-url-filter
-
NVD3 & D3
-
Ractive.js
-
Leaflet
架構設計
和其他項目一樣，最終的架構和開始時的很不一樣。通常對我來說更簡單的做法是先探索性地寫程序，同時搞清楚我需要什么，然后再從頭開始寫，或者將之前的代碼整合到最終的架構中。一開始時，有幾點需求是明確的：
-
所有數據都應該在后臺處理，最好是數據庫中處理。
-
智能看板的所有視圖都可以收藏。
-
我需要一個能實時響應的前臺界面（reactive frontend）。
我要多說兩句，解釋下最后一點。什么是響應式編程（reactive programming）？簡單來說，就是當后臺數據更新的時，UI 也同時更新。我希望由數據驅動應用。下文我還會就此進行說明。
最終成果
Django
針對 Django，我對看板的每個頁面設置了一個 JSON 端點（endpoint）。其中一個頁面顯示撥打 911的總量，另外一個顯示電話接通時間，第三個以地圖的方式顯示所有電話的來源地，我們給每個頁面都設置了一個端點：
url(r'^api/call_volume/$', views.APICallVolumeView.as_view()),
url(r'^api/response_time/$', views.APICallResponseTimeView.as_view()),
url(r'^api/call_map/$', views.APICallMapView.as_view()),
為了生成這些端點的內容，我給每一組圖表創建了一個“摘要模型”（summary model）。這些摘要模型訪問數據庫，并為 API 的輸出建立數據結構。這些模型均繼承自一個基類，這樣就能輕松創建新的模型。
class CallOverview:
def __init__(self, filters):
self.filter = CallFilterSet(data=filters,
queryset=Call.objects.all(),
strict_mode=StrictMode.fail)
self.qs = self.filter.filter()
self.bounds = self.qs.aggregate(min_time=Min('time_received'),
max_time=Max('time_received'))
def by_dow(self):
results = self.qs \
.annotate(id=F('dow_received'), name=F('dow_received')) \
.values("id", "name") \
.annotate(**self.annotations)
return self.merge_data(results, range(0, 7))
class CallVolumeOverview(CallOverview):
annotations = dict(volume=Count("id"))
class CallResponseTimeOverview(CallOverview):
annotations = dict(mean=Avg(Secs("officer_response_time")))
我在每一類中使用 annotatations 決定看板的頁面上需要展示什么數據。
另外，我在摘要模型中自定義了對象關系映射（ORM）函數和集群（aggregations）。看下面這個例子：
def precision(self):
if self.span >= timedelta(days=365):
return 'month'
elif self.span >= timedelta(days=7):
return 'day'
else:
return 'hour'
def volume_by_date(self):
results = self.qs.annotate(date=DateTrunc('time_received',
precision=self.precision())) \
.values("date") \
.annotate(volume=Count("date")) \
.order_by("date")
return results
你會發現我在這里使用了 DateTrunc 。它使用了 PostgresSQL 中的一個同名函數，將根據你處理的數據量將時間數據截成月、天或者小時。
class DateTrunc(Func):
"""
Accepts a single timestamp field or expression and returns that timestamp
truncated to the specified *precision*. This is useful for investigating
time series.
The *precision* named parameter can take:
* microseconds
* milliseconds
* second
* minute
* hour
* day
* week
* month
* quarter
* year
* decade
* century
* millennium
Usage example::
checkin = Checkin.objects.
annotate(day=DateTrunc('logged_at', 'day'),
hour=DateTrunc('logged_at', 'hour')).
get(pk=1)
assert checkin.logged_at == datetime(2015, 11, 1, 10, 45, 0)
assert checkin.day == datetime(2015, 11, 1, 0, 0, 0)
assert checkin.hour == datetime(2015, 11, 1, 10, 0, 0)
"""
function = "DATE_TRUNC"
template = "%(function)s('%(precision)s', %(expressions)s)"
def __init__(self, expression, precision, **extra):
super().__init__(expression, precision=precision, **extra)
對這么一小段代碼而言，注釋好像有點太多了。不過很高興能看到拓展 Django 的對象關系映射是這么簡單。
我還自定義了集群， Percentile ：
class CallResponseTimeOverview(CallOverview):
annotations = dict(mean=Avg(Secs("officer_response_time")))
default = dict(mean=0)
def officer_response_time(self):
results = self.qs.filter(
officer_response_time__gt=timedelta(0)).aggregate(
avg=Avg(Secs('officer_response_time')),
quartiles=Percentile(Secs('officer_response_time'),
[0.25, 0.5, 0.75],
output_field=ArrayField(DurationField)),
max=Max(Secs('officer_response_time')))
每一個摘要模型都用方法 to_dic 生成輸出：
def to_dict(self):
return {
'filter': self.filter.data,
'bounds': self.bounds,
'precision': self.precision(),
'count': self.count(),
'volume_by_date': self.volume_by_date(),
'volume_by_source': self.volume_by_source(),
'volume_by_district': self.by_field('district'),
'volume_by_beat': self.by_field('beat'),
'volume_by_nature': self.by_field('nature'),
'volume_by_nature_group': self.by_nature_group(),
'volume_by_dow': self.by_dow(),
'volume_by_shift': self.by_shift(),
'heatmap': self.day_hour_heatmap(),
'beat_ids': self.beat_ids(),
'district_ids': self.district_ids(),
}
這些摘要模型由一系列過濾器驅動。這里，這些過濾器被指定為調用 API 時的 GET 實參。Django 主要有兩個包協助完成過濾工作， django-filter 和 django-url-filter 。 django-url-filter 不常見也很難用。我之所以用它，是因為它更容易被修改，這正是我需要的。對于類似的項目，無論哪一個應該都很好用。
我第一個要修改的就是讓查詢方法被能用于過濾器。 django-url-filter 可以將GET實參，如 district=7&nature=10 ，作為形參傳遞給模型類 .filter 方法。修改后，任何指向查詢集（queryset）的 GET 實參將調用那個方法，于是 shift=1&district=7 （ .shift 是用于查詢的方法）將最終調用 Call.objects.filter(district_id=7).shift(1) 。
下面一段代碼能更簡單，也能實現同樣的功能：
def filter(self):
include = {self.prepare_spec(i): self.prepare_value(i) for i in
self.includes}
qs = self.queryset
for k, v in include.items():
try:
qs = getattr(qs, k)(v)
except AttributeError:
qs = qs.filter(**{k: v})
另一個我需要的修改是從數據結構創建過濾器，而不是按類創建。為了防止代碼重復，我以 JSON 數據結構的方式將過濾器提供給前臺，這樣我就能用它控制智能看板。請看如下所示數據結構。就像大多數其他按需變化的事物一樣，代碼顯得很雜亂，但應該是不言自明的。
[ {"name": "time_received", "type": "daterange"},
{"name": "shift", "type": "select", "method": True,
"lookups": ["exact"],
"options": [[0, "Shift 1"], [1, "Shift 2"]]},
{"name": "dow_received", "label": "Day of Week", "type": "select",
"options": [
[0, "Monday"], [1, "Tuesday"], [2, "Wednesday"], [3, "Thursday"],
[4, "Friday"], [5, "Saturday"], [6, "Sunday"]
]},
{"name": "district", "rel": "District"},
{"name": "beat", "rel": "Beat"},
{"name": "squad", "rel": "Squad", "method": True, "rel": "Squad",
"lookups": ["exact"]},
{"name": "priority", "rel": "Priority"},
{"name": "nature", "rel": "Nature"},
{"name": "nature__nature_group", "label": "Nature Group", "rel": "NatureGroup"},
{"name": "initiated_by", "type": "select", "method": True,
"lookups": ["exact"],
"options": [[0, "Officer"], [1, "Citizen"]]},
{"name": "call_source", "rel": "CallSource"},
{"name": "cancelled", "type": "boolean"}, ]
注意： 將這些轉變成Python對象的代碼很拙劣 。
最后，我利用 Django REST 框架生成真正的結點。回想一下，這些都能用 Django 獨立實現。Django REST 框架是個很強大的平臺，因此，我在其他地方也使用了它的序列化器（Serializer）。
class APICallResponseTimeView(APIView):
"""Powers response time dashboard."""
def get(self, request, format=None):
overview = CallResponseTimeOverview(request.GET)
return Response(overview.to_dict())
前臺
由數據驅動
前文中，我提到過要讓數據驅動應用。實操中，那意味著前臺應用工作流如下：
- 用戶通過點擊圖表上的過濾器選項欄選擇不同的過濾器。
- URL會被更新以反映現有的過濾器。
- 應用監視 URL 的變化。只要 URL 一變，它就向后臺請求新數據。
- 當請求返回后，我們更新數據。
- 當數據更新，頁面也就更新了。
我們將這種架構稱為“響應式的”，但到底是什么意思？前臺訂閱事件（包括數據變更），然后響應事件后進行自我更新。最后兩點就是響應式的由來。
前三項很有意思，而且，顛覆了你之前的期待。過濾器改變時，我便更新內部狀態，而當狀態更新后，URL 的哈希值也更新了。我訂閱了「哈希值變化」事件，然后發出 Ajax 請求。根據下述應用流步驟，回想一下就看到了事件和響應。他們看起來同步，實際是異步的。
- 當用戶點擊圖表，過濾器改變。
- 當用戶使用過濾器欄，過濾器改變。
- 當過濾器改變：
- URL 哈希值更新
- 過濾器選項欄更新
- 當 URL 哈希值改變，發出調用新數據的 Ajax 請求。
- 當請求返回， 調用的數據被更新。
- 當被調用的數據被更新，圖表就更新了。
這種響應式范式，和你在著名的 React.js 中看到的是一樣的。我們用的是 Ractive.js ，這個庫沒那么有名，但是簡單易懂。對于本項目中的智能看板，Ractive 的復雜程度剛好合適。在兩個庫中，你都需要創建組件。這些組件當中有數據，當數據變化觸發的事件，并可以通過 UI 互動觸發事件。
下面是一個簡單的 Ractive 組件和相關的模板：
var ChartHeader = Ractive.extend({
template: require("../templates/chart_header.html"),
data: {
hidden: true
},
oninit: function() {
this.on("toggleExplanation", function() {
this.set("hidden", !this.get("hidden"));
});
}
});
<div>
<h3 class="chart-title">
{{ title }}
<i class="fa fa-info-circle clickable"
on-click="toggleExplanation"></i>
</h3>
{{ #unless hidden }}
<div class="explanation well">
{{ >content }}
</div>
{{ /unless }}
</div>
注意，這個模板并不止渲染一次。每當 Ractive 組件中的數據發生變化，就會重新計算該模板。
圖表
D3 是目前為止最好的、功能最齊全的可視化庫，但是說它是畫圖表的庫并不準確。它是一個可用于生成圖表和其他可視化形式的底層工具。如果你只需要一些圖表，選擇更高層級（high-level）工具會使你的工作變得輕松得多。你應該選擇一些基于 D3 的工具——有 D3 作為基礎，可以更容易地從生成可視化圖形，如我們用到的熱圖（heatmap）。一些推薦工具如下：
我們用的是 NVD3。 這是個不錯的選擇：因為 NVD3 的默認樣式很棒，圖表種類也不少。
我們來看一個 NVD3 圖表，以及如何更新。首先創建一個高級對象開始， HorizontalBarChart 。
var volumeByDOWChart = new HorizontalBarChart({
dashboard: dashboard,
filter: "dow_received",
el: "#volume-by-dow",
ratio: 1.5,
fmt: d3.format(",d"),
x: function (d) {
return d.name;
},
y: function (d) {
return d.volume;
}
});
monitorChart(dashboard, "data.volume_by_dow", volumeByDOWChart.update);
HorizontalBarChart 可接受多種選項，其中有的設計表現形式，有的更加底層。 dashboard 是真正的 Ractive 智能看板插件，我們需要從中獲取數據，然后觸發一個在看板加載完后繪制圖表的事件。 filter 是過濾器對象的關鍵，前端監控的鼠標點擊時將更新圖表。其他都是表現形式選項，指定圖表渲染位置及如何格式化數據
和所有高級組件一樣， HorizontalBarChart 有一個 .create 方法，在實例化（instantiation）時被調用, 還有一個被 monitorChart 調用的 .update 方法。 monitorChart 這個函數是用于監視看板中關鍵路徑（“keypath”，是指一個樹狀目錄型的數據）上的數據，當該數據的子集改變時調用一個函數。
function monitorChart (ractive, keypath, fn) {
ractive.observe(keypath, function (newData) {
if (!ractive.get("loading")) {
fn(newData);
}
});
}
制作一個新圖表，只需要定義一個新圖表對象，然后給它配置一個 monitorChart 即可。（回過頭來看， monitorChart 這個名字起得很糟糕：它應該叫 monitorRactive 。）
熱圖
高級庫不是什么事都能干。在本項目中，我們想要一個基于天/小時的熱圖，用于查看撥打 911 的通訊量。我們需要用 D3 來實現這個。我們從 bl.ocks.org 的一個例子 開始。
如果要一行一行代碼地介紹，可能得領寫一篇 D3 教程，但是我們有兩個貢獻反映了怎么將同樣的形式復制到其他可視化圖形中：
this.create = function () {
var bounds = this.getBounds(),
container = d3.select(this.el),
width = bounds.width,
height = bounds.height,
gridSize = Math.floor(width / this.ratio / 10);
container
.append("svg")
.attr("width", width)
.attr("height", height)
.attr("viewBox", "0 0 " + width + " " + height)
.append("g")
.attr("transform", "translate(" + margin.left + "," + margin.top + ")");
this.drawLabels();
d3.select(window).on("resize", function () {
self.resize();
});
this.drawn = true;
};
this.update = function (data) {
self.ensureDrawn().then(function () {
self._update(data);
});
};
我們用一個方法去建立可視化，然后當 Ractive 組件中正確的子集數據變化時調用一個 .update 方法。我們必須確保這個組件被完全繪制，因為Ractive組件中的數據會隨著頁面加載立即變化。
經驗總結
正如其他軟件工程，回看之前的過程我會以不同的方式去做很多事情。在寫這篇文章的時候，我發現了很多可以重寫的代碼。我還想嘗試其他不同的庫。總之，盡管如此，我還是對該項目很滿意，特別是它的架構。
這個架構的優勢在于它的具有單向流動性以及可標記狀態。數據處理都交由 PostgreSQL 完成，而 Django 則負責協調看板和數據庫的通訊。
來自：http://www.codingpy.com/article/build-dashboards-with-django-and-d3/