如何在Spark Scala/Java应用中调用Python脚本

这篇具有很好参考价值的文章主要介绍了如何在Spark Scala/Java应用中调用Python脚本。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

本文将介绍如何在 Spark scala 程序中调用 Python 脚本,Spark java程序调用的过程也大体相同

1.PythonRunner

对于运行与 JVM 上的程序(即Scala、Java程序),Spark 提供了 PythonRunner 类。只需要调用PythonRunner 的main方法,就可以在Scala或Java程序中调用Python脚本。在实现上,PythonRunner 基于py4j ,通过构造GatewayServer实例让python程序通过本地网络socket来与JVM通信。

    // Launch a Py4J gateway server for the process to connect to; this will let it see our
    // Java system properties and such
    val localhost = InetAddress.getLoopbackAddress()
    val gatewayServer = new py4j.GatewayServer.GatewayServerBuilder()
      .authToken(secret)
      .javaPort(0)
      .javaAddress(localhost)
      .callbackClient(py4j.GatewayServer.DEFAULT_PYTHON_PORT, localhost, secret)
      .build()
    val thread = new Thread(new Runnable() {
      override def run(): Unit = Utils.logUncaughtExceptions {
        gatewayServer.start()
      }
    })
    thread.setName("py4j-gateway-init")
    thread.setDaemon(true)
    thread.start()

    // Wait until the gateway server has started, so that we know which port is it bound to.
    // `gatewayServer.start()` will start a new thread and run the server code there, after
    // initializing the socket, so the thread started above will end as soon as the server is
    // ready to serve connections.
    thread.join()

在启动GatewayServer后,再通过ProcessBuilder构造子进程执行Python脚本,等待Python脚本执行完成后,根据exitCode判断是否执行成功,若执行失败则抛出异常,最后关闭gatewayServer。

    // Launch Python process
    val builder = new ProcessBuilder((Seq(pythonExec, formattedPythonFile) ++ otherArgs).asJava)
    try {
      val process = builder.start()

      new RedirectThread(process.getInputStream, System.out, "redirect output").start()

      val exitCode = process.waitFor()
      if (exitCode != 0) {
        throw new SparkUserAppException(exitCode)
      }
    } finally {
      gatewayServer.shutdown()
    }

2.调用方法

2、1 调用代码

PythonRunner的main方法中需要传入三个参数:

  • pythonFile:执行的python脚本
  • pyFiles:需要添加到PYTHONPATH的其他python脚本
  • otherArgs:传入python脚本的参数数组
    val pythonFile = args(0)
    val pyFiles = args(1)
    val otherArgs = args.slice(2, args.length)

具体样例代码如下,scala样例代码:

package com.huawei.bigdata.spark.examples

import org.apache.spark.deploy.PythonRunner
import org.apache.spark.sql.SparkSession

object RunPythonExample {
  def main(args: Array[String]) {
    val pyFilePath = args(0)
    val pyFiles = args(1)
    val spark = SparkSession
      .builder()
      .appName("RunPythonExample")
      .getOrCreate()

    runPython(pyFilePath, pyFiles)

    spark.stop()
  }

  def runPython(pyFilePath: String, pyFiles :String) : Unit = {
    val inputPath = "-i /input"
    val outputPath = "-o /output"
    PythonRunner.main(Array(pyFilePath, pyFiles, inputPath, outputPath))
  }
}

python样例代码:

#!/usr/bin/env python
# coding: utf-8
import sys
import argparse

argparser = argparse.ArgumentParser(description="ParserMainEntrance")
argparser.add_argument('--input', '-i', help="input path", default=list(), required=True)
argparser.add_argument('--output', '-o', help="output path", default=list(), required=True)
arglist = argparser.parse_args()

def getTargetPath(input_path, output_path):
    try:
        print("input path: {}".format(input_path))
        print("output path: {}".format(output_path))
        return True
    except Exception as ex:
        print("error with: {}".format(ex))
        return False

if __name__ == "__main__":
    ret = getTargetPath(arglist.input, arglist.output)
    if ret:
        sys.exit(0)
    else:
        sys.exit(1)

2、2 运行命令

执行python脚本需要设置pythonExec,即执行python脚本所使用的执行环境。默认情况下,使用的执行器为python(Spark 2.4 及以下)或 python3 (Spark 3.0 及以上)。

    //Spark 2.4.5
    val sparkConf = new SparkConf()
    val secret = Utils.createSecret(sparkConf)
    val pythonExec = sparkConf.get(PYSPARK_DRIVER_PYTHON)
      .orElse(sparkConf.get(PYSPARK_PYTHON))
      .orElse(sys.env.get("PYSPARK_DRIVER_PYTHON"))
      .orElse(sys.env.get("PYSPARK_PYTHON"))
      .getOrElse("python")

    //Spark 3.1.1
    val sparkConf = new SparkConf()
    val secret = Utils.createSecret(sparkConf)
    val pythonExec = sparkConf.get(PYSPARK_DRIVER_PYTHON)
      .orElse(sparkConf.get(PYSPARK_PYTHON))
      .orElse(sys.env.get("PYSPARK_DRIVER_PYTHON"))
      .orElse(sys.env.get("PYSPARK_PYTHON"))
      .getOrElse("python3")

如果要手动指定pythonExec,需要在执行前设置环境变量(无法通过spark-defaults传入)。在cluster模式下,可以通过 --conf “spark.executorEnv.PYSPARK_PYTHON=python3” --conf “spark.yarn.appMasterEnv.PYSPARK_PYTHON=python3” 设置。driver端还可以通过export PYSPARK_PYTHON=python3 设置环境变量。

若需要上传pyhton包,可以通过 --archive python.tar.gz 的方式上传。

