安排自定义 Spark 和 Spark SQL 任务

Dataplex 支持按计划执行自定义代码,可以一次性运行,也可以按常规时间表运行,还可以按需运行。按需模式现已启用 预览版,只能通过 API。您可以使用 Spark (Java)、 PySpark(仅限于 Spark 3.2 版)或 Spark SQL。Dataplex 使用无服务器 Spark 处理和内置无服务器调度程序来执行代码。

术语

任务
Dataplex 任务代表您需要 Dataplex 按时间表执行任务。它封装了您的代码、参数和时间表。
作业

一个作业表示 Dataplex 任务的单次运行。对于 例如,如果某个任务安排每天运行,Dataplex 将 每天创建一份工作。

对于 2023 年 5 月 10 日当天或之后创建的作业,触发器字段显示 作业的执行触发器类型。

作业执行触发器类型如下:

  • RUN_REQUEST:表示作业因调用 RunTask API。

  • TASK_CONFIG:表示作业因 TriggerSpec 任务配置。

安排模式

Dataplex 支持以下调度模式:

运行一次
使用此模式可仅运行一次任务。您可以选择立即运行,也可以选择在未来的某个时间运行。如果运行该任务 但最长可能需要两分钟才会开始执行。
按计划运行
使用此模式可按重复频率运行任务。支持的重复时间包括每天、每周、每月或自定义。
按需运行

使用此模式可以按需运行先前创建的任务。只有 RunTask API 支持按需运行模式。当作业按需运行时,Dataplex 会使用现有参数创建作业。您可以指定 ExecutionSpec 参数和标签来执行作业。

准备工作

  1. 启用 Dataproc API。

    启用 Dataproc API

  2. 为您的网络和子网启用专用 Google 访问通道。在网络上启用专用 Google 访问通道 用于 Dataplex 任务的应用场景。如果您在创建 Dataplex 任务时未指定网络或子网,则 Dataplex 会使用默认子网,并且您必须为默认子网启用专用 Google 访问通道。

  3. 创建服务账号。必须使用服务账号才能安排任何 Dataplex 任务。该服务账号必须属于您在其中执行任务的项目。 该服务账号必须具有以下权限:

    • 访问正在处理的 BigQuery 和 Cloud Storage 数据。

    • Dataproc Worker 角色 权限。

    • 如果任务需要读取或更新 Dataproc Metastore 实例附加到数据湖,则服务账号需要 Dataproc Metastore Viewer 或 Editor 角色。 必须在 Dataplex 所在的项目中授予此角色 数据湖现已设置完毕

    • 如果任务是 Spark SQL 作业,您需要向服务账号授予 Dataplex 开发者角色。必须在设置 Dataplex 数据湖的项目中授予此角色。

    • 如果任务是 Spark SQL 作业,您需要对写入结果的存储桶拥有 Cloud Storage 管理员权限。

    • 如需安排和运行 Spark SQL 和自定义 Spark 任务,您必须获得服务账号的 Dataplex Metadata Reader (roles/dataplex.metadataReader)、Dataplex Viewer (roles/dataplex.viewer) 和 Dataproc Metastore Metadata User (roles/metastore.metadataUser) IAM 角色。

  4. 向提交作业的用户授予 Service Account User 角色 (roles/iam.serviceAccountUser) 服务账号有关说明,请参阅管理对服务账号的访问权限

  5. 向 Dataplex Lake 服务账号授予使用 服务账号。您可以在 Google Cloud 控制台数据湖详情页面中找到 Dataplex 数据湖服务账号。

  6. 如果包含 Dataplex 数据湖的项目不同于要执行任务的项目,请在执行任务的项目中向 Dataplex 数据湖服务账号授予 Dataproc 编辑者角色

  7. 将所需的代码工件(JAR、Python 或 SQL 脚本文件)或归档文件(.jar.tar.tar.gz.tgz.zip)放置在 Cloud Storage 路径中。

  8. 确保服务账号具有所需的 storage.objects.get 对存储这些代码的 Cloud Storage 存储桶的权限 工件

安排 Spark(Java 或 Python)任务

