使用 Spark 访问 Workday 数据

借助 Managed Service for Apache Spark，您可以直接从 Spark 工作负载查询和分析外部商家数据。您可以使用 Workday Live Data Query JDBC 驱动程序连接到 Workday 租户，将实时 Workday 数据集加载到 Spark DataFrame 中，并将这些数据集成到 Google Cloud上的分析流水线中。

在本指南中，您将配置 Workday Live Data Query，将私钥安全地存储在 Secret Manager 中，并使用 Google Cloud Data Agent Kit 扩展程序在 Visual Studio Code 或 Cloud Shell 编辑器中针对 Managed Service for Apache Spark 运行 PySpark 笔记本。

费用

Data Agent Kit 扩展程序无需额外付费。不过，本指南使用 Google Cloud的以下收费组件：

如需根据您的预计使用量来估算费用，请使用价格计算器。

准备工作

  1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  2. Verify that you have the permissions required to complete this guide.

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Dataplex, Cloud Storage, and Secret Manager APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

所需的 Identity and Access Management (IAM) 角色

您需要拥有特定 IAM 角色才能运行本页面上的示例。这些角色可能已获授予，具体取决于组织政策。如需检查角色授予情况，请参阅您是否需要授予角色？。如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您需要的角色取决于工作负载的运行方式。 Managed Service for Apache Spark 无服务器 2.3 及更早版本的无服务器运行时会以服务账号身份运行工作负载。默认情况下，Managed Service for Apache Spark 无服务器运行时 3.0 及更高版本会使用您的最终用户凭据 (EUC) 运行工作负载，因此您需要向用户账号（而非服务账号）授予数据访问角色。如果您为 3.0 或更高版本的运行时指定自定义服务账号，则工作负载会使用服务账号凭证运行。如需了解详情，请参阅无服务器部署角色。

用户角色

如需获得创建 Managed Service for Apache Spark 交互式笔记本会话和访问资源所需的权限，请让您的管理员为您授予以下 IAM 角色：

  • 使用无服务器运行时 2.3 或更早版本：
  • 使用无服务器运行时 3.0 或更高版本：项目的 Dataproc Serverless Editor (roles/dataproc.serverlessEditor)
  • 使用最终用户凭证从 Cloud Storage 读取 JDBC 驱动程序：针对包含驱动程序的存储桶的 Storage Object Viewer (roles/storage.objectViewer) 角色
  • 使用最终用户凭据从 Secret Manager 读取私钥 Secret：Secret 或项目上的 Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)

服务账号角色

向运行工作负载的服务账号授予以下角色。 仅当工作负载使用服务账号凭证（而非默认最终用户凭证 [EUC]）运行时，Managed Service for Apache Spark 无服务器运行时 3.0 及更高版本才需要这些角色。

为确保 Compute Engine 默认服务账号或自定义服务账号具有运行 Managed Service for Apache Spark 工作负载、读取驱动程序和访问 Secret 所需的权限，请让您的管理员为 Compute Engine 默认服务账号或自定义服务账号授予以下 IAM 角色：

  • 使用无服务器运行时 2.3 或更早版本：针对项目的 Dataproc Worker (roles/dataproc.worker)
  • 使用无服务器运行时 3.0 或更高版本，并使用服务账号凭证，而不是默认最终用户凭证 (EUC)：
  • 从 Cloud Storage 读取 JDBC 驱动程序：针对包含驱动程序的存储桶的 Storage Object Viewer (roles/storage.objectViewer) 角色
  • 从 Secret Manager 读取私钥 Secret：Secret 或项目上的 Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)

Workday 前提条件

在使用 Live Data Query 连接到 Workday 之前，请完成以下设置：

  1. 从 Workday 下载 Workday Live Data Query JDBC 驱动程序。

  2. 将下载的 JDBC 驱动程序 JAR 文件上传到项目中的 Cloud Storage 存储桶。

  3. 按照 Workday Live Data Query 文档中的说明，完成在 Workday 租户中设置 Live Data Query 的配置步骤。

    在 Workday 中进行配置期间，请创建一个集成系统用户 (ISU)，注册一个 API 客户端，并生成一个私钥对。

  4. 请注意以下 Workday 配置值：

    • Workday 主机：您的 Workday 租户主机，例如 services1.workday.com。

    • 租户名称：您的 Workday 租户名称。

    • 客户端 ID：Workday 租户中已注册的 API 客户端 ID。

    • ISU 用户名：集成系统用户的用户名。

将私钥存储在 Secret Manager 中

