案例实操：Azkaban调度spark作业

柔情只为你懂 2023-10-03 15:36 24阅读 0赞

新建AccessLogDriverCluster类

![img][]

package com.it19gong.clickproject;
    
    
    
    import java.sql.PreparedStatement;
    import java.util.ArrayList;
    import java.util.List;
    
    
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.VoidFunction;
    import org.apache.spark.sql.DataFrame;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.RowFactory;
    import org.apache.spark.sql.SQLContext;
    import org.apache.spark.sql.types.DataTypes;
    import org.apache.spark.sql.types.StructField;
    import org.apache.spark.sql.types.StructType;
    
    
    
    public class AccessLogDriverCluster {
        static DBHelper db1=null;
        public static void main(String[] args) throws Exception {
            
            // 创建SparkConf、JavaSparkContext、SQLContext
                    SparkConf conf = new SparkConf() 
                            .setAppName("RDD2DataFrameProgrammatically");  
                    JavaSparkContext sc = new JavaSparkContext(conf);
                    SQLContext sqlContext = new SQLContext(sc);
                
                    // 第一步，创建一个普通的RDD，但是，必须将其转换为RDD<Row>的这种格式
                    //获取昨天时间
                    JavaRDD<String> lines = sc.textFile("hdfs://node1/data/clickLog/2019/08/31");
                    
                    // 分析一下
                    // 它报了一个，不能直接从String转换为Integer的一个类型转换的错误
                    // 就说明什么，说明有个数据，给定义成了String类型，结果使用的时候，要用Integer类型来使用
                    // 而且，错误报在sql相关的代码中
                    // 所以，基本可以断定，就是说，在sql中，用到age<=18的语法，所以就强行就将age转换为Integer来使用
                    // 但是，肯定是之前有些步骤，将age定义为了String
                    // 所以就往前找，就找到了这里
                    // 往Row中塞数据的时候，要注意，什么格式的数据，就用什么格式转换一下，再塞进去
                    JavaRDD<Row> clickRDD = lines.map(new Function<String, Row>() {
    
                        private static final long serialVersionUID = 1L;
    
                        @Override
                        public Row call(String line) throws Exception {
                            String itr[] = line.split(" ");
                               
                                String ip = itr[0];
                                String date = AnalysisNginxTool.nginxDateStmpToDate(itr[3]);
                                String url = itr[6];
                                String upFlow = itr[9];
                            
                            return RowFactory.create(
                                    ip,
                                    date,
                                    url,
                                    Integer.valueOf(upFlow)
                                    );      
                        }
                        
                    });
                    
                    // 第二步，动态构造元数据
                    // 比如说，id、name等，field的名称和类型，可能都是在程序运行过程中，动态从mysql db里
                    // 或者是配置文件中，加载出来的，是不固定的
                    // 所以特别适合用这种编程的方式，来构造元数据
                    List<StructField> structFields = new ArrayList<StructField>();
                    structFields.add(DataTypes.createStructField("ip", DataTypes.StringType, true));  
                    structFields.add(DataTypes.createStructField("date", DataTypes.StringType, true));  
                    structFields.add(DataTypes.createStructField("url", DataTypes.StringType, true)); 
                    structFields.add(DataTypes.createStructField("upflow", DataTypes.IntegerType, true));  
                    StructType structType = DataTypes.createStructType(structFields);
                    
                    // 第三步，使用动态构造的元数据，将RDD转换为DataFrame
                    DataFrame studentDF = sqlContext.createDataFrame(clickRDD, structType);
                
                    // 后面，就可以使用DataFrame了
                    studentDF.registerTempTable("log");  
                    
                    DataFrame sumFlowDF = sqlContext.sql("select ip,sum(upflow) as sum from log group by ip order by sum desc"); 
                    
                    db1=new DBHelper();
                    final String sql="insert into upflow(ip,sum) values(?,?) ";
                    sumFlowDF.javaRDD().foreach(new VoidFunction<Row>() {
                        
                        @Override
                        public void call(Row t) throws Exception {
                            // TODO Auto-generated method stub
                            PreparedStatement pt = db1.conn.prepareStatement(sql);
                            pt.setString(1,t.getString(0));
                            pt.setString(2,String.valueOf(t.getLong(1)));
                            pt.executeUpdate();
                        }
                    });;
                    
            
        }
    
    }

打包

![img][img 1]

![img][img 2]

报错

![img][img 3]

删除apptest文件

![img][img 4]

再次打包

![img][img 5]

把打好的包拷贝出来

![img][img 6]

并且重命名

![img][img 7]

![img][img 8]

vim project.sh

