Spark3.5连接Google Cloud Spanner报错及JDBC连接方案咨询

环境信息 #

• Spark版本：3.5.0
• Scala版本：2.12
• 使用驱动：spark-3.1-spanner-1.0.0.jar

连接代码 #

import os
from google.oauth2 import service_account
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]") \
.config("spark.jars","/home/arvind/Downloads/spark-3.1-spanner-1.0.0.jar")\
.appName("cloud spanner Testing").getOrCreate()
print(spark)
credentialFilePath = "/home/arvind/Downloads/jsonKeysFiles/citric-sol-436208-r8-099f62042467.json"
spark.conf.set("google.cloud.auth.service.account.json.keyfile", credentialFilePath)

# credentials = service_account.Credentials.from_service_account_file(credentialFilePath)
# print(credentials)
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credentialFilePath

try:
   df = spark.read.format('cloud-spanner') \
      .option("projectId", "citric-sol-436208-r8") \
      .option("instanceId", "cloudspanner2") \
      .option("databaseId", "testcloudspanner") \
      .option("table", "moonatable1") \
      .load()
   df.show()

except Exception as error:
   print("Error ==",error)

报错信息 #

An error occurred while calling o39.load.
: com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.SpannerException: INVALID_ARGUMENT: Invalid credentials path specified: There are no credentials set in the connection string, and the default application credentials are not set or are pointing to an invalid or non-existing file.
Please check the GOOGLE_APPLICATION_CREDENTIALS environment variable and/or the credentials that have been set using the Google Cloud SDK gcloud auth application-default login command
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.SpannerExceptionFactory.newSpannerExceptionPreformatted(SpannerExceptionFactory.java:291)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.SpannerExceptionFactory.newSpannerExceptionPreformatted(SpannerExceptionFactory.java:297)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.SpannerExceptionFactory.newSpannerException(SpannerExceptionFactory.java:61)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.CredentialsService.createCredentials(CredentialsService.java:71)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.ConnectionOptions.(ConnectionOptions.java:748)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.ConnectionOptions.(ConnectionOptions.java:85)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.ConnectionOptions$Builder.build(ConnectionOptions.java:617)
at com.google.cloud.spark.spanner.SpannerUtils.connectionFromProperties(SpannerUtils.java:180)
at com.google.cloud.spark.spanner.SpannerTable.(SpannerTable.java:62)
at com.google.cloud.spark.spanner.Spark31SpannerTableProvider.inferSchema(Spark31SpannerTableProvider.java:39)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.getTableFromProvider(DataSourceV2Utils.scala:90)
at org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils$.loadV2Source(DataSourceV2Utils.scala:140)
at org.apache.spark.sql.DataFrameReader.$anonfun$load$1(DataFrameReader.scala:210)
at scala.Option.flatMap(Option.scala:271)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:208)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:172)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.io.IOException: The Application Default Credentials are not available. They are available if running in Google Compute Engine. Otherwise, the environment variable GOOGLE_APPLICATION_CREDENTIALS must be defined pointing to a file defining the credentials. See https://developers.google.com/accounts/docs/application-default-credentials for more information.
at com.google.cloud.spark.spanner.repackaged.com.google.auth.oauth2.DefaultCredentialsProvider.getDefaultCredentials(DefaultCredentialsProvider.java:127)
at com.google.cloud.spark.spanner.repackaged.com.google.auth.oauth2.GoogleCredentials.getApplicationDefault(GoogleCredentials.java:129)
at com.google.cloud.spark.spanner.repackaged.com.google.auth.oauth2.GoogleCredentials.getApplicationDefault(GoogleCredentials.java:101)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.CredentialsService.internalGetApplicationDefault(CredentialsService.java:98)
at com.google.cloud.spark.spanner.repackaged.com.google.cloud.spanner.connection.CredentialsService.createCredentials(CredentialsService.java:55)
... 24 more

疑问 #

1. 确认当前使用的spark-3.1-spanner-1.0.0.jar驱动是否适配Spark 3.5.0 + Scala 2.12环境
2. 如何通过JDBC方式实现PySpark、Databricks与Google Cloud Spanner的连接

解答 #

一、驱动版本适配问题 #