将 Workday Live Data Query 配置期间生成的私钥文件安全地存储在 Secret Manager 中：

  1. 在 Cloud Shell 中，运行以下命令以生成 x509 密钥对：

    openssl req -x509 -newkey rsa:2048 -keyout private-key.pem -out public-key.cert -sha256 -days 365 -nodes

  2. 如需查看私钥，请运行以下命令：

    cat private-key.pem

  3. 将私钥存储在 Secret Manager 中：

    1. 在 Google Cloud 控制台中，前往 Secret Manager 页面：

      前往 Secret Manager

    2. 点击创建 Secret。

    3. 在名称字段中，输入 Secret 的名称，例如 workday-private-key。

    4. 在Secret 值下，上传 private-key.pem 或粘贴私钥内容，包括 -----BEGIN PRIVATE KEY----- 和 -----END PRIVATE KEY----- 标头和页脚行。

    5. 点击创建 Secret。

  4. 如需查看公钥，请运行以下命令：

    cat public-key.cert

  5. 在 Workday 中，搜索 Create x509 Public Key（创建 x509 公钥），粘贴 public-key.cert 中的公钥，然后点击 OK（确定）。

  6. 在之前创建的 Workday API 客户端中，将 x509 证书更新为新的 x509 密钥。

在 IDE 中设置 Data Agent Kit

您可以通过在 Visual Studio Code 或其他受支持的 IDE 中安装扩展程序来设置 Data Agent Kit，也可以使用 Cloud Shell Editor 中预安装的扩展程序。

如需在 IDE 或 Cloud Shell Editor 中配置 Data Agent Kit，请完成以下步骤：

  1. 设置并配置 Data Agent Kit：

    1. 安装 Data Agent Kit 扩展程序。

    2. 配置 Data Agent Kit。

创建笔记本

使用 Data Agent Kit 创建交互式 PySpark 笔记本：

  1. 在 IDE 或 Cloud Shell Editor 中，打开 Data Agent Kit 面板。

  2. 在 Apache Spark 下，选择新建笔记本，然后选择本地内核。

  3. 从可用运行时（也称为会话模板）列表中选择一个运行时。

  4. 选择从新笔记本开始。

    系统会打开一个新笔记本，其中初始单元格已填充 Spark Connect 会话初始化代码。

  5. 在笔记本编辑器顶部，点击选择内核，然后为笔记本选择本地 Python 内核。

在 PySpark 中查询 Workday 数据

  1. 在第一个笔记本单元中，安装所需的客户端库并设置环境变量：

    %pip install dataproc-spark-connect google-cloud-dataproc google-cloud-secret-manager google-cloud-storage

#@title Set up all environment variables

# Replace with your Project ID
%env PROJECT_ID=PROJECT_ID

# Replace with the name of the Secret Manager global secret in the project
%env SECRET_NAME=workday-private-key
# Replace with the version of the secret to use
%env SECRET_VERSION=latest

# Replace with the Cloud Storage bucket where you have the Workday Data Cloud JDBC Driver file stored
%env JDBC_DRIVER_GCS_BUCKET=BUCKET_NAME
# Replace with the name of the Workday Data Cloud JDBC Driver file in the Cloud Storage bucket
%env JDBC_DRIVER_FILE_NAME=ldq-jdbc-driver-1.1.0-all.jar
# Replace with the local directory of the notebook to copy the JDBC Driver File to
%env JDBC_DRIVER_LOCAL_DIRECTORY=/tmp
# Replace with the Driver Class for the Workday Data Cloud JDBC Driver
%env WORKDAY_DRIVER_CLASS=com.workday.ldq.jdbc.Driver

# ------------------------------------------------------------------------------

# Replace with the Workday URL of your tenant
%env WORKDAY_HOST=WORKDAY_HOST
# Replace with your Workday tenant name
%env TENANT_NAME=TENANT_NAME
# Replace with the Client ID of the "API Client"
%env CLIENT_ID=CLIENT_ID
# Replace with the Integration System User username
%env ISU_USERNAME=ISU_USERNAME

  2. 在新的单元格中，向 Compute Engine 默认服务账号授予 Secret Manager Secret Accessor 角色 (roles/secretmanager.secretAccessor)：

    %%bash
COMPUTE_ENGINE_DEFAULT_SERVICE_ACCOUNT=$(gcloud iam service-accounts list --project="${PROJECT_ID}" \
  --filter="email ~ -compute@" \
  --format="value(email)")

echo "COMPUTE_ENGINE_DEFAULT_SERVICE_ACCOUNT: ${COMPUTE_ENGINE_DEFAULT_SERVICE_ACCOUNT}"

