Hive Metastore-spark技术分享

Hive Metastore

Spark SQL uses a Hive metastore to manage the metadata of persistent relational entities (e.g. databases, tables, columns, partitions) in a relational database (for fast access).

A Hive metastore warehouse (aka spark-warehouse) is the directory where Spark SQL persists tables whereas a Hive metastore (aka metastore_db) is a relational database to manage the metadata of the persistent relational entities, e.g. databases, tables, columns, partitions.

By default, Spark SQL uses the embedded deployment mode of a Hive metastore with a Apache Derby database.

Important

The default embedded deployment mode is not recommended for production use due to limitation of only one active SparkSession at a time.

Read Cloudera’s Configuring the Hive Metastore for CDH document that explains the available deployment modes of a Hive metastore.

When SparkSession is created with Hive support the external catalog (aka metastore) is HiveExternalCatalog. HiveExternalCatalog uses spark.sql.warehouse.dir directory for the location of the databases and javax.jdo.option properties for the connection to the Hive metastore database.

Note	The metadata of relational entities is persisted in a metastore database over JDBC and DataNucleus AccessPlatform that uses javax.jdo.option properties. Read Hive Metastore Administration to learn how to manage a Hive Metastore.

Name Description

javax.jdo.option.ConnectionURL

The JDBC connection URL of a Hive metastore database to use



jdbc:derby:;databaseName=metastore_db;create=true

jdbc:derby:memory:;databaseName=${metastoreLocation.getAbsolutePath};create=true

jdbc:mysql://192.168.175.160:3306/metastore?useSSL=false

jdbc:derby:;databaseName=metastore_db;create=true

jdbc:derby:memory:;databaseName=${metastoreLocation.getAbsolutePath};create=true

jdbc:mysql://192.168.175.160:3306/metastore?useSSL=false

javax.jdo.option.ConnectionDriverName

The JDBC driver of a Hive metastore database to use



org.apache.derby.jdbc.EmbeddedDriver

org.apache.derby.jdbc.EmbeddedDriver

javax.jdo.option.ConnectionUserName

The user name to use to connect to a Hive metastore database

javax.jdo.option.ConnectionPassword

The password to use to connect to a Hive metastore database

You can configure javax.jdo.option properties in hive-site.xml or using options with spark.hadoop prefix.

You can access the current connection properties for a Hive metastore in a Spark SQL application using the Spark internal classes.



scala> :type spark
org.apache.spark.sql.SparkSession

scala> spark.sharedState.externalCatalog
res1: org.apache.spark.sql.catalyst.catalog.ExternalCatalog = org.apache.spark.sql.hive.HiveExternalCatalog@79dd79eb

// Use `:paste -raw` to paste the following code
// This is to pass the private[spark] "gate"
// BEGIN
package org.apache.spark
import org.apache.spark.sql.SparkSession
object jacek {
  def open(spark: SparkSession) = {
    import org.apache.spark.sql.hive.HiveExternalCatalog
    spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client
  }
}
// END
import org.apache.spark.jacek
val hiveClient = jacek.open(spark)
scala> hiveClient.getConf("javax.jdo.option.ConnectionURL", "")
res2: String = jdbc:derby:;databaseName=metastore_db;create=true

scala> :type spark

org.apache.spark.sql.SparkSession

scala> spark.sharedState.externalCatalog

res1: org.apache.spark.sql.catalyst.catalog.ExternalCatalog = org.apache.spark.sql.hive.HiveExternalCatalog@79dd79eb

// Use `:paste -raw` to paste the following code

// This is to pass the private[spark] "gate"

// BEGIN

package org.apache.spark

import org.apache.spark.sql.SparkSession

object jacek {

def open(spark: SparkSession) = {

import org.apache.spark.sql.hive.HiveExternalCatalog

spark.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog].client

}

// END

import org.apache.spark.jacek

val hiveClient = jacek.open(spark)

scala> hiveClient.getConf("javax.jdo.option.ConnectionURL", "")

res2: String = jdbc:derby:;databaseName=metastore_db;create=true

The benefits of using an external Hive metastore:

Allow multiple Spark applications (sessions) to access it concurrently
Allow a single Spark application to use table statistics without running “ANALYZE TABLE” every execution

