使用Operators（执行器）

译者：@ImPerat0R_、@ThinkingChen

operator（执行器）代表一个理想情况下是幂等的任务。operator（执行器）决定了DAG运行时实际执行的内容。

有关更多信息，请参阅Operators Concepts文档和Operators API Reference 。

• BashOperator
  • 模板
  • 故障排除
    • 找不到Jinja模板
• PythonOperator
  • 传递参数
  • 模板
• Google Cloud Platform Operators
  • GoogleCloudStorageToBigQueryOperator
  • GceInstanceStartOperator
  • GceInstanceStopOperator
  • GceSetMachineTypeOperator
  • GcfFunctionDeleteOperator
    • 故障排除
  • GcfFunctionDeployOperator
    • 故障排除
  • CloudSqlInstanceDatabaseCreateOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息
  • CloudSqlInstanceDatabaseDeleteOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息
  • CloudSqlInstanceDatabasePatchOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息
  • CloudSqlInstanceDeleteOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息
  • CloudSqlInstanceCreateOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息
  • CloudSqlInstancePatchOperator
    • 参数
    • 使用执行器
    • 模版
    • 更多信息

BashOperator

使用BashOperator在Bash shell中执行命令。

run_this = BashOperator(
    task_id='run_after_loop',
    bash_command='echo 1',
    dag=dag)

模板

您可以使用Jinja模板来参数化bash_command参数。

also_run_this = BashOperator(
    task_id='also_run_this',
    bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"',
    dag=dag,
)

故障排除

找不到Jinja模板

在使用bash_command参数直接调用Bash脚本时，需要在脚本名称后添加空格。这是因为Airflow尝试将Jinja模板应用于一个失败的脚本。

t2 = BashOperator(
    task_id='bash_example',
    # 这将会出现`Jinja template not found`的错误
    # bash_command="/home/batcher/test.sh",
    # 在加了空格之后，这会正常工作
    bash_command="/home/batcher/test.sh ",
    dag=dag)

PythonOperator

使用PythonOperator执行Python回调。

 def print_context ( ds , ** kwargs ):
    pprint ( kwargs )
    print ( ds )
    return 'Whatever you return gets printed in the logs'
run_this = PythonOperator (
    task_id = 'print_the_context' ,
    provide_context = True ,
    python_callable = print_context ,
    dag = dag )

传递参数

使用op_args和op_kwargs参数将额外参数传递给Python的回调函数。

def my_sleeping_function(random_base):
    """这是一个将在DAG执行体中运行的函数"""
    time.sleep(random_base)
# Generate 10 sleeping tasks, sleeping from 0 to 4 seconds respectively
for i in range(5):
    task = PythonOperator(
        task_id='sleep_for_' + str(i),
        python_callable=my_sleeping_function,
        op_kwargs={'random_base': float(i) / 10},
        dag=dag,
    )
    run_this >> task

模板

当您将provide_context参数设置为True，Airflow会传入一组额外的关键字参数：一个用于每个Jinja模板变量和一个templates_dict参数。

templates_dict参数是模板化的，因此字典中的每个值都被评估为Jinja模板。

Google 云平台 Operators（执行器）

GoogleCloudStorageToBigQueryOperator

使用GoogleCloudStorageToBigQueryOperator执行BigQuery加载作业。

GceInstanceStartOperator

允许启动一个已存在的Google Compute Engine实例。

在此示例中，参数值从Airflow变量中提取。此外，default_args字典用于将公共参数传递给单个DAG中的所有operator（执行器）。

PROJECT_ID = models.Variable.get('PROJECT_ID', '')
LOCATION = models.Variable.get('LOCATION', '')
INSTANCE = models.Variable.get('INSTANCE', '')
SHORT_MACHINE_TYPE_NAME = models.Variable.get('SHORT_MACHINE_TYPE_NAME', '')
SET_MACHINE_TYPE_BODY = {
    'machineType': 'zones/{}/machineTypes/{}'.format(LOCATION, SHORT_MACHINE_TYPE_NAME)
}
default_args = {
    'start_date': airflow.utils.dates.days_ago(1)
}

通过将所需的参数传递给构造函数来定义GceInstanceStartOperator。

