spark程序入门-第一次写spark程序

迈不过友情╰ 2022-12-20 08:48 204阅读 0赞

第一次写spark,找了很多的资料，终于梳理出来了相关程序和代码，具体如下：

总共两种方式：1种是用java 直接运用sparkSession操作hiveSql，需要spark2以上的依赖，另外一种是用scala编写spark程序，需要准备scala环境，具体小伙伴可以自己网上找教程。

1.  java代码示例

package com.tydic.spark.util;
    
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * Created by cuiyonggang on 2020/11/9.
     */
    public class HiveSqlOnSpark {
    
        /**
         * 通过spark sql 执行hive
         * @param sql
         * @param appname
         * @param master
         * @return
         */
        private List<List<String>> queryHiveSql(String sql, String appname,String master) {
            List<List<String>> returnData = new ArrayList<>();
            System.setProperty("hadoop.home.dir","/etc/hadoop/conf");//hadoop相关配置文件路径
            SparkSession spark = SparkSession.builder().appName(appname).master(master).enableHiveSupport().getOrCreate();//appname，如果是yarn-cluster执行的方式，是任务名称，master一般测试用local，生产用yarn
            Dataset<Row> result = spark.sql(sql);
            List<Row> rowList = result.collectAsList();
            System.out.print(rowList.toString());
            for (Row row : rowList) {
                List<String> list = new ArrayList<>();
                String orderId = row.get(0).toString();
                String userId = row.get(1).toString();
                list.add(orderId);
                list.add(userId);
                returnData.add(list);
            }
            return returnData;
        }
    
    
        public static void main(String args[]) {
            try {
                HiveSqlOnSpark hiveSqlOnSpark = new HiveSqlOnSpark();
    
                String sql="你要执行的sql";
                String appname="queryHiveData";
                String master="local";
                hiveSqlOnSpark.queryHiveSql(sql,appname,master);
    
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    
    }

2.scala代码示例

package hbhiveencry
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.sql.DataFrame
    import org.apache.spark.sql.hive.HiveContext
    import org.apache.spark.{SparkConf, SparkContext}
    import scalikejdbc.config.DBs
    
    /**
      * Created by cuiyg on 2020/11/09
      * 该工具类用于执行hive sql insert overwrite语句
      * submit
      */
    object HiveDataInsertOnSpark {
    
    
      /**
        * 控制输出日志
        */
      Logger.getLogger("org").setLevel(Level.INFO)
    
      private final val evn = "evn_zhuanqu"
    
      //加载配置文件初始化db
      DBs.setup()
      //加载集群hadoop配置文件
      System.setProperty("hadoop.home.dir", "/etc/hadoop/conf")
      //获取spark客户端
      val conf = new SparkConf().setAppName("HiveDataInsertOnSpark")
    
      val sc = new SparkContext(conf)
      val sparkOnHive: HiveContext = new HiveContext(sc)
    
      sparkOnHive.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
      sparkOnHive.setConf("hive.execution.engine", "tez")
      sparkOnHive.setConf("hive.exec.dynamic.partition", "true")
      sparkOnHive.setConf("tez.queue.name", "root.default")//有权限的队列名称
      sparkOnHive.setConf("hive.exec.max.dynamic.partitions", "100000")//动态分区数
      sparkOnHive.setConf("hive.exec.max.dynamic.partitions.pernode", "100000")
      sparkOnHive.setConf("hive.mapred.mode", "nonstrict")//矫检模式
      sparkOnHive.setConf("hive.exec.compress.output", "true")
      sparkOnHive.setConf("mapred.output.compress", "true")
      sparkOnHive.setConf("mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec")
      sparkOnHive.setConf("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec")
      sparkOnHive.setConf("hive.merge.mapfiles", "true")
      sparkOnHive.setConf("spark.sql.hive.verifyPartitionPath", "true")
    
      def main(args: Array[String]): Unit = {
        println(String.format("spark task start"))
        encryOnInsertSpark(args)
        println(String.format("spark task end"))
      }
    
      def encryOnInsertSpark(args: Array[String]): Unit = {
    
         sparkOnHive.setConf("spark.sql.hive.verifyPartitionPath", "true")//检查showpartitions中的分区在hdfs中是否有对应的路径
          try {
            val SQL: String ="你要执行的sparksql";
            println(String.format("要执行的SQL为 '%s'", SQL))
            sparkOnHive.sql(SQL)
          } catch {
            case ex: Throwable => println("found a unknown exception" + ex)
          }
        //统计count,并插入记录
        countSQL()
      }
      //查询表总数
      def countSQL():Unit= {
        var countSQL = "select count(1) from lj_pro2_role2.tf_analysis_result_h_test_11 where month_id='201807'";
        println(String.format("countSQL::::::::::::::::::: %s", countSQL))
        var df: DataFrame = sparkOnHive.sql(countSQL)
        df.show()
        var count = df.collect()(0).getAs[Long](0)
        println(String.format(":::::count::::::::::::::::::: "+ count))
      }
    
    
    }

3运用的maven

>     <dependency>
>         <groupId>org.apache.spark</groupId>
>         <artifactId>spark-core_2.11</artifactId>
>         <version>${spark.version}</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.spark</groupId>
>         <artifactId>spark-sql_2.11</artifactId>
>         <version>${spark.version}</version>
>     </dependency>
>     <dependency>
>         <groupId>org.apache.spark</groupId>
>         <artifactId>spark-hive_2.11</artifactId>
>         <version>${spark.version}</version>
>     </dependency>
> 
>

3开发完成后，直接mvn clean package 打包即可

4.权限

执行spark sql的时候，需要运用有权限执行的集群账号，如果你linux主机上本地的有票据或者环境变量配置了相关票据，执行spark包的时候，不需要指定票据相关的项。

同时，如果用spark-submit提交队列，也需要指定一个有权限的队列。设计权限的就是以上的两方面。

5 提交spark jar包命令解析。

> /usr/local/spark-2.1.1-bin-hadoop2.6/bin/spark-submit --class hbhiveencry.HiveDataInsertOnSpark --master yarn-cluster  --driver-memory 2g  --executor-memory 3g  --executor-cores 2  --num-executors 4  --queue root.zq.tenant.lj\_pro2\_ddqueue /usr/local/spark-2.1.1-bin-hadoop2.6/temp/sparktest-1.0-SNAPSHOT.jar
> 
> 开头是spark安装的路径
> 
> \--class 指定的是你要执行的类，必须要有main方法
> 
> \--master 测试提交spark任务的方式
> 
> \--driver-memory 驱动的内存
> 
> \--executor-memory 执行内存
> 
> \--executor-cores 执行指定的vcore
> 
> \--queue 指定要提交的队列
> 
> 最后是你spark jar包的路径地址

6问题

在测试提交spark jar包任务的时候，也遇到了一些问题，不过都是小问题，把一些典型的问题，简单描述说一下

如果用的yarn的提交方式，如果保存，看日志是有一个application的路径，如果在当前输出日志看不出具体的报错，可以复制该链接，查看具体的log,定位报错。

问题1:会出现。any valid kerberos相关的错误，可能是集群账号的票据过期，你本地环境变量没有配置用户的票据缓存文件或者是没有keytab文件导致的。

问题2:会出现找不到jar包class文件的main方法，我这边执行的时候是因为提交命令写的有问题导致，你可以仔细检查一下自己的提交命令或者检查一下自己的代码。

问题3:会报表不存在的问题，我定位问题时，首先去hive上查看了相应的库表是否存在，如果存在还有问题，可能是你hive配置文件的问题，看一下hive的核心配置文件相关的metastore 的链接地址元数据库的地址是否是你所在集群的地址，如果没有问题，你可以看一下 代码中指定hadoop.home.dir的时候，里面的配置文件是否少了，hive,hadoop和yarn的配置文件必须都要有。我是因为少了hive的配置文件导致的问题，我把配置文件cp过去就可以正常执行。

以上就是第一次写spark程序的相关经验，记录了下来，如果又问题，欢迎指出交流。

笔者不易，希望大家多多支持呢！