分析网络托管的 JSON 数据

内容

实时笔记本

您可以在 live session 中运行此笔记本，或查看 Github 上的内容。

分析网络托管的 JSON 数据¶

本笔记本使用 Dask Bag 和 Dask Dataframe 的组合来读取和处理托管在网络上的 JSON 编码数据。

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

数据集介绍¶

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

[ ]:

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

我们看到它包含了每次有人在网站上启动实时笔记本时的一行记录。它包含了笔记本启动的时间，以及提供该笔记本的仓库。

在这个笔记本中，我们将查看许多这样的文件，将它们从JSON解析为Python字典，然后从那里转换为Pandas数据框。然后，我们将在这些数据上进行一些简单的分析。

启动 Dask 客户端以使用仪表板¶

启动 Dask 客户端是可选的。它将启动仪表板，这对于深入了解计算非常有用。

[ ]:

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

获取网页上的文件列表¶

mybinder.org 团队维护了一个索引文件，该文件指向所有其他可用的数据 JSON 文件。让我们将其转换为 URL 列表，我们将在下一节中读取这些 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]

创建所有事件的包¶

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

[ ]:

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

最受欢迎的笔记本¶

让我们做一个简单的频率统计，找出那些最常运行的绑定器。

[ ]:

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

转换为 Dask 数据框¶

最后，我们可以将我们的 Python 字典包转换为 Dask 数据框，并继续进行更多类似 Pandas 的计算。

我们将使用Pandas语法进行与上述相同的计算。

[ ]:

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

[ ]:

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

持久化内存¶

这个数据集非常适合存储在内存中。让我们避免每次执行操作时都下载数据，而是将数据保存在本地内存中。

[ ]:

df = df.persist()

老实说，在这个阶段切换到 Pandas 更有意义，但这是一个 Dask 示例，所以我们继续使用 Dask 数据帧。

调查除 Github 以外的提供商¶

大多数绑定器被指定为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())

奇异值分解

异步/等待与非阻塞执行

Dask Examples 文档