自定义气流操作员在 Salesforce 表中运行查询

如何解决自定义气流操作员在 Salesforce 表中运行查询

我是气流的新手，我想安排一个作业，其中来自不同数据库记录计数的两个表必须检查它是否匹配。一个来源是 GCP，另一个来源是 Salesforce。

所以我找到了 BigQueryOperator 在 GCP 端点击查询并返回计数结果，但我找不到任何看起来像 SalesforceQueryOperator 的运算符，我可以在 Airflow 任务中分配它。

所以基本上，我说的是我们可以用来带来计数结果的这个：

t1 = BigQueryOperator(
        task_id='bigquery_test',bql='SELECT COUNT(userId) FROM [events:EVENTS_20160501]',destination_dataset_table=False,bigquery_conn_id='bigquery_default',google_cloud_storage_conn_id='bigquery_default',delegate_to=False,udf_config=False,dag=dag,)

我知道我们可以创建一个函数、导入库、创建与 Salesforce 的连接并运行查询以获取 Count 结果，但我不想遵循我已经在下面给出的这种方法（代码的一部分）试过了。

def salesforcequery_count():
from simple_salesforce import Salesforce
import requests

session = requests.Session()
# manipulate the session instance (optional)
sf = Salesforce(
   username='[e-mail]',password='password',organizationId='OrgId',session=session)
   count_record = sf.query("SELECT count(id) FROM Contact")
//   for row in data:
//   process(row)
    return 'count_record'

我想创建一个看起来像 SalesforceQueryOperator 并且应该像 BigQueryOperator 一样的自定义运算符来点击 Salesforce 表中的查询并带来结果。

这里是参考：https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html

任何帮助将不胜感激。

解决方法

您可以使用现有的 SalesforceHook 创建您自己的自定义运算符。

这是一个例子：

from airflow.contrib.hooks.salesforce_hook import SalesforceHook
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults


class SalesforceQueryOperator(BaseOperator):
    """
    Make a query against Salesforce
    Return result as dict.
    """
    template_fields = ("query",)

    @apply_defaults
    def __init__(self,conn_id,query=None,*args,**kwargs
                 ):
        super(SalesforceQueryOperator,self).__init__(*args,**kwargs)

        self.conn_id = conn_id
        self.query = query

    def execute(self,context):
        sf_hook = SalesforceHook(conn_id=self.conn_id)

        results = sf_hook.make_query(self.query)

        return results

然后在您的 DAG 中使用它：

t2 = SalesforceQueryOperator(
        task_id='salesforce_test',query='SELECT count(id) FROM Contact',conn_id='salesforce_default',dag=dag,)

其中 salesforce_default 是您在 AirFlow 中添加的连接。您可以在此处查看如何添加它：Salesforce Connection