为了使应用能够获取到py脚本文件,还需要在启动命令中添加 --file pythonFile.py 将python脚本上传到 yarn 上。

运行命令参考如下:

spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.RunPythonExample --files /usr/local/test.py --conf "spark.executorEnv.PYSPARK_PYTHON=python3" --conf "spark.yarn.appMasterEnv.PYSPARK_PYTHON=python3" /usr/local/test.jar test.py test.py

如果需要使用其他python环境,而非节点上已安装的,可以通过 --archives 上传python压缩包,再通过环境变量指定pythonExec,例如:

spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.RunPythonExample --files /usr/local/test.py --archives /usr/local/python.tar.gz#myPython --conf "spark.executorEnv.PYSPARK_PYTHON=myPython/bin/python3" --conf "spark.yarn.appMasterEnv.PYSPARK_PYTHON=myPython/bin/python3" /usr/local/test.jar test.py test.py

本文由华为云发布。文章来源地址https://www.toymoban.com/news/detail-423242.html

到了这里,关于如何在Spark Scala/Java应用中调用Python脚本的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • DevOps系列文章之 java调用python脚本

    在java类中直接执行python语句 在java类中直接调用本地python脚本 使用Runtime.getRuntime()执行python脚本文件(推荐) 调用python脚本中的函数 官网地址 首页 | (jython.org) Jython项目提供了Java中的Python实现, 为Python提供了在JVM上运行的好处,并可以访问用Java编写的类。 当前版本(Jyt

    2024年02月14日
    浏览(44)
  • Spark项目Java和Scala混合打包编译

    实际开发用有时候引用自己写的一些java工具类,但是整个项目是scala开发的spark程序,在项目打包时需要考虑到java和scala混合在一起编译。 今天看到之前很久之前写的一些打包编译文章,发现很多地方不太对,于是重新整理更新如下。 我们的项目结构可能如下图,既包含ja

    2024年02月11日
    浏览(43)
  • 【spark】java类在spark中的传递,scala object在spark中的传递

    记录一个比较典型的问题,先讲一下背景,有这么一个用java写的类 然后在spark中使用的时候: 原因: scala的object对应的就是java的静态成员,可以反过来理解java的所有静态成员可被抽取成伴生对象(虽然现实中是scala最终编译成java)。以上面的JavaClass0 例子可理解为等价的

    2024年02月11日
    浏览(36)
  • spark 经典demo 的 scala 和 java 实现

    💐💐扫码关注公众号,回复 spark 下载geekbang 原价 90 元 零基础入门 Spark 学习资料💐💐 要先对文件中的单词做统计计数,然后再打印出频次最高的 5 个单词,江湖人称“Word Count”wikiOfSpark.txt 文件下载地址:这里 scala 实现 java实现  为了限制机动车保有量,从 2011 年

    2024年04月27日
    浏览(36)
  • 大数据平台开发——使用Java和Python调用Shell脚本

    大数据平台开发——使用Java和Python调用Shell脚本 在大数据平台开发中,经常会遇到需要调用Shell脚本的场景,倒不是说只能用Shell,毕竟大数据开发到头来一定是个语言无关的事情: 从Hive源码解读大数据开发为什么可以脱离SQL、Java、Scala:https://lizhiyong.blog.csdn.net/article/deta

    2023年04月09日
    浏览(47)
  • 用idea工具scala 和 Java开发 spark案例:WordCount

    目录 一 环境准备 二 scala代码编写 三 java 代码编写         创建一个 maven 工程         添加下列依赖         原本就下载过这些依赖的没必要再下一遍,可以用之前的,比如 json,mysql,mysq 这里版本是 mysql 5 ,不一样的注意修改                  首先准备好数据,即

    2024年02月07日
    浏览(59)
  • Spark—通过Java、Scala API实现WordCount案例的基本操作

    实验原理 Spark的核心就是RDD,所有在RDD上的操作会被运行在Cluster上,Driver程序启动很多Workers,Workers在(分布式)文件系统中读取数据后转化为RDD(弹性分布式数据集),然后对RDD在内存中进行缓存和计算。 而RDD有两种类型的操作 ,分别是Action(返回values)和Transformations(

    2024年02月15日
    浏览(41)
  • 【云计算】3台虚拟机完成Spark Yarn集群部署并编写Scala应用程序实现单词计数统计

    目录 1.准备环境          2.安装spark并配置环境 3.安装scala并配置环境 4.安装编辑器idea 5.编写Spark Scala应用程序实现单词计数统计 6.Spark On Yarn配置 虚拟机:vmware workstation16 linux版本:centOS 7 linux 分布式环境:hadoop 3.1.1 (1)创建三台虚拟机,并准备好linux环境和hadoop,确保h

    2023年04月15日
    浏览(45)
  • Java调用scala中map转换问题处理

    网上代码 把Javamap转为scala的map代码 我自身是用scala2.11.8, jdk8, 同样代码在本机报错如下: toSeq()方法就一个方法还会不明确也是奇葩, 遂去翻源码,然后发现seq和其他集和对象关系  那么算了不去转到seq了, 我转到他的子类总改明确吧, 遂把代码改成以下两种方式分别实验:  另外

    2024年02月11日
    浏览(40)
  • Kafka传输数据到Spark Streaming通过编写程序java、scala程序实现操作

    现有一电商网站数据文件,名为buyer_favorite1,记录了用户对商品的收藏数据,数据以“t”键分割,数据内容及数据格式如下: 项目环境说明 开启hadoop集群,zookeeper服务,开启kafka服务。再另开启一个窗口,在/apps/kafka/bin目录下创建一个topic。 1、新创一个文件folder命名为li

    2024年02月13日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包