使用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)查看

简单的Livy入门介绍到此，至于如何提交已打包的jar文件，可以参考livy的batches批提交里的"file"和"className "参数，如需深度使用，如获取任务状态、删除session等，自行参考livy-rest-api