gcloud secrets add-iam-policy-binding ${SECRET_NAME} \
  --project="${PROJECT_ID}" \
  --member="serviceAccount:${COMPUTE_ENGINE_DEFAULT_SERVICE_ACCOUNT}" \
  --role="roles/secretmanager.secretAccessor"

  3. 在新的单元中，将 JDBC 驱动程序从 Cloud Storage 复制到本地目录：

    gsutil cp gs://$JDBC_DRIVER_GCS_BUCKET/$JDBC_DRIVER_FILE_NAME $JDBC_DRIVER_LOCAL_DIRECTORY

  4. 在新的单元中，安装 Secret Manager 客户端库：

    %%bash
pip install google-cloud-secret-manager

  5. 在新的单元格中，验证驱动程序 JAR 文件是否已下载到本地：

    !gsutil cp gs://$JDBC_DRIVER_GCS_BUCKET/$JDBC_DRIVER_FILE_NAME /tmp/
!ls -lh /tmp/ldq-jdbc-driver-1.1.0-all.jar

  6. 在新的单元中，将 Workday 驱动程序 JAR 注入 PySpark jars 目录：

    %%bash

echo "Downloading Workday JAR from Cloud Storage..."
# Downloads your specific 1.1.0 driver JAR into the notebook's /tmp folder
gcloud storage cp gs://$JDBC_DRIVER_GCS_BUCKET/$JDBC_DRIVER_FILE_NAME /tmp/

echo "Dynamically finding the PySpark jars directory..."
PYSPARK_JARS=$(python -c "import os, pyspark; print(os.path.join(os.path.dirname(pyspark.__file__), 'jars'))")
echo "Found PySpark jars at: $PYSPARK_JARS"

echo "Injecting Workday driver into PySpark core..."
cp /tmp/$JDBC_DRIVER_FILE_NAME "$PYSPARK_JARS/"

echo "Success! The driver is now a native part of PySpark."

  7. 在新的单元中，配置 Spark 环境并从 Secret Manager 中检索 x509 私钥：

    #@title Configure the Spark environment and retrieve the x509 private key from Secret Manager

import os
import pyspark
from pyspark.sql import SparkSession
from google.cloud import secretmanager

PROJECT_ID = os.environ["PROJECT_ID"]
SECRET_NAME = os.environ["SECRET_NAME"]
SECRET_VERSION = os.environ["SECRET_VERSION"]
JDBC_DRIVER_LOCAL_DIRECTORY = os.environ["JDBC_DRIVER_LOCAL_DIRECTORY"]
JDBC_DRIVER_FILE_NAME = os.environ["JDBC_DRIVER_FILE_NAME"]
LOCAL_JAR_PATH = JDBC_DRIVER_LOCAL_DIRECTORY + "/" + JDBC_DRIVER_FILE_NAME
WORKDAY_HOST = os.environ["WORKDAY_HOST"]
TENANT_NAME = os.environ["TENANT_NAME"]
CLIENT_ID = os.environ["CLIENT_ID"]
ISU_USERNAME = os.environ["ISU_USERNAME"]
WORKDAY_DRIVER_CLASS = os.environ["WORKDAY_DRIVER_CLASS"]

jdbc_url = f"jdbc:workday://{WORKDAY_HOST}:443/system/information_schema?SSLVerification=CA"

def access_private_key_secret():
    print(f"Fetching secret contents from Secret Manager: {SECRET_NAME}...")
    client = secretmanager.SecretManagerServiceClient()
    secret_name = f"projects/{PROJECT_ID}/secrets/{SECRET_NAME}/versions/{SECRET_VERSION}"
    response = client.access_secret_version(request={"name": secret_name})
    print(f"Fetched secret contents from Secret Manager: {SECRET_NAME}")
    return response.payload.data.decode("UTF-8")

private_key_content = access_private_key_secret()

  8. 在新的单元中，使用类路径配置初始化 Spark 会话：

    # Stop the old cached Spark session so your newly downloaded JAR is not ignored!
try:
    spark.stop()
    print("Stopped existing Spark session.")
except Exception:
    pass

print("Initializing fresh Spark session with Classpath configuration...")
print(f"Loading JAR from: {LOCAL_JAR_PATH}")

# Rebuild Spark session with explicit Driver & Executor classpaths
spark = SparkSession.builder \
    .appName("WorkdayJDBC Connection Example") \
    .config("spark.jars", LOCAL_JAR_PATH) \
    .config("spark.driver.extraClassPath", LOCAL_JAR_PATH) \
    .config("spark.executor.extraClassPath", LOCAL_JAR_PATH) \
    .getOrCreate()
print("Spark Session successfully booted with JAR!")

  9. 在最后一个单元中，启动连接并查询 Workday：

    import os
import subprocess
import zipfile
from pyspark.sql import SparkSession

LOCAL_JAR_PATH = "/tmp/ldq-jdbc-driver-1.1.0-all.jar"