Note	As of Spark 2.2 (see SPARK-18112 Spark2.x does not support read data from Hive 2.x metastore) Spark SQL supports reading data from Hive 2.1.1 metastore.

Caution

FIXME Describe hive-site.xml vs config method vs --conf with spark.hadoop prefix.

Spark SQL uses the Hive-specific configuration properties that further fine-tune the Hive integration, e.g. spark.sql.hive.metastore.version or spark.sql.hive.metastore.jars.

`spark.sql.warehouse.dir` Configuration Property

spark.sql.warehouse.dir is a static configuration property that sets Hive’s hive.metastore.warehouse.dir property, i.e. the location of the Hive local/embedded metastore database (using Derby).

Tip	Refer to SharedState to learn about (the low-level details of) Spark SQL support for Apache Hive. See also the official Hive Metastore Administration document.

Hive Metastore Deployment Modes

Configuring External Hive Metastore in Spark SQL

In order to use an external Hive metastore you should do the following:

Enable Hive support in SparkSession (that makes sure that the Hive classes are on CLASSPATH and sets spark.sql.catalogImplementation internal configuration property to hive)
spark.sql.warehouse.dir required?
Define hive.metastore.warehouse.dir in hive-site.xml configuration resource
Check out warehousePath
Execute ./bin/run-example sql.hive.SparkHiveExample to verify Hive configuration

When not configured by the hive-site.xml, SparkSession automatically creates metastore_db in the current directory and creates a directory configured by spark.sql.warehouse.dir, which defaults to the directory spark-warehouse in the current directory that the Spark application is started.

Note	`hive.metastore.warehouse.dir` property in `hive-site.xml` is deprecated since Spark 2.0.0. Use spark.sql.warehouse.dir to specify the default location of the databases in a Hive warehouse. You may need to grant write privilege to the user who starts the Spark application.

Hadoop Configuration Properties for Hive

Name Description

hive.metastore.uris

The Thrift URI of a remote Hive metastore, i.e. one that is in a separate JVM process or on a remote node



config("hive.metastore.uris", "thrift://192.168.175.160:9083")

config("hive.metastore.uris", "thrift://192.168.175.160:9083")

hive.metastore.warehouse.dir

SharedState uses hive.metastore.warehouse.dir to set spark.sql.warehouse.dir if the latter is undefined.

Caution

FIXME How is hive.metastore.warehouse.dir related to spark.sql.warehouse.dir? SharedState.warehousePath? Review https://github.com/apache/spark/pull/16996/files

hive.metastore.schema.verification

Set to false (as seems to cause exceptions with an empty metastore database as of Hive 2.1)

You may also want to use the following Hive configuration properties that (seem to) cause exceptions with an empty metastore database as of Hive 2.1.

datanucleus.schema.autoCreateAll set to true

spark.hadoop Configuration Properties

Caution

FIXME Describe the purpose of spark.hadoop.* properties

You can specify any of the Hadoop configuration properties, e.g. hive.metastore.warehouse.dir with spark.hadoop prefix.



$ spark-shell --conf spark.hadoop.hive.metastore.warehouse.dir=/tmp/hive-warehouse
...
scala> spark.sharedState
18/01/08 10:46:19 INFO SharedState: spark.sql.warehouse.dir is not set, but hive.metastore.warehouse.dir is set. Setting spark.sql.warehouse.dir to the value of hive.metastore.warehouse.dir ('/tmp/hive-warehouse').
18/01/08 10:46:19 INFO SharedState: Warehouse path is '/tmp/hive-warehouse'.
res1: org.apache.spark.sql.internal.SharedState = org.apache.spark.sql.internal.SharedState@5a69b3cf

$ spark-shell --conf spark.hadoop.hive.metastore.warehouse.dir=/tmp/hive-warehouse

...

scala> spark.sharedState

18/01/08 10:46:19 INFO SharedState: spark.sql.warehouse.dir is not set, but hive.metastore.warehouse.dir is set. Setting spark.sql.warehouse.dir to the value of hive.metastore.warehouse.dir ('/tmp/hive-warehouse').

18/01/08 10:46:19 INFO SharedState: Warehouse path is '/tmp/hive-warehouse'.