gce_instance_start = GceInstanceStartOperator(
    project_id=PROJECT_ID,
    zone=LOCATION,
    resource_id=INSTANCE,
    task_id='gcp_compute_start_task'
)

GceInstanceStopOperator

允许停止一个已存在的Google Compute Engine实例。

参数定义请参阅上面的GceInstanceStartOperator。

通过将所需的参数传递给构造函数来定义GceInstanceStopOperator。

gce_instance_stop = GceInstanceStopOperator(
    project_id=PROJECT_ID,
    zone=LOCATION,
    resource_id=INSTANCE,
    task_id='gcp_compute_stop_task'
)

GceSetMachineTypeOperator

允许把一个已停止实例的机器类型改变至特定的类型。

参数定义请参阅上面的GceInstanceStartOperator。

通过将所需的参数传递给构造函数来定义GceSetMachineTypeOperator。

gce_set_machine_type = GceSetMachineTypeOperator(
    project_id=PROJECT_ID,
    zone=LOCATION,
    resource_id=INSTANCE,
    body=SET_MACHINE_TYPE_BODY,
    task_id='gcp_compute_set_machine_type'
)

GcfFunctionDeleteOperator

使用default_args字典来传递参数给operator（执行器）。

PROJECT_ID = models.Variable.get('PROJECT_ID', '')
LOCATION = models.Variable.get('LOCATION', '')
ENTRYPOINT = models.Variable.get('ENTRYPOINT', '')
# A fully-qualified name of the function to delete
FUNCTION_NAME = 'projects/{}/locations/{}/functions/{}'.format(PROJECT_ID, LOCATION,
                                                               ENTRYPOINT)
default_args = {
    'start_date': airflow.utils.dates.days_ago(1)
}

使用GcfFunctionDeleteOperator来从Google Cloud Functions删除一个函数。

t1 = GcfFunctionDeleteOperator(
    task_id="gcf_delete_task",
    name=FUNCTION_NAME
)

故障排除

如果你想要使用服务账号来运行或部署一个operator（执行器），但得到了一个403禁止的错误，这意味着你的服务账号没有正确的Cloud IAM权限。

1. 指定该服务账号为Cloud Functions Developer角色。
2. 授权Cloud Functions的运行账户为Cloud IAM Service Account User角色。

使用gcloud分配Cloud IAM权限的典型方法如下所示。只需将您的Google Cloud Platform项目ID替换为PROJECT_ID，将SERVICE_ACCOUNT_EMAIL替换为您的服务帐户的电子邮件ID即可。

gcloud iam service-accounts add-iam-policy-binding \
  PROJECT_ID@appspot.gserviceaccount.com \
  --member="serviceAccount:[SERVICE_ACCOUNT_EMAIL]" \
  --role="roles/iam.serviceAccountUser"

细节请参阅Adding the IAM service agent user role to the runtime service

GcfFunctionDeployOperator

使用GcfFunctionDeployOperator来从Google Cloud Functions部署一个函数。

以下Airflow变量示例显示了您可以使用的default_args的各种变体和组合。变量定义如下：

PROJECT_ID = models.Variable.get('PROJECT_ID', '')
LOCATION = models.Variable.get('LOCATION', '')
SOURCE_ARCHIVE_URL = models.Variable.get('SOURCE_ARCHIVE_URL', '')
SOURCE_UPLOAD_URL = models.Variable.get('SOURCE_UPLOAD_URL', '')
SOURCE_REPOSITORY = models.Variable.get('SOURCE_REPOSITORY', '')
ZIP_PATH = models.Variable.get('ZIP_PATH', '')
ENTRYPOINT = models.Variable.get('ENTRYPOINT', '')
FUNCTION_NAME = 'projects/{}/locations/{}/functions/{}'.format(PROJECT_ID, LOCATION,
                                                               ENTRYPOINT)
RUNTIME = 'nodejs6'
VALIDATE_BODY = models.Variable.get('VALIDATE_BODY', True)

使用这些变量，您可以定义请求的主体：

body = {
    "name": FUNCTION_NAME,
    "entryPoint": ENTRYPOINT,
    "runtime": RUNTIME,
    "httpsTrigger": {}
}

创建DAG时，default_args字典可用于传递正文和其他参数：