控制台

  1. 在 Google Cloud 控制台中,前往 Dataplex 页面:

    进入 Dataplex

  2. 转到进程视图。

  3. 点击创建任务

  4. 对于创建自定义 Spark 任务,请点击创建任务

  5. 选择一个 Dataplex 数据湖。

  6. 提供任务名称。

  7. 为任务创建 ID

  8. 任务配置部分,针对类型,选择 SparkPySpark

  9. 输入相关参数。

  10. 服务账号字段中,输入您的自定义 Spark 任务可通过哪个用户服务账号执行。

  11. 点击继续

  12. 可选:设置时间表:选择运行一次重复。填写必填字段。

  13. 点击继续

  14. 可选:自定义资源添加其他设置

  15. 点击创建

gcloud

您可以使用 gcloud CLI 命令安排 Spark(Java/Python)任务。下表列出了 使用:

参数 说明
--lake Dataplex 服务的数据湖资源的数据湖 ID。
--location Dataplex 服务的位置。
--spark-main-class 驱动程序的主类。包含该类的 jar 文件必须位于默认的 CLASSPATH 中。
--spark-main-jar-file-uri jar 文件的 Cloud Storage URI,该文件包含 主类。
--spark-archive-uris 可选:要提取到的归档的 Cloud Storage URI 每个执行器的工作目录。支持的文件类型:.jar.tar.tar.gz.tgz.zip
--spark-file-uris 可选:要放入各个执行程序的工作目录中的文件的 Cloud Storage URI。
--batch-executors-count 可选:作业执行程序的总数。默认值为 2。
--batch-max-executors-count 可选:可配置的执行器的数量上限。默认 值为 1000。如果 batch-max-executors-count 大于 batch-executors-count,则 Dataplex 会启用自动扩缩。
--container-image-java-jars 可选:要添加到类路径的 Java JAR 列表。有效输入包括指向 Jar 二进制文件的 Cloud Storage URI。
例如 gs://bucket-name/my/path/to/file.jar
--container-image-properties 可选:在 prefix:property 中指定的属性键 格式。
例如 core:hadoop.tmp.dir
如需了解详情,请参阅集群属性
--vpc-network-tags 可选:要应用于作业的网络标记列表。
--vpc-network-name 可选:运行作业的 Virtual Private Cloud 网络。默认情况下,Dataplex 会使用项目中名为 Default 的 VPC 网络。
您必须仅使用 --vpc-network-name--vpc-sub-network-name 中的一个。
--vpc-sub-network-name 可选:作业运行所在的 VPC 子网。
您只能使用下列其中一项:--vpc-sub-network-name--vpc-network-name
--trigger-type 用户指定任务的触发器类型。值必须为以下项之一:
ON_DEMAND - 任务在任务结束后不久运行一次 创建过程。
RECURRING - 任务会按计划定期运行。
--trigger-start-time 可选:首次运行任务的时间。格式 为 `{year}-{month}-{day}T{hour}:{min}:{sec}Z`,其中时区为 世界协调时间 (UTC)例如,"2017-01-15T01:30:00Z"采用世界协调时间 (UTC) 的 01:30 编码, 2017 年 1 月 15 日。如果未指定此值,则如果触发器类型为 ON_DEMAND,任务将在提交后运行;如果触发器类型为 RECURRING,则将按照指定的时间表运行。
--trigger-disabled 可选:阻止任务执行。此参数不会取消已在运行的任务,而是会暂时停用 RECURRING 任务。
--trigger-max-retires 可选:取消之前的重试尝试次数。设置值 零表示永不重试失败的任务。
--trigger-schedule Cron 时间表: 定期运行任务。
--description 可选:任务说明。
--display-name 可选:任务的显示名称。
--labels 可选:要添加的标签 KEY=VALUE 对的列表。
--execution-args 可选:要传递给任务的参数。参数可以是 键值对。您可以将以英文逗号分隔的键值对列表作为参数 执行参数。要传递位置参数,请将键设置为 TASK_ARGS),并将值设置为 所有位置参数。如需使用除英文逗号以外的分隔符,请参阅转义
如果传递 key-value 和位置参数 一起,则 TASK_ARGS 将作为最后一个参数传递。
--execution-service-account 用于执行任务的服务账号。
--max-job-execution-lifetime 可选:作业执行到期前的时长上限。
--container-image 可选:用于作业运行时环境的自定义容器映像。如果未指定,则系统将使用默认容器映像。
--kms-key 可选:用于加密的 Cloud KMS 密钥,格式为:
projects/{project_number}/locations/{location_id}/keyRings/{key-ring-name}/cryptoKeys/{key-name}