res1: org.apache.spark.sql.internal.SharedState = org.apache.spark.sql.internal.SharedState@5a69b3cf

hive-site.xml Configuration Resource

hive-site.xml configures Hive clients (e.g. Spark SQL) with the Hive Metastore configuration.

hive-site.xml is loaded when SharedState is created (which is…FIXME).

Configuration of Hive is done by placing your hive-site.xml, core-site.xml (for security configuration),
and hdfs-site.xml (for HDFS configuration) file in conf/ (that is automatically added to the CLASSPATH of a Spark application).

Tip	You can use `--driver-class-path` or `spark.driver.extraClassPath` to point to the directory with configuration resources, e.g. `hive-site.xml`.



<configuration>
  <property>
    <name>hive.metastore.warehouse.dir</name>
    <value>/tmp/hive-warehouse</value>
    <description>Hive Metastore location</description>
  </property>
</configuration>

<name>hive.metastore.warehouse.dir</name>

<value>/tmp/hive-warehouse</value>

<description>Hive Metastore location</description>

</property>

</configuration>

Tip	Read Resources section in Hadoop’s Configuration javadoc to learn more about configuration resources.

Tip

Use SparkContext.hadoopConfiguration to know which configuration resources have already been registered.



scala> sc.hadoopConfiguration
res1: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml

// Initialize warehousePath
scala> spark.sharedState.warehousePath
res2: String = file:/Users/jacek/dev/oss/spark/spark-warehouse/

// Note file:/Users/jacek/dev/oss/spark/spark-warehouse/ is added to configuration resources
scala> sc.hadoopConfiguration
res3: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml, file:/Users/jacek/dev/oss/spark/conf/hive-site.xml

scala> sc.hadoopConfiguration

res1: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml

// Initialize warehousePath

scala> spark.sharedState.warehousePath

res2: String = file:/Users/jacek/dev/oss/spark/spark-warehouse/

// Note file:/Users/jacek/dev/oss/spark/spark-warehouse/ is added to configuration resources

scala> sc.hadoopConfiguration

res3: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml, file:/Users/jacek/dev/oss/spark/conf/hive-site.xml

Enable org.apache.spark.sql.internal.SharedState logger to INFO logging level to know where hive-site.xml comes from.



scala> spark.sharedState.warehousePath
18/01/08 09:49:33 INFO SharedState: loading hive config file: file:/Users/jacek/dev/oss/spark/conf/hive-site.xml
18/01/08 09:49:33 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/Users/jacek/dev/oss/spark/spark-warehouse/').
18/01/08 09:49:33 INFO SharedState: Warehouse path is 'file:/Users/jacek/dev/oss/spark/spark-warehouse/'.
res2: String = file:/Users/jacek/dev/oss/spark/spark-warehouse/

scala> spark.sharedState.warehousePath

18/01/08 09:49:33 INFO SharedState: loading hive config file: file:/Users/jacek/dev/oss/spark/conf/hive-site.xml

18/01/08 09:49:33 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/Users/jacek/dev/oss/spark/spark-warehouse/').

18/01/08 09:49:33 INFO SharedState: Warehouse path is 'file:/Users/jacek/dev/oss/spark/spark-warehouse/'.

res2: String = file:/Users/jacek/dev/oss/spark/spark-warehouse/

Starting Hive

The following steps are for Hive and Hadoop 2.7.5.



$ ./bin/hdfs version
Hadoop 2.7.5
Subversion https://shv@git-wip-us.apache.org/repos/asf/hadoop.git -r 18065c2b6806ed4aa6a3187d77cbe21bb3dba075
Compiled by kshvachk on 2017-12-16T01:06Z
Compiled with protoc 2.5.0
From source with checksum 9f118f95f47043332d51891e37f736e9
This command was run using /Users/jacek/dev/apps/hadoop-2.7.5/share/hadoop/common/hadoop-common-2.7.5.jar

$ ./bin/hdfs version

Hadoop 2.7.5

Subversion https://shv@git-wip-us.apache.org/repos/asf/hadoop.git -r 18065c2b6806ed4aa6a3187d77cbe21bb3dba075

Compiled by kshvachk on 2017-12-16T01:06Z

Compiled with protoc 2.5.0

From source with checksum 9f118f95f47043332d51891e37f736e9

