背景介绍

Spark可以像Hadoop一样用来跑大作业,也可以像数据库一样做实时分析。对于后者,Spark的发行版中提供了spark shell、spark sql,数据科学家可以在这里面做一些分析。不过命令行的可编辑、可视化方面毕竟不友好,因此一些厂商开发了web IDE,而开源界也提供了这样的工具:Zeppelin(齐普林?)。

Zeppelin是一个Web笔记形式的交互式数据查询分析工具,除了可以对接spark进行scala、python、sql的分析外,还可以对接hive、cassandra、kylin、phoenix、elasticsearch 等等。

安装

Zeppelin的安装十分简单。我从官网下载了0.5.6的release bin包,解压。默认其监听端口是8080,但在我的机器上这个端口号已经被spark的web ui占用了,所以我改为了8888:将zeppelin-site.xml.template复制一份为zeppelin-site.xml,并修改zeppelin.server.port的value即可。

服务启动、停止:

bin/zeppelin-daemon.sh start
bin/zeppelin-daemon.sh stop

我没有把Zeppelin装到yarn上。start后,打开http://host:8888接能看到页面了。下面介绍几个概念:

  • Interpreter:注释器,对应实际运行的spark集群、hive服务器等等,可以在页面上配置服务地址等参数。
  • Notebook:笔记本,我理解为一系列的分析,范加尔的小本子。note可以通过%xxx来定义其对应的interpreter,例如%hive对应hive,%spark/%sql/%pyspark对应spark。

默认装好后有一个Tutorial,可以看看官方给的几个示例。我们自己create new note,下面的几个操作都在这个note里做。

对接hive

我用spark的thrift server来替代hive。首先配置Interpreter中hive的default.url(jdbc:hive2://ipaddr:port)和default.user,save后记得restart下。 下面是个简单的例子,其实跟beeline类似,但Zeppelin做的比较好的一点是它直接集成了图形化,SQL(包括spark sql)出来的数据可以转为直方图、饼图、趋势图、分布图等等。

示例

对接spark

类似hive,需要修改Interpreter中spark的master项。我这里直接提交到spark集群上:spark://spark1:7077。

例一:求Pi值

类似spark的解释器。 pi

例二:IP地址瓜分图

本例稍微复杂一点:我在一个note中读取hdfs上的文件并注册为临时表(%spark),然后在另一个note中使用sql分析(%sql)。Zeppelin的各个note实际共用一个spark context,用户不需要也不能自定义sc。

1、编写scala部分代码。

数据如下:

3757755392,3757834239,CN,310000,310100,-1,100023,31.0456,121.4
3757834240,3757835775,AU,AU_04,Brisbane,-1,3000683,-27.4679,153.028
3757835776,3757836031,AU,AU_04,Brisbane,-1,3000683,-27.4833,153.042
...

前面2列定义了IP段,第三列为国家缩写。我想统计的是不同的国家,其所获得的IP段数量,看看是不是美国占了最多的资源(那还用说)。

代码如下,聪明如你一看就懂。


%spark

import sqlContext.implicits._

val ip2region = sc.textFile("hdfs://spark1:8020/upload/ipdata_code_with_maxmind.txt")

case class Ip(ipstart:Long, ipend:Long, country:String, province: String, city : String, county: String)

val ip_map = ip2region.map(s=>s.split(",")).filter(s=>(s.length==9)).map(
s=>Ip(s(0).toLong,
s(1).toLong, s(2), s(3), s(4), s(5))
).toDF()

ip_map.registerTempTable("ip2region")

可以点击下执行,不过由于这里都是transform,所以实际没做啥。

插一下

由于我们是把Zeppelin直接提交到spark集群上的,所以在spark的页面上,可以看到Zeppelin的Application,这个app可以被kill掉(灰色的kill)。

zeppelin-spark-application

点进去后可以看到一个个的作业。

zeppelin-spark-jobs

如果只是transform,不会形成作业;直到要action才会提交作业上去(图中第一行就是下面跑sql出来的)。我理解这就是spark的惰性求值的意思吧。

2、编写sql

现在已经有了一个叫ip2region的临时表,下面就可以直接对这个表进行计算了。

select distinct country, count(1) as sum from ip2region group by country order by country

不过,上面的计算比较粗糙,实际上每行记录所包含的地址段并不都是256个,有的记录是子网段的,如只有64个、128个,按行来统计的结果可能不准。 结果如下。

zeppelin_spark-sql

右上角的点是美国,左下高亮的点是中国。虽然我对CN分配的IP地址段会比较少做好了心理准备,但跟美国差距这么大,还是没想到的。当然总的来说我们的IP地址还是挺多的。

其他功能

  1. Zeppelin默认没有用户隔离,多个浏览器看到的是一样的(勉强认为具备了协同开发的能力吧)
  2. 可以将note保存成json串归档为文件,其他集群将该文件导入后,可以恢复note
  3. 可以根据note的URL来导入(how?)
  4. 可以定时执行note。由于定时执行时,note各部分是有执行顺序的,可以勉强认为具备了工作流的能力 XD