打破二八法则！每个数据科学家都得会一点SparkMagic-白红宇

打破二八法则！每个数据科学家都得会一点SparkMagic

阅读量：2089 次

发布时间：2019-04-29

本文共 4523 字，大约阅读时间需要 15 分钟。

全文共2504字，预计学习时长13分钟

图源：unsplash

著名的帕累托法则，即80/20定律，告诉我们：原因和结果、投入和产出、努力和报酬之间存在着无法解释的不平衡。即使是21世纪最具吸引力的工作，数据科学依然逃不脱这一定律。

商业数据科学家80％的时间都花在查找、清洗和准备数据上，这是数据科学家工作中效率最低也是最可怕的部分。互联网为如何打破数据科学的80/20定律提供了许多的意见，但却收效甚微。

其实，数据科学家生产率低下的主要原因在于数据准备工作的双重性：

· 快速访问、合并和聚合存储在企业数据湖中的大数据

· 探索和可视化数据中具有复杂依赖关系的Python数据包中的数据和统计信息

大数据大多是非结构化的，常常存储在具有企业管理和安全限制的生产环境中。快速访问数据需要昂贵的分布式系统，这些系统由IT集中管理，必须与其他数据科学家和分析师共享。

Spark是用于分布式数据湖中处理数据的行业黄金标准。但是，要以经济高效的方式使用Spark集群，甚至允许多租户，就很难满足单个需求和依赖关系。分布式数据基础架构的行业趋势是临时集群，这使得数据科学家更加难以部署和管理他们的Jupyter Notebook环境。

图源：CC BY-SA 3.0

很多数据科学家都在高规格笔记本电脑上进行本地工作，可以更加轻松地安装和持久保存Jupyter Notebook环境。那么这些数据科学家们如何将其本地开发环境与生产数据湖中的数据联系起来？通常，他们使用Spark实现了csv文件，并从云存储控制台下载了它们。

从云存储控制台手动下载csv文件既不高效，也没有特别强大的功能。如果能以终端用户友好且透明的方式无缝地将本地的Jupyter Notebook与远程集群连接起来，岂不是更好吗？

学好SparkMagic，打破数据科学二八法则的时间到了！

适用于Jupyter NoteBook的SparkMagic

Sparkmagic是一个通过Livy REST API与Jupyter Notebook中的远程Spark群集进行交互工作的项目。它提供了一组Jupyter Notebook单元魔术和内核，可将Jupyter变成用于远程集群的集成Spark环境。

合理使用和公共领域图标和svg | 图源：MIT

SparkMagic能够：

· 以多种语言运行Spark代码

· 提供可视化的SQL查询

· 轻松访问Spark应用程序日志和信息

· 针对任何远程Spark集群自动创建带有SparkContext和HiveContext的SparkSession

· 将Spark查询的输出捕获为本地Pandas数据框架，以轻松与其他Python库进行交互（例如matplotlib）

· 发送本地文件或Pandas数据帧到远程集群（例如，将经过预训练的本地ML模型直接发送到Spark集群）

可以使用以下Dockerfile来构建具有SparkMagic支持的Jupyter Notebook：

FROM jupyter/all-spark-notebook:7a0c7325e470USER$NB_USERRUN pip install --upgrade pipRUN pip install --upgrade --ignore-installed setuptoolsRUN pip install pandas --upgradeRUN pip install sparkmagicRUN mkdir /home/$NB_USER/.sparkmagicRUN wget https://raw.githubusercontent.com/jupyter-incubator/sparkmagic/master/sparkmagic/example_config.jsonRUN mv example_config.json /home/$NB_USER/.sparkmagic/config.jsonRUN sed -i  s/localhost:8998/host.docker.internal:9999/g /home/$NB_USER/.sparkmagic/config.jsonRUN jupyter nbextension enable --py --sys-prefix widgetsnbextensionRUN jupyter-kernelspec install --user --name SparkMagic $(pip show sparkmagic |grep Location | cut -d" " -f2)/sparkmagic/kernels/sparkkernelRUN jupyter-kernelspec install --user --name PySparkMagic $(pip show sparkmagic| grep Location | cut -d" " -f2)/sparkmagic/kernels/pysparkkernelRUN jupyter serverextension enable --py sparkmagicUSER rootRUN chown $NB_USER /home/$NB_USER/.sparkmagic/config.jsonCMD ["start-notebook.sh","--NotebookApp.iopub_data_rate_limit=1000000000"]USER $NB_USER