This command was run using /Users/jacek/dev/apps/hadoop-2.7.5/share/hadoop/common/hadoop-common-2.7.5.jar

Tip	Read the section Pseudo-Distributed Operation about how to run Hadoop HDFS “on a single-node in a pseudo-distributed mode where each Hadoop daemon runs in a separate Java process.”

Tip

Use hadoop.tmp.dir configuration property as the base for temporary directories.



<property>
  <name>hadoop.tmp.dir</name>
  <value>/tmp/my-hadoop-tmp-dir/hdfs/tmp</value>
  <description>The base for temporary directories.</description>
</property>

<name>hadoop.tmp.dir</name>

<value>/tmp/my-hadoop-tmp-dir/hdfs/tmp</value>

<description>The base for temporary directories.</description>

</property>

Use ./bin/hdfs getconf -confKey hadoop.tmp.dir to check out the value



$ ./bin/hdfs getconf -confKey hadoop.tmp.dir
/tmp/my-hadoop-tmp-dir/hdfs/tmp

$ ./bin/hdfs getconf -confKey hadoop.tmp.dir

/tmp/my-hadoop-tmp-dir/hdfs/tmp

Edit etc/hadoop/core-site.xml to add the following:



<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>

<name>fs.defaultFS</name>

<value>hdfs://localhost:9000</value>

</property>

</configuration>

./bin/hdfs namenode -format right after you’ve installed Hadoop and before starting any HDFS services (NameNode in particular)



$ ./bin/hdfs namenode -format
18/01/09 15:48:28 INFO namenode.NameNode: STARTUP_MSG:
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG:   host = japila.local/192.168.1.2
STARTUP_MSG:   args = [-format]
STARTUP_MSG:   version = 2.7.5
...
18/01/09 15:48:28 INFO namenode.NameNode: createNameNode [-format]
...
Formatting using clusterid: CID-bfdc81da-6941-4a93-8371-2c254d503a97
...
18/01/09 15:48:29 INFO common.Storage: Storage directory /tmp/hadoop-jacek/dfs/name has been successfully formatted.
18/01/09 15:48:29 INFO namenode.FSImageFormatProtobuf: Saving image file /tmp/hadoop-jacek/dfs/name/current/fsimage.ckpt_0000000000000000000 using no compression
18/01/09 15:48:29 INFO namenode.FSImageFormatProtobuf: Image file /tmp/hadoop-jacek/dfs/name/current/fsimage.ckpt_0000000000000000000 of size 322 bytes saved in 0 seconds.
18/01/09 15:48:29 INFO namenode.NNStorageRetentionManager: Going to retain 1 images with txid >= 0
18/01/09 15:48:29 INFO util.ExitUtil: Exiting with status 0

$ ./bin/hdfs namenode -format

18/01/09 15:48:28 INFO namenode.NameNode: STARTUP_MSG:

