Dataplex Universal Catalog 支持安排自定义代码执行，可以是一次性运行、定期运行或按需运行。按需处于预览版阶段，仅可通过 API 使用。您可以使用 Spark (Java)、PySpark（仅限 Spark 3.2 版）或 Spark SQL 安排客户数据转换。Dataplex Universal Catalog 会使用无服务器 Spark 处理和内置无服务器调度器执行代码。
术语
- 任务
- Dataplex Universal Catalog 任务表示您希望 Dataplex Universal Catalog 按时间表执行的工作。它封装了您的代码、参数和时间表。
- 作业
作业表示 Dataplex Universal Catalog 任务的单次运行。例如，如果任务安排为每天运行，则 Dataplex Universal Catalog 每天都会创建一个作业。
对于 2023 年 5 月 10 日或之后创建的作业，触发器字段会显示作业的执行触发器类型。
以下是作业执行触发器类型：
RUN_REQUEST：表示作业是因调用
RunTaskAPI 而执行的。TASK_CONFIG：表示作业是因任务的
TriggerSpec配置而执行的。
安排模式
Dataplex Universal Catalog 支持以下安排模式：
- 运行一次
- 使用此模式可仅运行一次任务。您可以选择立即运行，也可以选择在未来的设置时间运行。如果您立即运行任务，执行操作可能仍需最多两分钟才能开始。
- 按计划运行
- 使用此模式可按重复的频率运行任务。支持的重复频率包括每天、每周、每月或自定义。
- 按需运行
使用此模式可按需运行之前创建的任务。只有
RunTaskAPI 支持按需运行模式。当作业按需运行时，Dataplex Universal Catalog 会使用现有参数来创建作业。您可以指定ExecutionSpec参数和标签来执行作业。
准备工作
启用 Dataproc API。
为您的网络和子网启用专用 Google 访问通道。在您用于 Dataplex Universal Catalog 任务的网络上启用专用 Google 访问通道。如果您在创建 Dataplex Universal Catalog 任务时未指定网络或子网，Dataplex Universal Catalog 会使用默认子网，并且您必须为默认子网启用专用 Google 访问通道。
创建服务账号。您必须拥有服务账号才能安排任何 Dataplex Universal Catalog 任务。服务账号必须属于您在其中执行任务的项目。该服务账号必须具有以下权限：
对正在处理的 BigQuery 和 Cloud Storage 数据的访问权限。
您在其中执行任务的项目的 Dataproc Worker 角色权限。
如果任务需要读取或更新附加到数据湖的 Dataproc Metastore 实例，则服务账号需要具有 Dataproc Metastore Viewer 或 Editor 角色。必须在设置了 Dataplex Universal Catalog 数据湖的项目中授予此角色。
如果任务是 Spark SQL 作业，您需要向服务账号授予 Dataplex Universal Catalog Developer 角色。必须在设置了 Dataplex Universal Catalog 数据湖的项目中授予此角色。
如果任务是 Spark SQL 作业，您需要拥有在其中写入结果的存储桶的 Cloud Storage 管理员权限。
如需安排和运行 Spark SQL 及自定义 Spark 任务，您必须获得您的服务账号的 Dataplex Universal Catalog Metadata Reader (
roles/dataplex.metadataReader)、Dataplex Universal Catalog Viewer (roles/dataplex.viewer) 和 Dataproc Metastore Metadata User (roles/metastore.metadataUser) IAM 角色。
向提交作业的用户授予服务账号的 Service Account User 角色 (
roles/iam.serviceAccountUser)。如需查看相关说明，请参阅管理对服务账号的访问权限。向 Dataplex Universal Catalog 数据湖服务账号授予使用该服务账号的权限。您可以在Google Cloud 控制台的数据湖详情页面中找到 Dataplex Universal Catalog 数据湖服务账号。
如果包含 Dataplex Universal Catalog 数据湖的项目与在其中执行任务的项目不同，请为 Dataplex Universal Catalog 数据湖服务账号授予您在其中执行任务的项目的 Dataproc Editor 角色。
将所需的代码制品（JAR、Python 或 SQL 脚本文件）或归档文件（
.jar、.tar、.tar.gz、.tgz、.zip）放在 Cloud Storage 路径中。确保服务账号具有存储这些代码制品的 Cloud Storage 存储桶的所需
storage.objects.get权限。
安排 Spark（Java 或 Python）任务
控制台
在 Google Cloud 控制台中，前往 Dataplex Universal Catalog 页面。
前往进程视图。
点击创建任务。
在创建自定义 Spark 任务部分，点击创建任务。
选择一个 Dataplex Universal Catalog 数据湖。
提供任务名称。
为任务创建 ID。
在任务配置部分的类型字段中，选择 Spark 或 PySpark。
输入相关参数。
在服务账号字段中，输入您的自定义 Spark 任务执行时可以使用的用户服务账号。
点击继续。
可选：设置时间表：选择运行一次或重复。填写必填字段。
点击继续。
可选：自定义资源和添加其他设置。
点击创建。
gcloud
您可以使用 gcloud CLI 命令安排 Spark (Java / Python) 任务。下表列出了要使用的必需参数和可选参数：
| 参数 | 说明 |
|---|---|
--lake |
Dataplex Universal Catalog 服务的数据湖资源的数据湖 ID。 |
--location |
Dataplex Universal Catalog 服务的位置。 |
--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 |
可选：可配置的执行器的数量上限。默认值为 1,000。如果 batch-max-executors-count 大于 batch-executors-count，则 Dataplex Universal Catalog 会启用自动扩缩。
|
--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 Universal Catalog 使用项目中名为 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) 2017 年 1 月 15 日 01:30。如果未指定此值，则当触发器类型为 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
如需创建任务，请使用 APIs Explorer。
监控任务
控制台
在 Google Cloud 控制台中，前往 Dataplex Universal Catalog 页面：
前往进程视图。
在任务标签页中，系统会显示按任务模板类型过滤的任务列表。
在名称列中，点击您要查看的任何任务。
点击要查看的任务的作业 ID。
Dataproc 页面会在Google Cloud 控制台中打开，可让您查看监控和输出详情。
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 Universal Catalog 在 Dataproc Serverless（批处理）上执行作业。如需查看 Dataplex Universal Catalog 作业的执行日志，请按照以下步骤操作：
获取 Dataproc Serverless（批处理）作业 ID。运行以下命令：
gcloud dataplex tasks jobs describe --project=<project-name> --location=<location> --lake=<lake-id> --task=<task-id> <job-id>查看日志。 运行以下命令，并使用您通过运行上述命令获得的作业 ID：
gcloud beta dataproc batches wait --project=<project-name> --region=<location> <job-id>
REST
管理时间表
在 Google Cloud 控制台的 Dataplex Universal Catalog 中，您可以修改任务的时间表、删除任务或取消正在运行的作业。下表列出了这些操作对应的 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> |
后续步骤
- 请参阅 Dataproc 模板。
- 尝试使用预构建的模板，以将数据从 Dataplex Universal Catalog Cloud Storage 资产增量移动到 BigQuery。
- 请参阅为 Dataplex Universal Catalog 任务设置提醒和通知。