default_args = {
    'start_date': dates.days_ago(1),
    'project_id': PROJECT_ID,
    'location': LOCATION,
    'body': body,
    'validate_body': VALIDATE_BODY
}

请注意，在上面的示例中，body和default_args都是不完整的。根据设置的变量，如何传递源代码相关字段可能有不同的变体。目前，您可以传递sourceArchiveUrl，sourceRepository或sourceUploadUrl，CloudFunction API规范中所述。此外，default_args可能包含zip_path参数，以在部署源代码之前运行上载源代码的额外步骤。在最后一种情况下，您还需要在正文中提供一个空的sourceUploadUrl参数。

基于上面定义的变量，此处显示了设置源代码相关字段的示例逻辑：

if SOURCE_ARCHIVE_URL:
    body['sourceArchiveUrl'] = SOURCE_ARCHIVE_URL
elif SOURCE_REPOSITORY:
    body['sourceRepository'] = {
        'url': SOURCE_REPOSITORY
    }
elif ZIP_PATH:
    body['sourceUploadUrl'] = ''
    default_args['zip_path'] = ZIP_PATH
elif SOURCE_UPLOAD_URL:
    body['sourceUploadUrl'] = SOURCE_UPLOAD_URL
else:
    raise Exception("Please provide one of the source_code parameters")

创建operator（执行器）的代码如下：

deploy_task = GcfFunctionDeployOperator(
    task_id="gcf_deploy_task",
    name=FUNCTION_NAME
)

Troubleshooting

如果你想要使用服务账号来运行或部署一个operator（执行器），但得到了一个403禁止的错误，这意味着你的服务账号没有正确的Cloud IAM权限。

1. 指定该服务账号为Cloud Functions Developer角色。
2. 授权Cloud Functions的运行账户为Cloud IAM Service Account User角色。

使用gcloud分配Cloud IAM权限的典型方法如下所示。只需将您的Google Cloud Platform项目ID替换为PROJECT_ID，将SERVICE_ACCOUNT_EMAIL的替换为您的服务帐户的电子邮件ID即可。

gcloud iam service-accounts add-iam-policy-binding \
  PROJECT_ID@appspot.gserviceaccount.com \
  --member="serviceAccount:[SERVICE_ACCOUNT_EMAIL]" \
  --role="roles/iam.serviceAccountUser"

细节请参阅Adding the IAM service agent user role to the runtime service

如果您的函数的源代码位于Google Source Repository中，请确保您的服务帐户具有Source Repository Viewer角色，以便在必要时可以下载源代码。

CloudSqlInstanceDatabaseCreateOperator

在Cloud SQL实例中创建新数据库。

有关参数定义，请参阅上面的GceInstanceStartOperator。

通过将所需的参数传递给构造函数来定义CloudSqlInstanceDatabaseCreateOperator。

参数

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

使用operator（执行器）

sql_db_create_task = CloudSqlInstanceDatabaseCreateOperator(
    project_id=PROJECT_ID,
    body=db_create_body,
    instance=INSTANCE_NAME,
    task_id='sql_db_create_task'
)

示例请求体：

db_create_body = {
    "instance": INSTANCE_NAME,
    "name": DB_NAME,
    "project": PROJECT_ID
}

模版

template_fields = ('project_id', 'instance', 'gcp_conn_id', 'api_version')

更多信息

有关数据库插入，请参阅Google Cloud SQL API文档。

CloudSqlInstanceDatabaseDeleteOperator

在Cloud SQL实例中删除数据库。

有关参数定义，请参阅CloudSqlInstanceDatabaseDeleteOperator。

参数

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

使用operator（执行器）

sql_db_delete_task = CloudSqlInstanceDatabaseDeleteOperator(
    project_id=PROJECT_ID,
    instance=INSTANCE_NAME,
    database=DB_NAME,
    task_id='sql_db_delete_task'
)

模版

template_fields = ('project_id', 'instance', 'database', 'gcp_conn_id',
                   'api_version')

更多信息

有关数据库删除，请参阅Google Cloud SQL API文档。

CloudSqlInstanceDatabasePatchOperator

使用修补程序语义更新包含有关Cloud SQL实例内数据库的信息的资源。请参阅: https://cloud.google.com/sql/docs/mysql/admin-api/how-tos/performance#patch

