Spark连接hive的多种方式,并处理数据

ゝ一纸荒年。 2023-03-04 12:29 13阅读 0赞

大家都知道，Spark可以和各种数据库进行连接并读取数据，今天就说说如何连接hive并读取数据。

## 方法一：在IDEA中连接hive ##

**在pom.xml中添加依赖**  
导入mysql的驱动包和spark连接hive的驱动包

<dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <version>5.1.38</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-hive_2.11</artifactId>
         <version>2.3.4</version>    
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-core_2.11</artifactId>
         <version>2.3.4</version>
     </dependency>
    <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-sql_2.11</artifactId>
         <version>2.3.4</version>
    </dependency>

在main主目录下创建一个**resources**文件夹，创建好后右键单击，选择**Make Directory as**,选择**Resources Root**选项，接下来就是把配置文件放到这个文件夹下；  
前面已经讲过配置hadoop和hive，在这里就不多说了；  
把hadoop的配置文件**hdfs-site.xml core-site.xml**和hive的配置文件**hive-site.xml**复制到resources文件夹下；

hadoop的配置文件在hadoop260/etc/hadoop目录下；  
hive的配置文件在hive110/conf目录下  
![在这里插入图片描述][20200803191923461.png]  
如图所示的位置；  
然后在**hive-site.xml**中修改jdbc的url的ip为自己主机的ip地址  
配置好以上信息后，启动hive就不多说了，然后选择一个数据库和其中一张表，先查看数据，查询完有数据时，回到IDEA中，创建scala目录并新建一个object类，接下来开始连接hive

import org.apache.spark.sql.SparkSession
    
    object MyHiveSpark { 
    
      def main(args: Array[String]): Unit = { 
        val spark = SparkSession.builder()
          .master("local[*]")
          .appName("spark_hive")
          .enableHiveSupport()
          .getOrCreate()
        val df = spark.sql("select * from userinfos.user")
        df.show()
      }
    }

**注意：**  
sql语句中有点不一样，就是查询的时候要带上**数据库名.表名**的语法

这样就能查询hive中的数据

## 方法二：在Linux系统下连接hive ##

1.把上面那所说的三个配置文件复制到**spark/conf**的文件下

cp /opt/soft/hive110/conf/hive-site.xml /opt/soft/spark240/conf/
    cp /opt/soft/hadoop260/etc/hadoop/core-site.xml /opt/soft/spark240/conf/
    cp /opt/soft/hadoop260/etc/hadoop/hdfs-site.xml /opt/soft/spark240/conf/

2.把mysql的驱动包和spark连接hive的驱动包复制到spark240/jars文件夹里

3.驱动包可以到本地maven仓库中去找，直接拖到上面所说的地方，在这里就不多说了,也可以去hive的lib下去找驱动包。

4.启动spark并进入shell界面，前面在讲spark安装的时候有说过如何启动，可以自己去看

[Spark简介以及最详细安装教程][Spark]

cd /opt/soft/spark240/bin
    ./spark-shell
    import org.apache.spark.sql.hive.HiveContext
    val hiveContext = new HiveContext(sc)
    hiveContext.sql("select * from test").show()

或者直接使用Spark Session 中创建好的对象spark去直接把hive中的表加载成dataframe。  
只需要在hive-site.xml中添加一个配置就行

<property>
          <name>hive.metastore.uris</name>
          <value>thrift://spark1:9083</value>
    </property>

这样在spark-shell下可直接访问Hive中的表hive\_table

val df=spark.table("hive_table")
    df.printSchema
    df.show

**注意：**  
1.一定要先启动hive元数据服务：hive --service metastore  
2.hive-site.xml拷贝至$\{SPARK\_HOME\}/conf下并添加配置  
![20200803195839524.png][]如图所示，这样就能查询到hive中的数据了！  
**注意：**  
在执行hiveContext.sql(“select \* from test”).show() 会报一个异常：

The root scratch dir: /tmp/hive on HDFS should be writable.
    Current permissions are: rwxr-xr-x

如下图所式：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70]  
**解决办法：**  
**更改HDFS目录/tmp/hive的权限**

hadoop fs -chmod 777 /tmp/hive

同时删HDFS与本地的目录/tmp/hive

hadoop fs -rm -r /tmp/hive

## 方法三：使用SparkSQL连接hive ##

同样先添加依赖就不都说了  
这种方式通过指定config的话就不用再配置那三个.xml文件了，就方便很多

def main(args: Array[String]): Unit = { 
     val Array(inpath, dt, hour) = args
     val conf = new SparkConf().setAppName(this.getClass.getSimpleName)
    
     //.setMaster("local[*]")
     .setMaster("spark://192.168.5.150:7077")
    
     val spark = SparkSession.builder()
     .config(conf)
    
     // 指定hive的metastore的端口 默认为9083 在hive-site.xml中查看
     .config("hive.metastore.uris", "thrift://192.168.5.150:9083")
    
     //指定hive的warehouse目录
     .config("spark.sql.warehouse.dir", "hdfs://192.168.5.150:9000/user/hive/warehouse")
     //直接连接hive
     .enableHiveSupport()
     .getOrCreate()
    
    import spark.implicits._
    val df1: DataFrame = spark.sql("select * from user") 
    df1.printSchema()
    df1.show()

[20200803191923461.png]: /images/20230208/b7cb5ad0f5314993a41d6a467c520db8.png
[Spark]: https://blog.csdn.net/zp17834994071/article/details/107650820
[20200803195839524.png]: /images/20230208/b059da90a6334ba6a6d0b14610afdc66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70]: /images/20230208/cb27854308cc4ae5a58b039f92bbcf93.png