如何从外部PySpark环境连接Databricks托管的Hive元数据存储

核心配置前提 #

• 确保Kubernetes PySpark集群的Spark版本与你使用的Databricks Runtime对应的Spark版本完全一致，避免metastore客户端兼容问题
• 确保Kubernetes集群所有节点都有权限读写对应Azure Blob Storage容器，无需通过Databricks挂载路径做中转

具体配置步骤 #

第一步：获取Databricks Hive Metastore连接信息 #

你需要在Databricks Notebook中运行以下命令获取必要的配置项：

set javax.jdo.option.ConnectionURL;
set javax.jdo.option.ConnectionDriverName;
set javax.jdo.option.ConnectionUserName;
set javax.jdo.option.ConnectionPassword;
set hive.metastore.uris;

运行后会输出metastore的连接地址、驱动、账号密码等核心参数，全部复制保存。

第二步：配置外部PySpark会话的Hive参数 #

在Kubernetes集群提交PySpark作业时，需要在spark配置中加入以下参数：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ConnectDatabricksHive") \
    # 替换为上一步获取的对应Hive Metastore配置值
    .config("hive.metastore.uris", "thrift://<你的Databricks-metastore地址>:9083")
    .config("javax.jdo.option.ConnectionURL", "<你获取的ConnectionURL值>")
    .config("javax.jdo.option.ConnectionDriverName", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
    .config("javax.jdo.option.ConnectionUserName", "<你获取的用户名>")
    .config("javax.jdo.option.ConnectionPassword", "<你获取的密码>")
    # 开启Hive支持
    .enableHiveSupport()
    # Delta兼容配置
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    # 路径映射配置：把Databricks的挂载路径映射到原生wasb路径，解决路径不一致问题
    .config("spark.sql.warehouse.dir", "wasbs://container@storage.blob.core.windows.net/")
    .config("spark.databricks.mountPoint.mappings", "/mnt/db=wasbs://container@storage.blob.core.windows.net")
    .getOrCreate()

路径映射配置是解决Databricks挂载路径无法被外部集群识别的核心步骤，配置完成后你调用saveAsTable("db.table_name")时，数据会自动写入你预期的Blob存储路径，同时元数据同步写入Databricks的Hive Metastore。
如果你使用的Spark版本不支持spark.databricks.mountPoint.mappings参数，也可以直接在Databricks侧重新创建数据库，指定原生wasb路径作为LOCATION，避免路径转换问题。

第三步：验证配置是否生效 #

配置完成后运行以下测试代码验证：

# 查看数据库列表，应该能看到你在Databricks侧创建的db数据库
print(spark.catalog.listDatabases())

# 写入测试表
df = spark.createDataFrame([(1, "test"), (2, "demo")], ["id", "name"])
df.write.mode('overwrite').format("delta").saveAsTable("db.test_table")

# 验证表是否能正常查询
spark.sql("SELECT * FROM db.test_table").show()

运行完成后回到Databricks环境，执行SELECT * FROM db.test_table也能查询到刚写入的数据，说明配置成功。

注意事项 #

• 所有依赖的Jar包（包括SQL Server JDBC驱动、对应版本的Delta Jar包、Hive metastore客户端Jar包）需要提前放到Kubernetes Spark集群的driver和executor的classpath下
• 如果出现元数据写入权限问题，需要确认你获取的metastore账号密码有写入权限
• 不要在外部集群修改Databricks系统内置库的元数据，避免影响Databricks集群正常运行

内容的提问来源于stack exchange，提问作者Roman