/************************************************************

STARTUP_MSG: Starting NameNode

STARTUP_MSG: host = japila.local/192.168.1.2

STARTUP_MSG: args = [-format]

STARTUP_MSG: version = 2.7.5

...

18/01/09 15:48:28 INFO namenode.NameNode: createNameNode [-format]

...

Formatting using clusterid: CID-bfdc81da-6941-4a93-8371-2c254d503a97

...

18/01/09 15:48:29 INFO common.Storage: Storage directory /tmp/hadoop-jacek/dfs/name has been successfully formatted.

18/01/09 15:48:29 INFO namenode.FSImageFormatProtobuf: Saving image file /tmp/hadoop-jacek/dfs/name/current/fsimage.ckpt_0000000000000000000 using no compression

18/01/09 15:48:29 INFO namenode.FSImageFormatProtobuf: Image file /tmp/hadoop-jacek/dfs/name/current/fsimage.ckpt_0000000000000000000 of size 322 bytes saved in 0 seconds.

18/01/09 15:48:29 INFO namenode.NNStorageRetentionManager: Going to retain 1 images with txid >= 0

18/01/09 15:48:29 INFO util.ExitUtil: Exiting with status 0

Note

Use ./bin/hdfs namenode to start a NameNode that will tell you that the local filesystem is not ready.



$ ./bin/hdfs namenode
18/01/09 15:43:11 INFO namenode.NameNode: STARTUP_MSG:
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG:   host = japila.local/192.168.1.2
STARTUP_MSG:   args = []
STARTUP_MSG:   version = 2.7.5
...
18/01/09 15:43:11 INFO namenode.NameNode: fs.defaultFS is hdfs://localhost:9000
18/01/09 15:43:11 INFO namenode.NameNode: Clients are to use localhost:9000 to access this namenode/service.
...
18/01/09 15:43:12 INFO hdfs.DFSUtil: Starting Web-server for hdfs at: http://0.0.0.0:50070
...
18/01/09 15:43:13 WARN common.Storage: Storage directory /private/tmp/hadoop-jacek/dfs/name does not exist
18/01/09 15:43:13 WARN namenode.FSNamesystem: Encountered exception loading fsimage
org.apache.hadoop.hdfs.server.common.InconsistentFSStateException: Directory /private/tmp/hadoop-jacek/dfs/name is in an inconsistent state: storage directory does not exist or is not accessible.
	at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverStorageDirs(FSImage.java:382)
	at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverTransitionRead(FSImage.java:233)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFSImage(FSNamesystem.java:984)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFromDisk(FSNamesystem.java:686)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.loadNamesystem(NameNode.java:586)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.initialize(NameNode.java:646)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:820)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:804)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.createNameNode(NameNode.java:1516)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.main(NameNode.java:1582)
...
18/01/09 15:43:13 ERROR namenode.NameNode: Failed to start namenode.
org.apache.hadoop.hdfs.server.common.InconsistentFSStateException: Directory /private/tmp/hadoop-jacek/dfs/name is in an inconsistent state: storage directory does not exist or is not accessible.
	at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverStorageDirs(FSImage.java:382)
	at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverTransitionRead(FSImage.java:233)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFSImage(FSNamesystem.java:984)
	at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFromDisk(FSNamesystem.java:686)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.loadNamesystem(NameNode.java:586)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.initialize(NameNode.java:646)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:820)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:804)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.createNameNode(NameNode.java:1516)
	at org.apache.hadoop.hdfs.server.namenode.NameNode.main(NameNode.java:1582)

$ ./bin/hdfs namenode

18/01/09 15:43:11 INFO namenode.NameNode: STARTUP_MSG:

/************************************************************

STARTUP_MSG: Starting NameNode

STARTUP_MSG: host = japila.local/192.168.1.2

STARTUP_MSG: args = []

STARTUP_MSG: version = 2.7.5

...

18/01/09 15:43:11 INFO namenode.NameNode: fs.defaultFS is hdfs://localhost:9000

18/01/09 15:43:11 INFO namenode.NameNode: Clients are to use localhost:9000 to access this namenode/service.

...

18/01/09 15:43:12 INFO hdfs.DFSUtil: Starting Web-server for hdfs at: http://0.0.0.0:50070

...

18/01/09 15:43:13 WARN common.Storage: Storage directory /private/tmp/hadoop-jacek/dfs/name does not exist

18/01/09 15:43:13 WARN namenode.FSNamesystem: Encountered exception loading fsimage

org.apache.hadoop.hdfs.server.common.InconsistentFSStateException: Directory /private/tmp/hadoop-jacek/dfs/name is in an inconsistent state: storage directory does not exist or is not accessible.

at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverStorageDirs(FSImage.java:382)

at org.apache.hadoop.hdfs.server.namenode.FSImage.recoverTransitionRead(FSImage.java:233)

at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFSImage(FSNamesystem.java:984)

at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.loadFromDisk(FSNamesystem.java:686)

at org.apache.hadoop.hdfs.server.namenode.NameNode.loadNamesystem(NameNode.java:586)

at org.apache.hadoop.hdfs.server.namenode.NameNode.initialize(NameNode.java:646)

at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:820)

at org.apache.hadoop.hdfs.server.namenode.NameNode.<init>(NameNode.java:804)

at org.apache.hadoop.hdfs.server.namenode.NameNode.createNameNode(NameNode.java:1516)

at org.apache.hadoop.hdfs.server.namenode.NameNode.main(NameNode.java:1582)

...

18/01/09 15:43:13 ERROR namenode.NameNode: Failed to start namenode.