Google Dataprep 操作符¶
Dataprep 是智能云数据服务，可用于可视化地探索、清理和准备数据，以进行分析和机器学习。该服务可用于探索和转换来自不同和/或大型数据集的原始数据，将其转换为干净且结构化的数据，以进行进一步的分析和处理。Dataprep 作业是内部对象，编码运行 Cloud Dataprep 作业组的一部分所需的信息。有关该服务的更多信息，请访问 Google Dataprep API 文档
开始之前¶
在 Airflow 中使用 Dataprep 之前，您需要使用 TOKEN 对您的帐户进行身份验证。要将 Dataprep 与 Airflow 连接，您需要 Dataprep 令牌。请按照 Dataprep 说明 进行操作。
TOKEN 应以 JSON 格式添加到 Airflow 中的连接。您可以查看 管理连接
DataprepRunJobGroupOperator 将运行指定的作业。操作符需要一个配方 ID。要识别配方 ID，请使用 runJobGroup 的 API 文档。例如，如果 URL 是 /flows/10?recipe=7，则配方 ID 为 7。配方不能通过此操作符创建。它只能通过 UI 创建，UI 可在这里找到。某些参数可以被 DAG 的主体请求覆盖。如何在示例 dag 中显示。
请看以下示例：设置这些字段的值： .. code-block
Connection Id: "your_conn_id"
Extra: {"token": "TOKEN", "base_url": "https://api.clouddataprep.com"}
先决条件任务¶
要使用这些操作符，您必须完成以下几项操作
使用 Cloud Console 选择或创建一个 Cloud Platform 项目。
按照 Google Cloud 文档中所述，为您的项目启用结算功能。
按照 Cloud Console 文档中所述，启用 API。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'有关详细信息，请参阅 安装。
运行作业组¶
操作符的任务是创建一个作业组，该作业组以经过身份验证的用户身份启动指定的作业。这执行与在应用程序中单击“运行作业”按钮相同的操作。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepRunJobGroupOperator
示例用法
run_job_group_task = DataprepRunJobGroupOperator(
task_id="run_job_group",
dataprep_conn_id=CONNECTION_ID,
project_id=GCP_PROJECT_ID,
body_request={
"wrangledDataset": {"id": DATASET_WRANGLED_ID},
"overrides": WRITE_SETTINGS,
},
)
获取作业组的作业¶
操作符的任务是获取有关 Cloud Dataprep 作业中批处理作业的信息。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepGetJobsForJobGroupOperator
示例用法
get_jobs_for_job_group_task = DataprepGetJobsForJobGroupOperator(
task_id="get_jobs_for_job_group",
dataprep_conn_id=CONNECTION_ID,
job_group_id="{{ task_instance.xcom_pull('run_flow')['data'][0]['id'] }}",
)
获取作业组¶
操作符的任务是获取指定的作业组。作业组是从流中的特定节点执行的作业。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepGetJobGroupOperator
示例用法
get_job_group_task = DataprepGetJobGroupOperator(
task_id="get_job_group",
dataprep_conn_id=CONNECTION_ID,
project_id=GCP_PROJECT_ID,
job_group_id="{{ task_instance.xcom_pull('run_flow')['data'][0]['id'] }}",
embed="",
include_deleted=False,
)
复制流¶
操作符的任务是复制流。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepCopyFlowOperator
示例用法
copy_task = DataprepCopyFlowOperator(
task_id="copy_flow",
dataprep_conn_id=CONNECTION_ID,
project_id=GCP_PROJECT_ID,
flow_id=FLOW_ID,
name=f"copy_{DATASET_NAME}",
)
运行流¶
操作符的任务是运行流。流是用于管理逻辑的容器，其中包含导入的数据集、配方、输出对象和引用。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepRunFlowOperator
示例用法
run_flow_task = DataprepRunFlowOperator(
task_id="run_flow",
dataprep_conn_id=CONNECTION_ID,
project_id=GCP_PROJECT_ID,
flow_id=FLOW_COPY_ID,
body_request={},
)
删除流¶
操作符的任务是删除流。流是用于管理逻辑的容器，其中包含导入的数据集、配方、输出对象和引用。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepDeleteFlowOperator
示例用法
delete_flow_task = DataprepDeleteFlowOperator(
task_id="delete_flow",
dataprep_conn_id=CONNECTION_ID,
flow_id="{{ task_instance.xcom_pull('copy_flow')['id'] }}",
)
检查作业组是否已完成¶
传感器任务是告诉系统何时启动的作业组已完成，无论成功与否。作业组是从流中的特定节点执行的作业。
要获取有关 Cloud Dataprep 作业中作业的信息，请使用：DataprepJobGroupIsFinishedSensor
示例用法
check_flow_status_sensor = DataprepJobGroupIsFinishedSensor(
task_id="check_flow_status",
dataprep_conn_id=CONNECTION_ID,
job_group_id="{{ task_instance.xcom_pull('run_flow')['data'][0]['id'] }}",
)