spark-3.1-spanner-1.0.0.jar是针对Spark 3.1.x版本开发的连接器，和你当前使用的Spark 3.5.0存在版本不兼容问题，这是导致凭证加载异常的核心原因之一。你需要更换为对应Spark 3.5.x版本的连接器，比如spark-3.5-spanner-2.18.0（具体版本以官方兼容矩阵为准），且选择Scala 2.12对应的包。

同时，当前代码中设置GOOGLE_APPLICATION_CREDENTIALS的时机错误——SparkSession创建后再设置环境变量，JVM进程无法感知该变更。正确配置方式如下：

import os
from pyspark.sql import SparkSession

# 先设置环境变量，再创建SparkSession
credentialFilePath = "/home/arvind/Downloads/jsonKeysFiles/citric-sol-436208-r8-099f62042467.json"
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = credentialFilePath

spark = SparkSession.builder.master("local[*]") \
    .config("spark.jars","/path/to/spark-3.5-spanner-xxx.jar") \
    .appName("cloud spanner Testing").getOrCreate()

# 也可直接通过Spark配置传递凭证，无需依赖环境变量
spark.conf.set("google.cloud.auth.service.account.json.keyfile", credentialFilePath)

try:
   df = spark.read.format('cloud-spanner') \
      .option("projectId", "citric-sol-436208-r8") \
      .option("instanceId", "cloudspanner2") \
      .option("databaseId", "testcloudspanner") \
      .option("table", "moonatable1") \
      .load()
   df.show()
except Exception as error:
   print("Error ==",error)

二、JDBC连接方案 #

1. PySpark本地环境连接

首先下载Cloud Spanner JDBC驱动（如cloud-spanner-jdbc-2.18.0.jar），然后使用以下代码：

from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[*]") \
    .config("spark.jars","/path/to/cloud-spanner-jdbc-2.18.0.jar") \
    .appName("Spanner JDBC Test").getOrCreate()

# JDBC URL格式：jdbc:cloudspanner:/projects/{projectId}/instances/{instanceId}/databases/{databaseId}
jdbc_url = "jdbc:cloudspanner:/projects/citric-sol-436208-r8/instances/cloudspanner2/databases/testcloudspanner"
properties = {
    "driver": "com.google.cloud.spanner.jdbc.JdbcDriver",
    "credentialsFile": "/home/arvind/Downloads/jsonKeysFiles/citric-sol-436208-r8-099f62042467.json"
}

# 读取数据
df = spark.read.jdbc(url=jdbc_url, table="moonatable1", properties=properties)
df.show()

# 写入数据示例（可选）
# df.write.jdbc(url=jdbc_url, table="moonatable1", mode="append", properties=properties)

2. Databricks环境连接

先通过Maven坐标安装JDBC驱动（坐标：com.google.cloud:cloud-spanner-jdbc:2.18.0，版本按需调整），然后在Notebook中使用以下代码：

jdbc_url = "jdbc:cloudspanner:/projects/citric-sol-436208-r8/instances/cloudspanner2/databases/testcloudspanner"
properties = {
    "driver": "com.google.cloud.spanner.jdbc.JdbcDriver",
    # 使用Databricks Secret管理凭证，避免硬编码
    "credentialsFile": "/dbfs/FileStore/secrets/your-service-account-key.json"
}

# 读取数据
df = spark.read.jdbc(url=jdbc_url, table="moonatable1", properties=properties)
df.display()

更安全的凭证管理方式（通过Databricks Secrets）：

# 从Secrets中获取凭证内容并写入临时文件
credential_content = dbutils.secrets.get(scope="your-secret-scope", key="service-account-key")
temp_cred_path = "/tmp/service-account-key.json"
dbutils.fs.put(temp_cred_path, credential_content, overwrite=True)

jdbc_url = "jdbc:cloudspanner:/projects/citric-sol-436208-r8/instances/cloudspanner2/databases/testcloudspanner"
properties = {
    "driver": "com.google.cloud.spanner.jdbc.JdbcDriver",
    "credentialsFile": "/dbfs" + temp_cred_path
}

df = spark.read.jdbc(url=jdbc_url, table="moonatable1", properties=properties)
df.display()

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