Palantir Foundry中Webhook调用与Pipeline Builder集成问题咨询

我目前已经搭建好了代码仓库，并且导入了一个名为Gdelt2Retrieve的REST API作为数据源，同时在这个API里配置了一个名为Gdelt2的Webhook。我能在数据连接标签页中运行它，并且成功看到Webhook/API返回的正确结果，也已经设置好了出口策略（egress policies）。

我的核心需求是：

• 尝试在代码仓库中配置这个Webhook，最终目标是把这个功能导入到Pipeline Builder中，作为后续流水线的输入
• 如果不用函数的话，也希望能直接把Webhook放到Pipeline Builder里，让它每小时左右运行一次，并将结果存入数据集

我已经按照文档启用了transform-external-systems设置。根据我的理解，我需要调用这个Webhook，并且输出结果到数据集。

有人能帮我看看我写的这段代码方向对不对吗？我是照着文档示例写的：

from palantir.datasets.core import Dataset
from palantir.datasets.webhooks import WebhookClient
from pyspark.sql import function


@function(sources=["Gdelt2Retrieve"])  
def call_webhook() -> str:
    # Create a WebhookClient instance
    webhook_client = WebhookClient()

    # Execute the webhook
    try:
        response = webhook_client.execute("Gdelt2")
    except Exception as e:
        return f"Error: Webhook call failed due to an exception: {e}"

    # Check if the webhook execution was successful
    if response.status_code != 200:
        return f"Error: Webhook call failed with status code {response.status_code}, response: {response.text}"

    # Process the response data
    try:
        data = response.json()  # Extract JSON data
    except ValueError:
        return "Error:"


    return str(data)

另外，在我导入APIGdelt2Retrieve之后，资源侧边栏给了我一段起始代码：

import requests
@function(sources=["Gdelt2Retrieve"])
def my_function() -> String:
    # TODO: specify endpoint url
    response = requests.get(...)
    if response.status_code != 200:
        # Handle error
    data = response.json()
    # Use response data

我不一定非要用函数，但我觉得这是把Webhook集成到Pipeline Builder最简单的方式。

备注：内容来源于stack exchange，提问作者gmelgkermglermgmrelkgm