spark程序中调用shell脚本

神秘的寇先森

发布于 2019-03-11 11:29:01

2.4K0

发布于 2019-03-11 11:29:01

文章被收录于专栏：Java进阶之路Java进阶之路

scala直接调用shell脚本是不行的，但是可以利用java调用shell脚本然后在spark代码中引入java代码实现。

参考：java代码调用shell脚本
shell脚本必须在spark的driver端调用，在worker端只能处理数据。因此必须在spark的DAG引擎开始或者结束以及两个job之间调用shell脚本。
根据以上前提，spark执行shell脚本代码只能在以下几个位置：
SparkContext创建之前和创建之后但是创建RDD之前。
每一个job的Spark的action函数执行之后以及下一个job的transformation函数执行之前。
最后一个job的action函数执行结束之后。

demo

先写一个java类，拥有一个调用shell脚本的方法。

public class MyJavaClass {

    public void executeShell(String shpath) {
        try {
            Process ps = Runtime.getRuntime().exec(shpath);
            ps.waitFor();
            BufferedReader br = new BufferedReader(new InputStreamReader(ps.getInputStream()));
            StringBuffer sb = new StringBuffer();
            String line;
            while ((line = br.readLine()) != null) {
                sb.append(line).append("\n");
            }
            String result = sb.toString();
            System.out.println(result);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

spark程序的主类

object SparkWordCount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("wordCount").setMaster("yarn")
    val sc=new SparkContext(conf)

    val rdd=sc.textFile("hdfs:///test/Hamlet.txt")
    val rdd1=rdd.flatMap(x=>x.split(" "))
      .filter(x=>x.size>1)
      .map(x=>(x,1))
      .reduceByKey(_+_)
      .map{case(x,y)=>(y,x)}
    rdd1.sortByKey(false)
      .map{case(a,b)=>(b,a)}
      .saveAsTextFile("hdfs:///test/output.txt")//saveAsTextFile是个action，真正开始提交job，
    // 调用shell脚本
    val shpath = "/data/spark-runshell.sh";
    val javaClass = new MyJavaClass()
    val addResult = javaClass.executeShell(shpath)
    println(addResult);
  }
}

注意：在maven工程中，scala包里不能写java类，否则编译不通过，必须将java类放在java包里才能编译通过。

jar包提交到spark集群

#!/bin/bash
spark-submit \
--master yarn \
--class com.kouyy.test.SparkWordCount /data/sparkdemo-1.0-SNAPSHOT.jar

服务器上spark程序Jar包及运行脚本.png

spark-runshell.sh内容

#!/bin/bash
echo 'spark调用shell脚本运行成功'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'hahahahahhaha'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'
echo 'spark调用shell脚本运行成功'