使用Crontab定时执行Spark任务

Posted 2022-06-20 Trigl

今天的主要内容有：

1. Linux下使用定时器crontab
2. Linux下如何编写Perl脚本
3. 在Java程序中调用Linux命令
4. 实例：每天0点30分执行Spark任务

1. Linux下使用定时器crontab

1、安装

yum -y install vixie-cron
yum -y install crontabs

2、启停命令

service crond start     //启动服务
service crond stop      //关闭服务
service crond restart   //重启服务
service crond reload    //重新载入配置
service crond status    //查看crontab服务状态

3、查看所有定时器任务

crontab -l

这个定时器任务是每分钟用sh执行test.sh脚本

4、添加定时器任务

crontab -e

5、crontab的时间表达式

基本格式 :

*　　*　　*　　*　　*　　command
分　 时　 日　 月　  周　   命令

6、常用实例

// 每分钟执行一次
   *  *  *  *  * 

// 每隔一小时执行一次
    00  *  *  *  *   
    * */1 * * *  (/表示频率)

// 每小时的15和45分各执行一次 
    15,45 * * * * （,表示并列）

// 在每天上午 8- 11时中间每小时 15 ，45分各执行一次
    15,45 8-11 * * * command （-表示范围）

// 每个星期一的上午8点到11点的第3和第15分钟执行
    3,15 8-11 * * 1 command

// 每隔两天的上午8点到11点的第3和第15分钟执行
    3,15 8-11 */2 * * command

2. Linux下编写Perl脚本

1、首先安装Perl

yum -y install gcc gcc-c++  make automake autoconf libtool perl

2、写一个最简单的Perl脚本

vi test.pl

内容如下：

#!/usr/bin/perl

use strict;
print "HellonWorld!\\n";

第一个“#”表示是这一行是注释
第二个“!”表示这一行不是普通注释，而是解释器路径的声明行
后面的“/usr/bin/perl”是perl解释器的安装路径，也有可能是：“/usr/local/bin/perl”，如果那个不行，就换这个
use strict是严格检查语法

3、给该脚本添加可执行权限

chmod 764 test.pl

Linux档案的基本权限有9个，分别是owner/group/other三种身份各有read/write/execute权限，而各权限的分数对照是r:4,w:2,x:1;
每种身份的权限都是需要累加的，比如当权限是[-rwxrwx—]，则表明：
owner:rwx=4+2+1=7
group:rwx=4+2+1=7
other:—=0+0+0=0
即该档案的权限数字就是770

4、然后执行该Perl文件即可

./test.pl

由于我们声明了解释器路径，所以不需要使用perl test.pl，而是直接 ./ 就可以执行了

3. 在Java程序中调用Linux命令

主要用到两个类Process和Runtime，代码示例如下：

    Runtime rt = Runtime.getRuntime();
    String[] cmd =  "/bin/sh", "-c", "cd ~" ;
    Process proc = rt.exec(cmd);
    proc.waitFor();
    proc.destroy();

如果-c 选项存在，命令就从字符串中读取

4. 实例：每天0点30分执行Spark任务

1、首先编写执行Spark任务的Perl脚本：getappinfo.pl

#!/usr/bin/perl

use strict;

# 获取上一天的日期
my ($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = localtime(time - 3600 * 24);
# $year是从1900开始计数的，所以$year需要加上1900；
$year += 1900;
# $mon是从0开始计数的，所以$mon需要加上1；
$mon += 1;


print "$year-$mon-$mday-$hour-$min-$sec, wday: $wday, yday: $yday, isdst: $isdst\\n";

sub exec_spark

    my $dst_date = sprintf("%d%02d%02d", $year, $mon, $mday);
    my $spark_generateapp = "nohup /data/install/spark-2.0.0-bin-hadoop2.7/bin/spark-submit  --master spark://hxf:7077  --executor-memory 30G --executor-cores 24  --conf spark.default.parallelism=300 --class com.analysis.main.GenAppInfo  /home/hadoop/jar/analysis.jar $dst_date > /home/hadoop/logs/genAppInfo.log &";
    print "$spark_generateapp\\n";

    return system($spark_generateapp);


if (!exec_spark())

    print "done\\n";
    exit(0);

2、添加定时器任务：每天的0点30分执行getappinfo.pl

crontab -e

添加以下内容：

30 0 * * * /data/tools/getappinfo.pl

3、脚本中的Spark程序如下：

package com.analysis.main

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object TestCrontab 

  // args -> 20170101
  def main(args: Array[String]) 

    if (args.length == 0) 
      System.err.println("参数异常")
      System.exit(1)
    

    val year = args(0).substring(0, 4)
    val month = args(0).substring(4, 6)
    val day = args(0).substring(6, 8)

    //设置序列化器为KryoSerializer,也可以在配置文件中进行配置
    System.setProperty("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    // 设置应用名称，新建Spark环境
    val sparkConf = new SparkConf().setAppName("GenerateAppInfo_" + args(0))
    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()
      .getOrCreate()
    println("Start " + "GenerateAppInfo_" + args(0))

    import spark.sql

    sql("use arrival")
    val sqlStr = "select opttime, firstimei, secondimei, thirdimei, applist, year, month, day from base_arrival where year=" + year + " and month=" + month + " and day=" + day
    sql(sqlStr).show()

    // 跑GenAppInfoNew
    val rt = Runtime.getRuntime()
    val cmd = Array("/bin/sh", "-c", "/data/tools/getappinfo_new.pl")
    try 
      val proc = rt.exec(cmd)
      proc.waitFor()
      proc.destroy()
      println("执行提取appinfo_new任务")
     catch 
      case e: Exception => println("执行提取appinfo_new任务失败：" + e.getMessage())
    
  

这个程序首先从Hive中查询数据并展示出来，然后再调用Linux的shell执行另一个Perl脚本getappinfo_new.pl，我们可以在这个脚本中写入其他操作