仅在文件存在时运行数据融合管道

Question

我在 Data Fusion 中已经有一个工作管道，它使所有 ETL 进程，但只有当它在 Cloud Storage 存储桶中找到一个名为 SUCCESS.txt 的文件时，我才需要它运行。
这可能吗？
在其他平台上，我使用了文件观察器（每分钟运行一个作业来验证我指定的文件是否存在于某个位置，如果文件在那里，它会执行其他作业）但我找不到类似的东西。
提前非常感谢！

Answer 1

您可以通过使用Cloud Functions GCS triggers 来实现此目的，条件是调用Data Fusion API 以仅在上传的文件为SUCCESS.txt 时启动您的管道。
请注意，无论是否调用Data Fusion API，该函数都会在每次上传文件时触发。

创建云函数时：

1。选择 Cloud Storage 触发器类型和 Finalize/Create 事件类型。

2。使用您自己的值添加 environment variables，然后单击下一步。

3。将运行时设置为 python 3.7，入口点中 python 函数的名称（在本例中为run_pipeline），并在main.py 中添加您的 python 脚本（或下面的示例）。

import requests
import json
import os

def get_access_token():

    # scope of the API access. Limit it to just cloud platform services
    scopes='https://www.googleapis.com/auth/cloud-platform'
    headers={'Metadata-Flavor': 'Google'}

    # add the scopes
    api="http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token?scopes=" + scopes

    # api call to get the access token
    r = requests.get(api,headers=headers).json()

    # return the access token
    return r['access_token']

def run_pipeline(data, context):
    '''
    Calls the Data Fusion API to start the pipeline
    '''
    
    # get environmental variables set in the inital configuraiton.
    PROJECT_ID=os.environ.get('PROJECT_ID', 'Specified environment variable is not set.')
    TOPIC_ID=os.environ.get('TOPIC_ID', 'Specified environment variable is not set.')
    PIPELINE_NAME=os.environ.get('PIPELINE_NAME', 'Specified environment variable is not set.')
    INSTANCE_ID=os.environ.get('INSTANCE_ID', 'Specified environment variable is not set.')
    REGION=os.environ.get('REGION', 'Specified environment variable is not set.')
    NAMESPACE_ID=os.environ.get('NAMESPACE_ID', 'Specified environment variable is not set.')
    CDAP_ENDPOINT=os.environ.get('CDAP_ENDPOINT', 'Specified environment variable is not set.')

    # get uploaded file name
    file_name = data['name']
    
    # get access token
    auth_token=get_access_token()
    
    # api call full endpoint
    post_endpoint = CDAP_ENDPOINT + "/v3/namespaces/" + NAMESPACE_ID + "/apps/" + PIPELINE_NAME + "/workflows/DataPipelineWorkflow/start"
    
    # If the pipeline has any macros that need to be set, you can pass them in as a payload
    data = '{"my-file":' + file_name +'}'
    
    # add bearer token to the header
    post_headers = {"Authorization": "Bearer " + auth_token,"Accept": "application/json"}
    
    # condition to start the job:
    if file_name == 'SUCCESS.txt':
        # start the job
        r1 = requests.post(post_endpoint,data=data,headers=post_headers)

4。部署您的函数，当它准备就绪时，通过上传您的 SUCCESS.txt 文件或任何其他文件对其进行测试。

我已经对其进行了测试，它工作正常（基于此post）。