生成图像并用以下代码标记：

docker build -t sparkmagic

并在Spark Magic支持下启动本地Jupyter容器，以安装当前工作目录：

docker run -ti --name"${PWD##*/}-pyspark" -p 8888:8888 --rm -m 4GB --mounttype=bind,source="${PWD}",target=/home/jovyan/work sparkmagic

为了能够连接到远程Spark集群上的Livy REST API，必须在本地计算机上使用ssh端口转发。获取你的远程集群的IP地址并运行：

ssh -L 0.0.0.0:9999:localhost:8998REMOTE_CLUSTER_IP

首先，使用启用了SparkMagic的PySpark内核创建一个新的Notebook，如下所示：

在启用了SparkMagic的Notebook中，你可以使用一系列单元魔术来在本地笔记本电脑以及作为集成环境的远程Spark集群中使用。%% help魔术输出所有可用的魔术命令：

可以使用%%configuremagic配置远程Spark应用程序：

如图所示，SparkMagic自动启动了一个远程PySpark会话，并提供了一些有用的链接以连接到Spark UI和日志。

Notebook集成了2种环境：

1.%%local，可在笔记本电脑和jupyter docker映像提供的anaconda环境中本地执行单元

2.%%spark，通过远程Spark集群上的PySpark REPL，再通过Livy REST API远程执行单元

首先将以下code cell远程导入SparkSql数据类型；其次，它使用远程SparkSession将Enigma-JHU Covid-19数据集加载到我们的远程Spark集群中。可以在Notebook中看到remote .show（）命令的输出：

但这就是魔术开始的地方。可以将数据框注册为Hive表，并使用%%sql魔术对远程群集上的数据执行Hive查询，并在本地Notebook中自动显示结果。这不是什么高难度的事，但对于数据分析人员和数据科学项目早期的快速数据探索而言，这非常方便。

SparkMagic真正有用之处在于实现本地Notebook和远程群集之间无缝传递数据。数据科学家的日常挑战是在与临时集群合作以与其公司的数据湖进行交互的同时，创建并保持其Python环境。

在下例中，我们可以看到如何将seaborn导入为本地库，并使用它来绘制covid_data pandas数据框。

这些数据从何而来？它是由远程Spark集群创建并发送的。神奇的%%spark-o允许我们定义一个远程变量，以在单元执行时转移到本地笔记本上下文。我们的变量covid_data是一个远程集群上的SparkSQL Data Frame，和一个本地JupyterNotebook中的PandasDataFrame。

使用Pandas在Jupyter Notebook中聚合远程集群中的大数据以在本地工作的能力对于数据探索非常有帮助。例如，使用Spark将直方图的数据预汇总为bins，以使用预汇总的计数和简单的条形图在Jupyter中绘制直方图。

另一个有用的功能是能够使用魔术%%spark-o covid_data -m sample -r 0.5来采样远程Spark DataFrame。集成环境还允许你使用神奇的%%send_to_spark将本地数据发送到远程Spark集群。

图源：pexels

PandasDataFrames和字符串支持的两种数据类型。要将其他更多或更复杂的东西（例如，经过训练的scikit模型用于评分）发送到远程Spark集群，可以使用序列化创建用于传输的字符串表示形式：

import pickleimport gzipimport base64serialised_model = base64.b64encode(    gzip.compress(       pickle.dumps(trained_scikit_model)    )).decode()

但正如你所见，这种短暂的PySpark集群模式有一大诟病：使用Python软件包引导EMR集群，且这个问题不会随着部署生产工作负载而消失。我之前的这篇文章可以告诉你如何将PySpark工作负载部署到生产环境中（https://towardsdatascience.com/pex-the-secret-sauce-for-the-perfect-pyspark-deployment-of-aws-emr-workloads-9aef0d8fa3a5）

快将自己的生产力从数据准备的低效率中拯救出来吧，用80%中节省出来的时间去创造更多价值。

推荐阅读专题

留言点赞发个朋友圈

我们一起分享AI学习与发展的干货

编译组：高淳子、钟惠