Java 示例:

glcoud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --trigger-type=ON_DEMAND spark-main-jar-file-uri=<gcs location to java file> --execution-service-account=<service-account-email> --trigger-start-time=<timestamp after which job starts ex. 2099-01-01T00:00:00Z> --labels=key1=value1,key2=value3,key3=value3 --execution-args=arg1=value1,arg2=value3,arg3=value3 <task-id>

PySpark 示例:

gcloud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --trigger-type=RECURRING --trigger-schedule=<Cron schedule https://en.wikipedia.org/wiki/Cron> --spark-python-script-file=<gcs location to python script> --execution-service-account=<service-account-email> --execution-args=^::^arg1=value1::arg2=value2::TASK_ARGS="pos-arg1, pos-arg2" <task-id>

REST

如需创建任务,请按以下步骤操作: 请使用 APIs Explorer。

安排 Spark SQL 任务

gcloud

如需安排 Spark SQL 任务,请运行与安排 Spark(Java 或 Python)任务中相同的 gcloud CLI 命令,并添加以下额外参数:

参数 说明
--spark-sql-script SQL 查询文本。spark-sql-script 或 必须提供spark-sql-script-file
--spark-sql-script-file 对查询文件的引用。此值可以是查询文件的 Cloud Storage URI,也可以是 SQL 脚本内容的路径。 spark-sql-script 或 必须提供spark-sql-script-file
--execution-args 对于 Spark SQL 任务,以下参数是必需的,并且需要作为位置参数传递:
--output_location, <GCS uri of the output directory>
--output_format, <output file format>
支持的格式有 CSV 文件、JSON 文件、parquet 和 orc。
gcloud dataplex tasks create --project=<project-name> --location=<location> --lake=<lake-id> --execution-service-account=<service-account-email> --trigger-type=ON_DEMAND --spark-sql-script=<sql-script> --execution-args=^::^TASK_ARGS="--output_location, <gcs folder location>, --output_format, json" <sql-task-id>

REST

如需创建任务,请使用 API Explorer。

监控任务

控制台

  1. 在 Google Cloud 控制台中,前往 Dataplex 页面:

    进入 Dataplex

  2. 转到进程视图。

  3. Tasks 标签页中会显示任务列表,这些任务会按任务模板类型进行过滤。

  4. 名称列中,点击要查看的任意任务。

  5. 点击要查看的任务的作业 ID

    Google Cloud 控制台中会打开 Dataproc 页面,您可以在其中查看监控和输出详情。

gcloud

下表列出了用于监控任务的 gcloud CLI 命令。

操作 gcloud CLI 命令
列出任务 gcloud dataplex tasks list --project=<project-name> --location=<location> --lake=<lake-id>
查看任务详细信息 gcloud dataplex tasks describe --project=<project-name> --location=<location> --lake=<lake-id> <task-id>
列出任务的作业 gcloud dataplex tasks jobs list --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id>
查看作业详情 gcloud dataplex tasks jobs describe --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>

Dataplex 在 Dataproc Serverless 上执行作业 (批次)。如需查看 Dataplex 作业的执行日志,请按以下步骤操作:

  1. 获取 Dataproc Serverless(批处理)作业 ID。运行以下命令:

    gcloud dataplex tasks jobs describe --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>
    
  2. 查看日志。使用您从中获得的作业 ID 运行以下命令 运行上一个命令:

    gcloud beta dataproc batches wait --project=<project-name> --region=<location> <job-id>
    

REST

如需 getlist 任务作业,请使用 API Explorer。

管理时间表

在 Google Cloud 控制台的 Dataplex 中,您可以修改 任务计划、删除任务或取消正在进行的作业。下表列出了执行这些操作的 gcloud CLI 命令。

操作 gcloud CLI 命令
修改任务时间表 gcloud dataplex tasks update --project=<project-name> --location=<location> --lake=<lake-id> --trigger-schedule=<updated-schedule> <task-id>
删除任务 gcloud dataplex tasks delete --project=<project-name> --location=<location> --lake=<lake-id> <task-id>
取消作业 gcloud dataplex tasks jobs cancel --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>

后续步骤