有关参数定义，请参阅CloudSqlInstanceDatabasePatchOperator。

参数

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

使用operator（执行器）

sql_db_patch_task = CloudSqlInstanceDatabasePatchOperator(
    project_id=PROJECT_ID,
    body=db_patch_body,
    instance=INSTANCE_NAME,
    database=DB_NAME,
    task_id='sql_db_patch_task'
)

示例请求体：

db_patch_body = {
    "charset": "utf16",
    "collation": "utf16_general_ci"
}

模版

template_fields = ('project_id', 'instance', 'database', 'gcp_conn_id',
                   'api_version')

更多信息

有关数据库修改，请参阅Google Cloud SQL API文档。

CloudSqlInstanceDeleteOperator

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

使用operator（执行器）

sql_instance_delete_task = CloudSqlInstanceDeleteOperator(
    project_id=PROJECT_ID,
    instance=INSTANCE_NAME,
    task_id='sql_instance_delete_task'
)

模版

template_fields = ('project_id', 'instance', 'gcp_conn_id', 'api_version')

更多信息

有关删除，请参阅Google Cloud SQL API文档。

CloudSqlInstanceCreateOperator

在Google Cloud Platform中创建新的Cloud SQL实例。

有关参数定义，请参阅CloudSqlInstanceCreateOperator。

如果存在具有相同名称的实例，则不会执行任何操作，并且operator（执行器）将成功执行。

参数

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

定义实例的示例：

body = {
    "name": INSTANCE_NAME,
    "settings": {
        "tier": "db-n1-standard-1",
        "backupConfiguration": {
            "binaryLogEnabled": True,
            "enabled": True,
            "startTime": "05:00"
        },
        "activationPolicy": "ALWAYS",
        "dataDiskSizeGb": 30,
        "dataDiskType": "PD_SSD",
        "databaseFlags": [],
        "ipConfiguration": {
            "ipv4Enabled": True,
            "requireSsl": True,
        },
        "locationPreference": {
            "zone": "europe-west4-a"
        },
        "maintenanceWindow": {
            "hour": 5,
            "day": 7,
            "updateTrack": "canary"
        },
        "pricingPlan": "PER_USE",
        "replicationType": "ASYNCHRONOUS",
        "storageAutoResize": False,
        "storageAutoResizeLimit": 0,
        "userLabels": {
            "my-key": "my-value"
        }
    },
    "databaseVersion": "MYSQL_5_7",
    "region": "europe-west4",
}

使用operator（执行器）

sql_instance_create_task = CloudSqlInstanceCreateOperator(
    project_id=PROJECT_ID,
    body=body,
    instance=INSTANCE_NAME,
    task_id='sql_instance_create_task'
)

模版

template_fields = ('project_id', 'instance', 'gcp_conn_id', 'api_version')

更多信息

有关插入，请参阅Google Cloud SQL API文档。

CloudSqlInstancePatchOperator

更新Google Cloud Platform中的Cloud SQL实例的设置（部分更新）。

有关参数定义，请参阅CloudSqlInstancePatchOperator。

这是部分更新，因此仅设置/更新正文中指定的设置的值。现有实例的其余部分将保持不变。

参数

示例DAG中的一些参数取自环境变量：

PROJECT_ID = os.environ.get('PROJECT_ID', 'example-project')
INSTANCE_NAME = os.environ.get('INSTANCE_NAME', 'testinstance')
DB_NAME = os.environ.get('DB_NAME', 'testdb')

定义实例的示例：

patch_body = {
    "name": INSTANCE_NAME,
    "settings": {
        "dataDiskSizeGb": 35,
        "maintenanceWindow": {
            "hour": 3,
            "day": 6,
            "updateTrack": "canary"
        },
        "userLabels": {
            "my-key-patch": "my-value-patch"
        }
    }
}

使用operator（执行器）

sql_instance_patch_task = CloudSqlInstancePatchOperator(
    project_id=PROJECT_ID,
    body=patch_body,
    instance=INSTANCE_NAME,
    task_id='sql_instance_patch_task'
)

模版

template_fields = ('project_id', 'instance', 'gcp_conn_id', 'api_version')

更多信息

有关部分更新，请参阅Google Cloud SQL API文档。