我在本地运行spark，并希望访问位于远程Hadoop集群中的Hive表。
我能够通过在SPARK_HOME
下直接启动来访问hive表[ml@master spark-2.0.0]$./bin/beeline
Beeline version 1.2.1.spark2 by Apache Hive
beeline> !connect jdbc:hive2://remote_hive:10000
Connecting to jdbc:hive2://remote_hive:10000
Enter username for jdbc:hive2://remote_hive:10000: root
Enter password for jdbc:hive2://remote_hive:10000: ******
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/home/ml/spark/spark-2.0.0/jars/slf4j-log4j12-1.7.16.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
16/10/12 19:06:39 INFO jdbc.Utils: Supplied authorities: remote_hive:10000
16/10/12 19:06:39 INFO jdbc.Utils: Resolved authority: remote_hive:10000
16/10/12 19:06:39 INFO jdbc.HiveConnection: Will try to open client transport with JDBC Uri: jdbc:hive2://remote_hive:10000
Connected to: Apache Hive (version 1.2.1000.2.4.2.0-258)
Driver: Hive JDBC (version 1.2.1.spark2)
Transaction isolation: TRANSACTION_REPEATABLE_READ
0: jdbc:hive2://remote_hive:10000>
如何从spark编程访问远程hive表?
不需要JDBC
Spark直接连接到Hive metastore，而不是通过HiveServer2。要配置这个，
-
将
hive-site.xml放在classpath上，并指定hive.metastore.uris到hive metastore所在的位置。参见如何在SparkSQL中以编程方式连接到Hive metastore ? -
导入
org.apache.spark.sql.hive.HiveContext，因为它可以在Hive表上执行SQL查询。 -
定义
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) -
验证
sqlContext.sql("show tables")是否有效
Hive表上的SparkSQL
结论:如果你必须使用jdbc方式
查看apache spark与apache hive的远程连接。
请注意，直线也通过jdbc连接。从你的日志中可以看出。
[ml@master spark-2.0.0]美元。Beeline 1.2.1版本。spark2由Apache Hive line> !connect jdbc:hive2://remote_hive:10000
连接jdbc:hive2://remote_hive:10000
所以请看看这篇有趣的文章
- 方法1:使用JDBC将表拉入Spark
- 方法2:使用Spark JdbcRDD与HiveServer2 JDBC驱动程序
- 方法3:在客户端获取数据集，然后手动创建RDD
目前HiveServer2驱动程序不允许我们使用"起泡"方法1和2，我们只能依靠方法3
下面是可以实现的示例代码片段
通过HiveServer2 JDBC连接将数据从一个Hadoop集群(又名"远程")加载到另一个Hadoop集群(我的Spark生活又名"国内")
import java.sql.Timestamp
import scala.collection.mutable.MutableList
case class StatsRec (
first_name: String,
last_name: String,
action_dtm: Timestamp,
size: Long,
size_p: Long,
size_d: Long
)
val conn: Connection = DriverManager.getConnection(url, user, password)
val res: ResultSet = conn.createStatement
.executeQuery("SELECT * FROM stats_201512301914")
val fetchedRes = MutableList[StatsRec]()
while(res.next()) {
var rec = StatsRec(res.getString("first_name"),
res.getString("last_name"),
Timestamp.valueOf(res.getString("action_dtm")),
res.getLong("size"),
res.getLong("size_p"),
res.getLong("size_d"))
fetchedRes += rec
}
conn.close()
val rddStatsDelta = sc.parallelize(fetchedRes)
rddStatsDelta.cache()
// Basically we are done. To check loaded data:
println(rddStatsDelta.count)
rddStatsDelta.collect.take(10).foreach(println)
将HIVE -ste.xml配置提供给SPARK并启动HIVE Metastore服务后，
当连接到HIVE时，在SPARK会话中需要配置两件事:
- 由于Spark SQL使用thrift连接到Hive metastore，所以我们需要在创建Spark会话时提供thrift服务器的uri。
- Hive Metastore warehouse，这是Spark SQL保存表的目录。使用属性spark.sql.warehouse。Dir '对应于'hive.metastore.warehouse '。
类似:
SparkSession spark=SparkSession.builder().appName("Spark_SQL_5_Save To Hive").enableHiveSupport().getOrCreate();
spark.sparkContext().conf().set("spark.sql.warehouse.dir", "/user/hive/warehouse");
spark.sparkContext().conf().set("hive.metastore.uris", "thrift://localhost:9083");
希望这是有帮助的!
根据文档:
注意hive-site.xml中的hive.metastore.warehouse.dir属性自Spark 2.0.0以来已弃用。相反，使用spark.sql.warehouse.dir指定数据库在仓库中的默认位置。
所以在SparkSession中你需要指定spark.sql.uris而不是hive.metastore.uris
from pyspark.sql import SparkSession
spark = SparkSession
.builder
.appName("Python Spark SQL Hive integration example")
.config("spark.sql.uris", "thrift://<remote_ip>:9083")
.enableHiveSupport()
.getOrCreate()
spark.sql("show tables").show()