Google Display & Video 360 Operators¶

Google Display & Video 360 拥有您所需的端到端营销活动管理功能。

前置任务¶

要使用这些 Operators，您必须执行以下几项操作

使用 Cloud Console 选择或创建 Cloud Platform 项目。

按照 Google Cloud 文档中的说明，为您的项目启用结算功能。

按照 Cloud Console 文档中的说明，启用 API。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'
安装的详细信息可用。
设置 Google Cloud 连接.

创建查询¶

要创建 Display&Video 360 查询，请使用 GoogleDisplayVideo360CreateQueryOperator。

tests/system/google/marketing_platform/example_display_video.py

create_query_v2 = GoogleDisplayVideo360CreateQueryOperator(body=REPORT_V2, task_id="create_query")

query_id = cast("str", XComArg(create_query_v2, key="query_id"))

对 body, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。您可以提供使用 `` .json`` 文件定义的 body，因为此 Operator 支持此模板扩展。结果将保存到 XCom 中，其他 Operator 可以使用该结果。

运行查询¶

GoogleDisplayVideo360RunQueryOperator.

tests/system/google/marketing_platform/example_display_video.py

run_query_v2 = GoogleDisplayVideo360RunQueryOperator(
    query_id=query_id, parameters=PARAMETERS, task_id="run_report"
)

query_id = cast("str", XComArg(run_query_v2, key="query_id"))
report_id = cast("str", XComArg(run_query_v2, key="report_id"))

对 query_id, parameters, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。结果将保存到 XCom 中，其他 Operator 可以使用该结果。

删除报告¶

要删除 Display&Video 360 报告，请使用 GoogleDisplayVideo360DeleteReportOperator。

tests/system/google/marketing_platform/example_display_video.py

delete_report_v2 = GoogleDisplayVideo360DeleteReportOperator(
    report_id=report_id, task_id="delete_report", trigger_rule=TriggerRule.ALL_DONE
)

对 report_id, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

等待查询¶

要等待报告，请使用 GoogleDisplayVideo360RunQuerySensor。

tests/system/google/marketing_platform/example_display_video.py

wait_for_query = GoogleDisplayVideo360RunQuerySensor(
    task_id="wait_for_query",
    query_id=query_id,
    report_id=report_id,
)

对 query_id, report_id, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

下载报告¶

要将报告下载到 GCS bucket，请使用 GoogleDisplayVideo360DownloadReportV2Operator。

tests/system/google/marketing_platform/example_display_video.py

get_report_v2 = GoogleDisplayVideo360DownloadReportV2Operator(
    query_id=query_id,
    report_id=report_id,
    task_id="get_report",
    bucket_name=BUCKET_NAME,
    report_name="test1.csv",
)

对 query_id, report_id, bucket_name, report_name, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

下载订单项¶

此 Operator 接受 body request

与 Google API 一致

REQUEST_BODY = {
"filterType": ADVERTISER_ID,
"format": "CSV",
"fileSpec": "EWF"
}

要下载 CSV 格式的订单项报告，请使用 GoogleDisplayVideo360DownloadLineItemsOperator。

tests/system/google/marketing_platform/example_display_video.py

download_line_items = GoogleDisplayVideo360DownloadLineItemsOperator(
    task_id="download_line_items",
    request_body=DOWNLOAD_LINE_ITEMS_REQUEST,
    bucket_name=BUCKET_NAME,
    object_name=OBJECT_NAME,
    gzip=False,
)

对 request_body, bucket_name, object_name, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

上传订单项¶

要运行 Display&Video 360 订单项上传，请使用 GoogleDisplayVideo360UploadLineItemsOperator。

tests/system/google/marketing_platform/example_display_video.py

upload_line_items = GoogleDisplayVideo360UploadLineItemsOperator(
    task_id="upload_line_items",
    bucket_name=BUCKET_NAME,
    object_name=BUCKET_FILE_LOCATION,
)

对 bucket_name, object_name, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

创建 SDF 下载任务¶

要创建 SDF 下载任务，请使用 GoogleDisplayVideo360CreateSDFDownloadTaskOperator。

tests/system/google/marketing_platform/example_display_video.py

create_sdf_download_task = GoogleDisplayVideo360CreateSDFDownloadTaskOperator(
    task_id="create_sdf_download_task", body_request=CREATE_SDF_DOWNLOAD_TASK_BODY_REQUEST
)
operation_name = '{{ task_instance.xcom_pull("create_sdf_download_task")["name"] }}'

对 body_request, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

将 SDF 文件保存在 Google Cloud Storage 中¶

要保存 SDF 文件并将其保存在 Google Cloud Storage 中，请使用 GoogleDisplayVideo360SDFtoGCSOperator。

tests/system/google/marketing_platform/example_display_video.py

save_sdf_in_gcs = GoogleDisplayVideo360SDFtoGCSOperator(
    task_id="save_sdf_in_gcs",
    operation_name=operation_name,
    bucket_name=BUCKET_NAME,
    object_name=BUCKET_FILE_LOCATION,
    gzip=False,
)

对 operation_name, bucket_name, object_name, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。

等待 SDF 操作¶

通过以下方式执行等待 SDF 操作：GoogleDisplayVideo360GetSDFDownloadOperationSensor。

tests/system/google/marketing_platform/example_display_video.py

wait_for_operation = GoogleDisplayVideo360GetSDFDownloadOperationSensor(
    task_id="wait_for_operation",
    operation_name=operation_name,
)

对 operation_name, impersonation_chain 参数使用 Jinja 模板，这些参数允许您动态确定值。