分析网络托管的 JSON 数据

分析网络托管的 JSON 数据¶

本 notebook 使用 Dask Bag 和 Dask DataFrame 的组合读取并处理网络托管的 JSON 编码数据。

这些数据来自 mybinder.org，这是一个在网络上实时运行 Jupyter notebook 的网络服务（您现在可能正在那里运行此 notebook）。My Binder 发布每次有人启动像这样的实时 notebook 的记录，并将该记录存储在一个公开可访问的 JSON 文件中，每天一个文件。

数据集介绍¶

这些数据以 JSON 编码文本文件的形式存储在公共网络上。以下是一些示例行。

[ ]:

import dask.bag as db
db.read_text('https://archive.analytics.mybinder.org/events-2018-11-03.jsonl').take(3)

我们看到它包含该网站上每次有人启动实时 notebook 的一条记录。它包括 notebook 启动的时间以及提供该 notebook 的仓库。

在此 notebook 中，我们将查看许多此类文件，将它们从 JSON 解析为 Python 字典，然后从中转换为 Pandas 数据帧。然后，我们将对这些数据进行一些简单的分析。

启动 Dask Client 以查看仪表盘¶

启动 Dask Client 是可选的。它将启动仪表盘，这对于深入了解计算很有用。

[ ]:

from dask.distributed import Client, progress
client = Client(threads_per_worker=1,
                n_workers=4,
                memory_limit='2GB')
client

获取网络上的文件列表¶

mybinder.org 团队维护着一个索引文件，该文件指向所有其他可用的 JSON 数据文件。让我们将其转换为 URL 列表，以便在下一节中读取。

[ ]:

import dask.bag as db
import json

[ ]:

db.read_text('https://archive.analytics.mybinder.org/index.jsonl').map(json.loads).compute()

[ ]:

filenames = (db.read_text('https://archive.analytics.mybinder.org/index.jsonl')
               .map(json.loads)
               .pluck('name')
               .compute())

filenames = ['https://archive.analytics.mybinder.org/' + fn for fn in filenames]
filenames[:5]

创建包含所有事件的 Bag¶

我们现在围绕该 URL 列表创建一个 Dask Bag，然后对每一行调用 json.loads 函数，将这些 JSON 编码的文本行转换为可以更容易操作的 Python 字典。

[ ]:

events = db.read_text(filenames).map(json.loads)
events.take(2)

最受欢迎的 Binder¶

让我们做一个简单的频率计数，找出那些运行次数最多的 Binder。

[ ]:

events.pluck('spec').frequencies(sort=True).take(20)

转换为 Dask DataFrame¶

最后，我们可以将我们的 Python 字典 bag 转换为 Dask DataFrame，然后进行更多类似 Pandas 的计算。

现在，我们将使用 Pandas 语法执行与上面相同的计算。

[ ]:

df = events.to_dataframe()
df.head()

[ ]:

df.spec.value_counts().nlargest(20).to_frame().compute()

持久化到内存¶

这个数据集非常适合内存。为了避免每次操作都下载数据，我们可以将数据保留在本地内存中。

[ ]:

df = df.persist()

老实说，到目前为止，直接切换到 Pandas 更有意义，但这毕竟是 Dask 的示例，所以我们将继续使用 Dask DataFrame。

调查 Github 以外的提供商¶

大多数 Binder 都被指定为 GitHub 上的 git 仓库，但并非全部。让我们调查其他提供商。

[ ]:

import urllib

[ ]:

df.provider.value_counts().compute()

[ ]:

(df[df.provider == 'GitLab']
 .spec
 .map(urllib.parse.unquote, meta=('spec', object))
 .value_counts()
 .to_frame()
 .compute())

[ ]:

(df[df.provider == 'Git']
 .spec
 .apply(urllib.parse.unquote, meta=('spec', object))
 .value_counts()
 .to_frame()
 .compute())

奇异值分解

Async/Await 和非阻塞执行

Dask 示例文档