有没有一种方法可以通过python从本地触发云数据融合管道？

Question

我正在尝试构建一个代码，在其中我需要从一个基于桌面的应用程序触发云数据融合管道，这就是我正在构建的应用程序。它目前在python上。谁能建议一种不使用google云数据融合UI并使用一些python代码来启动管道的方法呢？

Answer 1

您可能会发现PycURL库正在接近cURL客户端HTTP方法发送方，或者可以用来从cURL代码触发对上述CDAP清单的调用。

下面的示例展示了Python代码，为启动提供了HTTP方法

PyCurl中的批处理管道，作为参考，我使用了与@Edwin提到的文档链接相同的环境变量：

设置环境变量：

export AUTH_TOKEN=$(gcloud auth print-access-token)

export CDAP_ENDPOINT=$(gcloud beta data-fusion instances describe \
--location=<region> \
--format="value(apiEndpoint)" \
${INSTANCE_ID})v3/namespaces/namespace-id/apps/pipeline-name/workflows/DataPipelineWorkflow/start

Python代码片段：

import pycurl
import os

CDAP_ENDPOINT = os.environ['CDAP_ENDPOINT']
AUTH_TOKEN = os.environ['AUTH_TOKEN']

c = pycurl.Curl()
c.setopt(pycurl.URL, CDAP_ENDPOINT)
c.setopt(pycurl.HTTPHEADER, ['Authorization: Bearer %s' %(AUTH_TOKEN)])
c.setopt(pycurl.POST, 1)
c.perform()

Answer 2

正如@mk_sta在推荐中提到的，云数据融合管道可以使用REST触发，具体而言：

POST -H "Authorization: Bearer ${AUTH_TOKEN}" "${CDAP_ENDPOINT}/v3/namespaces/<namespace-id>/apps/<pipeline-name>/workflows/DataPipelineWorkflow/start"

有关更多信息，您可以参考文档。

Answer 3

以补充/连接以上两个答案。要在命令行中运行，可以对HTTP请求使用curl：

> export AUTH_TOKEN=$(gcloud auth print-access-token)
> export REGION=your-region
> export INSTANCE_ID=your-instance
> export PIPELINE=your-pipeline
> export NAMESPACE=your-namespace

> export CDAP_ENDPOINT=$(gcloud beta data-fusion instances describe \
    --location=${REGION} \
    --format="value(apiEndpoint)" \
  ${INSTANCE_ID})

> curl -H "Authorization: Bearer ${AUTH_TOKEN}" --request POST "${CDAP_ENDPOINT}/v3/namespaces/${NAMESPACE}/apps/${PIPELINE}/workflows/DataPipelineWorkflow/start"

不需要使用命令行的完全Python解决方案是：

import requests
import os

# Set environment variables
AUTH_TOKEN = os.popen('gcloud auth print-access-token').read().strip()
REGION = 'your-region'
PIPELINE = 'your-pipeline_name'
NAMESPACE = 'your_namespace'
INSTANCE_ID = 'your-instance-id'

# CDAP endpoint of the instance
CDAP_ENDPOINT = os.popen('gcloud beta data-fusion instances describe --location=' +
                         REGION + ' --format="value(apiEndpoint)" ' + INSTANCE_ID).read().strip()

# Send post request
response = requests.post(CDAP_ENDPOINT + "/v3/namespaces/" + NAMESPACE + "/apps/" + PIPELINE + "/workflows/DataPipelineWorkflow/start",
                        headers={"Authorization": "Bearer " + AUTH_TOKEN})

确保您已通过身份验证才能运行这些命令，为此，请参阅文档：https://cloud.google.com/data-fusion/docs/reference/cdap-reference