# Load bucket and file name from your Cell 1 environment variables
GCS_BUCKET = os.environ.get("JDBC_DRIVER_GCS_BUCKET", "BUCKET_NAME")
GCS_FILE = os.environ.get("JDBC_DRIVER_FILE_NAME", "ldq-jdbc-driver-1.1.0-all.jar")
GCS_URI = f"gs://{GCS_BUCKET}/{GCS_FILE}"

# 1. Ensure the JAR file is in the notebook's /tmp directory
if not os.path.exists(LOCAL_JAR_PATH):
    print(f"Downloading {GCS_URI} to {LOCAL_JAR_PATH}...")
    subprocess.run(["gsutil", "cp", GCS_URI, LOCAL_JAR_PATH], check=True)
print(f"JAR verified on disk: {LOCAL_JAR_PATH} ({os.path.getsize(LOCAL_JAR_PATH):,} bytes)")

# 2. Read the real JDBC driver class name (skipping any '#' comment lines!)
with zipfile.ZipFile(LOCAL_JAR_PATH, "r") as z:
    raw_lines = (
        z.read("META-INF/services/java.sql.Driver")
        .decode("utf-8")
        .splitlines()
    )
    # Filter out empty lines and '#' comment lines
    valid_classes = [
        line.strip()
        for line in raw_lines
        if line.strip() and not line.strip().startswith("#")
    ]
    driver_class = valid_classes[0]

print(f"Real Driver Class inside JAR: {driver_class}")

# 3. Get or create the Spark session
spark = SparkSession.builder.appName("WorkdayJDBC Connection Example").getOrCreate()

# 4. Inject the JAR directly into the running Java URLClassLoader (and Scala SparkContext)
jvm = spark.sparkContext._jvm
jar_url = jvm.java.io.File(LOCAL_JAR_PATH).toURI().toURL()

# Add to both the Spark classloader and the current thread's context classloader
jvm.org.apache.spark.util.Utils.getContextOrSparkClassLoader().addURL(jar_url)
jvm.java.lang.Thread.currentThread().getContextClassLoader().addURL(jar_url)
spark.sparkContext._jsc.sc().addJar(LOCAL_JAR_PATH)

# Register the driver class with Spark's JDBC DriverRegistry
jvm.org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry.register(driver_class)
print(f"Successfully registered {driver_class} in Spark's live JVM!")

# 5. Query Workday
print("Querying Workday...")
df_data = (
    spark.read.format("jdbc")
    .option("driver", driver_class)
    .option("url", jdbc_url)
    .option("wd.authn.clientId", CLIENT_ID)
    .option("wd.authn.isu", ISU_USERNAME)
    .option(
        "wd.authn.accessTokenEndpoint",
        f"https://{WORKDAY_HOST}/ccx/oauth2/{TENANT_NAME}/token",
    )
    .option("wd.authn.privateKey", private_key_content)
    .option("query", "SELECT * FROM system.jdbc.catalogs")
    .load()
)

print("Connection Successful! Data Retrieved:")
df_data.show(50, truncate=False)

    替换以下内容：

    • PROJECT_ID：您的 Google Cloud 项目 ID。
    • LOCATION：Managed Service for Apache Spark 会话模板的 Google Cloud 区域，例如 us-central1。
    • RUNTIME_ID：Managed Service for Apache Spark 会话模板的 ID。
    • BUCKET_NAME：包含 Workday JDBC 驱动程序的 Cloud Storage 存储桶的名称。
    • PATH/workday-live-data-query-jdbc-driver.jar：Cloud Storage 存储桶中 Workday JDBC 驱动程序 JAR 的对象路径。
    • WORKDAY_HOST：您的 Workday 租户主机，例如 services1.workday.com。
    • TENANT_NAME：您的 Workday 租户名称。
    • CLIENT_ID：Workday 租户中已注册的 API 客户端 ID。
    • ISU_USERNAME：Workday 租户中的集成系统用户用户名。

  10. 点击单元旁边的执行单元（播放图标），或按 Shift+Enter。

    Managed Service for Apache Spark 会执行查询，并在单元格下方显示 Workday DataFrame 输出。

清理

为避免因本指南中使用的资源导致您的 Google Cloud 账号产生费用，请终止活跃的交互式会话或删除已创建的资源。

终止交互式会话

  1. 在 IDE 或 Cloud Shell 编辑器中，当工作负载完成时，终止活跃的笔记本会话。

  2. 可选：如果您不再需要 Cloud Storage 存储桶或 JDBC 驱动程序 JAR 文件，请将其删除。

  3. 可选：如果不再需要该 Secret，请将其从 Secret Manager 中删除：

    gcloud secrets delete workday-private-key

后续步骤