/opt/modules/spark-1.5.1-bin-hadoop2.6/bin/spark-submit --class com.it19gong.clickproject.AccessLogDriverCluster --num-executors 3 --driver-memory 100m --executor-memory 100m --executor-cores 3 --files /opt/modules/hive/conf/hive-site.xml --driver-class-path /opt/modules/hive/lib/mysql-connector-java-5.1.28.jar /home/hadoop/sparkproject.jar

把原来的包删除

![img][img 9]

上传新的包

![img][img 10]

执行脚本

![img][img 11]

![img][img 12]

mysql数据多了两条

![img][img 13]

打开azkaban的页面，这里再次提醒要用谷歌浏览器

![img][img 14]

新建spark.job文件

#command.job
    type=command
    command=bash project.sh

打包成zip包

![img][img 15]

![img][img 16]

![img][img 17]

![img][img 18]

![img][img 19]

上传zip包

![img][img 20]

![img][img 21]

开始执行

![img][img 22]

![img][img 23]

![img][img 24]

![img][img 25]

mysql数据库多了两天数据

![img][img 26]

参考链接：https://www.cnblogs.com/braveym/p/12259956.html

[img]: https://img-blog.csdnimg.cn/img_convert/20004300c3a3762ed7b937ecb4fa0833.png
[img 1]: https://img-blog.csdnimg.cn/img_convert/5305b22eaae2b624ab927ff7c7f24f22.png
[img 2]: https://img-blog.csdnimg.cn/img_convert/6ca4e2690d3423dcfb2459d57fa23a78.png
[img 3]: https://img-blog.csdnimg.cn/img_convert/03f5603e54437d5036a93fee0040b570.png
[img 4]: https://img-blog.csdnimg.cn/img_convert/82e205695de7ab1c52e352af7b1787dc.png
[img 5]: https://img-blog.csdnimg.cn/img_convert/0d0c965223f798ea0ce83177fa9c45d1.png
[img 6]: https://img-blog.csdnimg.cn/img_convert/d65a6357f50b6de31e8f1b7814dc0760.png
[img 7]: https://img-blog.csdnimg.cn/img_convert/c71301db4080a6831fde56775f2856c4.png
[img 8]: https://img-blog.csdnimg.cn/img_convert/7e4339ff5fd19d7b1f40c06c0d3187f0.png
[img 9]: https://img-blog.csdnimg.cn/img_convert/7ecb7193da16a0dd93686eb859003ebe.png
[img 10]: https://img-blog.csdnimg.cn/img_convert/aead32f1610de9f973f2e8a3fa0360ea.png
[img 11]: https://img-blog.csdnimg.cn/img_convert/12394d326211531cd3539b385c0c2af7.png
[img 12]: https://img-blog.csdnimg.cn/img_convert/ccf75db4439fc61e6d11165573935c21.png
[img 13]: https://img-blog.csdnimg.cn/img_convert/4c73c68e45b5744dc4c22c976a010d0e.png
[img 14]: https://img-blog.csdnimg.cn/img_convert/e4a7efb0d56efc678cba755da4857f3c.png
[img 15]: https://img-blog.csdnimg.cn/img_convert/3eb3f91353a4e8c9707e67308a0ffb18.png
[img 16]: https://img-blog.csdnimg.cn/img_convert/57e8bb52f0e3d00d5ebe5d011a9208b3.png
[img 17]: https://img-blog.csdnimg.cn/img_convert/1f95bac8c490df3bf11d54b4b639610f.png
[img 18]: https://img-blog.csdnimg.cn/img_convert/3f0b08d8718296b2fd277410f990d7c5.png
[img 19]: https://img-blog.csdnimg.cn/img_convert/70f63233d9795bc45518682e58eefebe.png
[img 20]: https://img-blog.csdnimg.cn/img_convert/4c81ea515d57536896bbc34e2185df66.png
[img 21]: https://img-blog.csdnimg.cn/img_convert/4e5fa4628a5cc687fdeace82079490aa.png
[img 22]: https://img-blog.csdnimg.cn/img_convert/41cbe9002af05d16f09dbf3ac6e2186c.png
[img 23]: https://img-blog.csdnimg.cn/img_convert/c27bcf51047ad380f2f1018928c36a7c.png
[img 24]: https://img-blog.csdnimg.cn/img_convert/804f05e81128e5d20f85c7d3af42a448.png
[img 25]: https://img-blog.csdnimg.cn/img_convert/4fe40ac5b3563d849cc707940c46b451.png
[img 26]: https://img-blog.csdnimg.cn/img_convert/5ceb9ee816cb179202bdd7c499e0006a.png