使用livy提交supermap iobjects for spark任务

本文主要是介绍使用livy提交supermap iobjects for spark任务，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

使用Livy提交SuperMap iObjects Spark任务

Livy是一个提供标准rest接口和spark集群交互的服务，从官网的介绍来看它具备以下特点：多用户多客户端(web/iMobile)的提交方式、支持提交多种编程语言(scala/java/python)的执行代码片段和执行Jar包、现有spark项目可立即使用Livy且无需任何项目的改动、同步或异步的返回执行结果、提供SparkContext的管理，总之通过restfull接口或RPC客户端库，它简化了与spark应用的交互。本文简单演示如何通过Livy提交超图spark组件的执行任务。

在这里插入图片描述

环境

spark-2.1.1-bin-hadoop2.7
hadoop-2.7.3
supermap iobjects for spark 901、supermap iobjects for java 901
Apache Livy 0.5.0-incubating (zip)

Livy配置

前提已部署好spark，hadoop应用环境

部署及启动

# 解压zip
unzip Livy-0.5.0-incubating.zip 
# 添加Livy的conf及sh文件的配置
cd /opt/livy-0.5.0-incubating-bin/conf
# conf文件添加下面三个参数--其他参数参考官网
vi livy.conf
livy.server.host = 192.168.31.130
livy.server.port = 8998
livy.spark.master = spark://192.168.31.130:7077
# sh文件添加下面三个参数--其他参数参考官网
vi livy-env.sh
JAVA_HOME=/opt/jdk1.8.0_151
HADOOP_CONF_DIR=/opt/hadoop-2.7.3/etc/hadoop
SPARK_HOME=/opt/spark-2.1.1-bin-hadoop2.7
# 启动livy服务前,创建log文件目录,启动spark及hadoop
mkdir logs #与bin目录同级
./livy-server start

在这里插入图片描述

Livy与Spark交互提供了rest接口，也提供了编程语言的接口，支持spark的local、standlone、yarn集群模式，接下来我们采用简单的rest接口进行交互，使用python的request库与livy进行交互：

创建session

#由于需要依赖supermap spark组件的库，构建session时需要将依赖的jar添加到spark的jvm进程里，可以采用livy构建session提供的“jars”键来指定需要提交的jar列表。但我们采用了一种简单的方式-把依赖库com.supermap.bdt.core-9.1.0.jar直接放入spark的jars目录。
#构建session，是一个长连接，构建前一定要把依赖库jar放入spark的jars目录。
import json, pprint, requests, textwrap
host = 'http://192.168.31.130:8998'
data = {'kind': 'spark'}
headers = {'Content-Type': 'application/json'}
r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers)
session_url = host + r.headers['location']
session_url
# 构建成功，从下图可以看见id为0的session，运行状态是运行中，在spark里也能看见启动了一个应用。

在这里插入图片描述

提交读取udb的任务

# 以执行代码判断方式提交一个任务
# 注意：supermap iobjects spark产品提供了scala和python的api，故此任务包裹的执行代码是scala执行代码。
statements_url = session_url + '/statements'
data = {'code': textwrap.dedent("""val filem=\"/home/yb/opt/supermap-spark-9.0.0/data/NYCData.udb\";val roadRDD = com.supermap.bdt.io.sdx.SDXReader.readFromUDB(sc,filem,\"zones\");val str = roadRDD.count();println("记录总数"+str)""")
}
r = requests.post(statements_url, data=json.dumps(data), headers=headers)
r.json()
# 任务执行结果在log里查看如截图

在这里插入图片描述

提交读取hdfs系统上csv任务

statements_url = session_url + '/statements'
data = {'code': textwrap.dedent("""val filem=\"hdfs://192.168.31.130:9000/test/Convert.csv\";val roadRDD = com.supermap.bdt.io.simpleCSV.SimpleCSVReader.read(sc,filem);val str = roadRDD.count();println("记录总数"+str)""")
}
r = requests.post(statements_url, data=json.dumps(data), headers=headers)
r.json()
# 任务执行结果采用rest接口(session的id加statements的id)查看