sparksql读取hive_sparkhive与hadoop

ke03:

<configuration>

<property>
        <name>hive.metastore.warehouse.dir</name>
        <value>/user/hive/warehouse</value>
</property>
<property>
        <name>javax.jdo.option.ConnectionURL</name>
        <value>jdbc:mysql://ke01:3306/hive?createDatabaseIfNotExist=true</value>
</property>
<property>
        <name>javax.jdo.option.ConnectionDriverName</name>
        <value>com.mysql.jdbc.Driver</value>
</property>
<property>
        <name>javax.jdo.option.ConnectionUserName</name>
        <value>root</value>
</property>
<property>
        <name>javax.jdo.option.ConnectionPassword</name>
        <value>aa123456</value>
</property>
<property>
    <name>hive.metastore.schema.verification</name>
    <value>false</value>
 </property>
<property>
    <name>datanucleus.schema.autoCreateAll</name>
    <value>true</value>
 </property>

</configuration>


ke01.02.04:
<configuration>
 <property>
        <name>hive.metastore.uris</name>
        <value>thrift://ke03:9083</value>
 </property>
</configuration>

ke03:

hive --help // 技巧
hive --service --help //技巧
启动
hive --service metastore


ke04: hive = hive --service cli 连接hive   
// CLI指的是 command-line interface






测试：
create table test01( name string );


原理： ke04 通过command-line interface 将sql写进去，动用元数据，将建表语句通知ke03:metastore 最终存入mysql
mysql -> hive.TBL 有刚创建的表：test01

MetaException(message:Version information not found in metastore. )



配置：
<property>
    <name>hive.metastore.schema.verification</name>
    <value>false</value>
 </property>
<property>
    <name>datanucleus.schema.autoCreateAll</name>
    <value>true</value>
 </property>

ke02: hive --service hiveserver2
beeline 
!connect jdbc:hive2://ke02:10000
root  密码随机

测试： show databases

连接hive方式
1. hive --service cli 需要本地配置hive.metastore.uris
2. beeline  !connect jdbc:hive2://ke02:10000  只要能和hiveserver2通讯即可

   val session = SparkSession.builder()
      .master("local")
      .appName("spark on hive")
      .config("hive.metastore.uris", "thrift://ke03:9083")
      .enableHiveSupport() // 开启hive支持
      .getOrCreate()

    val sc = session.sparkContext
    sc.setLogLevel("ERROR")
    session.catalog.listTables().show()
    /**

+——+——–+———–+———+———–+
| name|database|description|tableType|isTemporary|
+——+——–+———–+———+———–+
|test01| default| null| MANAGED| false|
|test02| default| null| MANAGED| false|
+——+——–+———–+———+———–+

*/


    val dataFrame = session.sql("show tables ")
    dataFrame.show()

    /**

+——–+———+———–+
|database|tableName|isTemporary|
+——–+———+———–+
| default| test01| false|
| default| test02| false|
+——–+———+———–+

*/

   // session.sql("create table test02(id int)")
    // 数据存入HDFS，配置文件需要加入HDFS地址，resources目录下加入core-site.xml hdfs-site.xml
    session.sql("insert into test02 values (3),(4),(5)")


      import session.implicits._
    val df = List(
      "小柯",
      "张三",
      "lisi").toDF("name")
    df.createTempView("test03")


    /**
     * ooxx表虽然没有写入到hdfs、mysql中，但是可以加入到metastore中，成为临时表

+——+——–+———–+———+———–+
| name|database|description|tableType|isTemporary|
+——+——–+———–+———+———–+
|test01| default| null| MANAGED| false|
|test02| default| null| MANAGED| false|
|test03| default| null| MANAGED| false|
|test03| null| null|TEMPORARY| true|
+——+——–+———–+———+———–+

*/

    session.catalog.listTables().show()
    session.sql("show tables ").show()

    /**
     * ooxx是临时表

+——–+———+———–+
|database|tableName|isTemporary|
+——–+———+———–+
| default| test01| false|
| default| test02| false|
| default| test03| false|
| | test03| true|
+——–+———+———–+

*/

    /**
     * mysql 表TBLS数据
     * 主键  创建时间  DBS表      角色       输入输出格式表主键       表名       表类型
     * 1    1613414690      1       0      root    0    0       1                    test01      MANAGED_TABLE
     * 2    1613416318      1        0      root    0    0       2                    test02      MANAGED_TABLE
     *
     * DBS表        排序                            数据存储地址
     * 1        Default Hive database        hdfs://mycluster/user/hive/warehouse    default    public    ROLE
     */
    // test03写入到了hdfs中，临时表变成了持久表
    df.write.saveAsTable("test03")