Spark Scala/Java调用Python算法文件

        python算法中会调用许多额外的其它第三方库，使用pip install xxx安装时会经常因为一些原因（如网络延迟）导致安装失败，这时候可以安装Anaconda3来解决这个问题（Anaconda里面包含了大多数python用到的算法库如numpy、pandas等），下载地址https://www.anaconda.com/products/individual#linux。

        安装完成后，配置环境变量，一共有两个，添加到PATH里面，一个是python地址（E:softwareAnaconda3），一个是python算法库路径（E:softwareAnaconda3scripts），配置完成后重启电脑生效，可以用简单python算法文件验证一下；

      接着在IDEA的plugins中添加python的插件，编写Scala或Java程序就可以调用了，调用代码如下：

package com.hy.model.action

import com.hy.model.util.StringUtils
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.deploy.PythonRunner
import org.apache.spark.sql.SparkSession

object PythonApplication {
  def main(args: Array[String]) {
   
     val modelPath = "F:\projectCode\spark-model\src\main\resources\Health_singlePV_v1.3.py"
     val pyFiles = "Health_singlePV_v1.3.py"
    val ss = SparkSession
      .builder()
      .config("spark.executorEnv.PYSPARK_PYTHON","python3")
      .config("spark.yarn.appMasterEnv.PYSPARK_PYTHON","python3")
      .appName("RunPythonExample")
      .master("local[2]")
      .getOrCreate()


    
    val context = ss.sparkContext

    //val fileSystem = FileSystem.get(context.hadoopConfiguration)

    val modelFSpath = new Path(modelPath)
    val local = "/tmp" + modelPath
    println("local: " + local)
    val localPath = new Path(local)
   // fileSystem.copyToLocalFile(false, modelFSpath, localPath, true)

    runPython(modelPath, pyFiles)


    ss.stop()
  }

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

}

        根据上面Anaconda网址下载linux版本的安装包（是一个大的.sh文件）,使用命令bash Anaconda.xxx.sh命令安装，根据提示输入安装指令，最后将python地址添加到环境变量里面，如下：

PATH=$PATH:$HOME/bin:/opt/soft/anaconda3/bin

      然后将上面的spark程序打成jar包，执行spark程序即可调用。