使用现有API成为雪花数据市场提供商，第2部分2020年6月25日|10分钟读数作者：贾斯汀·朗塞斯数据交换几乎所有这些数据提供者都有一个restapi。Snowflake最近发布的外部函数功能允许Snowflake帐户调用外部api。通过使用外部函数，数据充实提供者可以满足来自雪花数据市场消费者的数据请求。通过代表客户调用他们自己的api，从消费者所在地区的Snowflake帐户，数据提供者可以简单地将结果数据共享给客户的Snowflake帐户通过这种方法，可以将数据丰富提供者快速添加到雪花数据市场中，并通过使用其现有的API立即满足任何雪花云或区域中的消费者请求，从而消除了提供商设计新数据流的需要，并消除了跨区域数据复制的需要这篇博客文章的第一部分介绍了Quantifind，一家金融犯罪预防公司，利用这种基于API的方法，在所有雪花地区的雪花数据市场上提供数据丰富服务，并以数据提供商现有的API为后盾这一部分提供了一个循序渐进的例子来说明Quantifind是如何做到这一点的先决条件要使Quantifind启用数据充实并开始自动满足消费者请求，数据提供程序需要运行一个设置脚本。他们在每个有顾客的雪花区只做一次要继续执行此示例，在继续之前，您需要让雪花销售工程师在您的帐户上启用名为enable_STREAMS_on_SHARED_TABLES的帐户参数，并将其设置为TRUE。这是必需的，因为共享表上的流当前是一个预览功能。端到端数据流图图1显示了我们将设置的端到端数据流。它显示了从一个示例消费者帐户到API，再到消费者帐户的数据流。图1：基于API的数据共享方法的流程图。启用API支持的数据扩展的步骤现在，让我们一步一步地看一下量词示例。步骤1：创建数据库和仓库首先，创建一个用于保存外部函数的数据库、一个作业跟踪表，以及要共享给客户的结果数据。还要创建一个新的雪花计算实例（"虚拟仓库"）来为请求提供服务，以便将此服务的资源与其他雪花作业隔离开来，并允许它们在没有作业挂起时自动挂起。使用ACCOUNTADMIN角色；创建或替换数据库QuantifNd_数据；使用数据库量化数据；创建或替换仓库提供者_ENRICH_WH WITH WAREHOUSE_SIZE='SMALL'仓库类型="标准"自动挂起=60自动继续=真最小聚类数=1最大聚类数=1SCALING_POLICY='标准'；步骤2：配置awsapi网关和AWS Lambda函数，并测试该函数继续之前，请配置AWS API网关和AWS Lambda函数，该函数将通过执行本页以下部分中的过程来调用远程REST API：步骤1：创建远程服务（AWS上的Lambda函数）步骤2：配置代理服务（AWS上的API网关）并创建API集成（在雪花中）第三步：创建外部函数第四步：调用外部函数现在，实现并测试该页上描述的示例函数。我们将用实际的lambda替换它，以便在后面的步骤中调用Quantifind的API注意：在"步骤2：配置代理服务（AWS上的API网关）并创建API集成（在雪花中）"一节中，当您进入"当要求"指定可以使用此角色的帐户"时，粘贴先前保存的云平台帐户Id，""您应该输入您自己的AWS帐户Id号，您可以找到它在AWS支持选项卡上。同样在该步骤中，如果在保存更新的资源策略时遇到"无效主体"错误，请离开页面并返回。尽管出现错误，您可能会看到策略已被保存，您应该能够重新保存它以确认并继续。步骤3：更新lambda函数用这个Quantifind函数替换示例AWS Lambda函数，然后单击saveinlambda。步骤4：创建API集成如果您已经在步骤2中创建了一个API集成，则可以重用它。如果要创建一个新的角色，请使用以下代码，用正确的值替换role\u arn和允许的前缀。如果您替换了现有的集成，您可能还需要更新AWS端的安全策略设置。创建或替换api集成Quantification\u apiapi_provider=aws_api_网关api_aws_role_arn='arn:aws公司：iam:：\启用=真api允许的前缀=('https://gu41r3343pc.execute-api.us-east-2.amazonaws.com/deploy/');描述集成量化api；//您需要下面的2个字段来更新AWS策略设置//用户arn____//外部id____步骤5：创建外部函数现在，创建用于调用远程API的外部函数。在这两个语句中，都需要替换api_integration字段以匹配集成的名称，并替换整个"as"字符串以匹配特定的awsapi集成URL。创建或替换外部函数quantifindd\u api（tablename varchar，app\u namevarchar，app\u token varchar，base\u url varchar）返回变量api_integration=量化和api最大批处理行数=200作为'https://gu41r343pc.execute-api.us-east-2.amazonaws.com/deploy/quantifind-api'    ;创建或替换外部函数quantifindd\u api（tablename varchar，app\u namevarchar，应用程序令牌varchar，基本url varchar，负载varchar）返回变量api_integration=量化和api最大批处理行数=200作为'https://gu41r343pc.execute-api.us-east-2.amazonaws.com/deploy/quantifind-api'    ;注意，这里创建了两个版本的函数：一个带有有效载荷字段，用于提交实际数据以进行扩展；另一个没有，用于检查现有作业的状态。这两个版本可以共存，根据调用函数时使用的参数，将自动使用正确的参数。步骤6：创建作业跟踪表Quantifind的数据扩展API是一个异步API，这意味着首先使用"submit"方法提交数据，然后使用"status"调用检查作业，当作业完成后，使用"results"调用检索结果。由于我们将为每个作业发出多个调用，因此需要在作业跟踪表中跟踪这些调用。这也可以作为所做操作的记录，也可以用于日志记录、计费、速率限制、呼叫上限或其他用途创建或替换表api_作业（customerid integer、app_name varchar、app_keyvarchar，jobid integer，totalsearches integer，状态标记varchar，submittedts时间戳_ltz（9），完成的时间戳_ltz（9））；ALTER TABLE api_jobs SET DATA_RETENTION_TIME_IN_DAYS=14更改_跟踪=TRUE；在表api上创建或替换流挂起的作业_jobs APPEND_ONLY=TRUE；上面的最后一行创建了一个雪花流，它将帮助我们稍后知道是否有任何作业处于挂起状态，这样，如果当前没有作业处于挂起状态，我们就不需要花费资源检查已完成的作业。步骤7：创建日志记录过程此步骤创建一个可用于记录其余步骤的过程。这对于开发存储过程非常有用，以便了解运行过程时发生的情况。创建或替换过程do\u log（MSG STRING）返回字符串语言JAVASCRIPT作为调用者执行作为$$ //查看是否应该对会话变量do\u log=true进行日志检查。试试看{变量stmt=雪花.createStatement（{sqlText:'选择$do\u log'}).execute（）；}捕获（错误）{return；//吞下错误，变量未设置，不记录    }    下一个stmt.next();如果(stmt.getColumnValue（1） ==true）{//如果值不是true，不记录试试看{           雪花.createStatement（{sqlText:'创建表标识符（$log\u table）如果不存在（ts number，msg string）}）.execute（）；           雪花.createStatement（{sqlText:'插入标识符（$log\u table）值（：1，：2），绑定：[日期。现在（），MSG]}）.execute（）；}捕获（错误）{抛出错误；       }    } $$;步骤8：创建处理已完成作业结果的过程下面的过程检查api_jobs表以获取所有挂起作业的列表，然后调用外部函数从远程api获取这些作业的状态。然后，它迭代结果，为每个完成的作业再次调用API，并将结果存储在特定于每个客户的结果表中。（在启用使用者后，将在后面的步骤中创建结果表。）创建或替换procedure quantified_process_results（master_db_name varchar，基本url VARCHAR）返回变量not null语言javascript作为调用者执行作为  $$  函数日志（msg）{        雪花.createStatement（{sqlText:'call do\u log（：1）"，绑定：[msg]}).execute（）；        }var my_sql_command="使用qs1 as（使用qs as（选择customerid、app_name，应用程序密钥、作业ID、状态标记来自"+MASTER_DB_NAME+"。public.api_作业其中状态标记='READY'order by jobid limit 10）选择customerid、jobid、app_name、app_key，quantifind_api（'状态？jobId='||作业ID,qs.app_名称, qs.app_密钥，'"+BASE_URL+"）作为value from qs）选择customerid、jobid、app_name、app_key、value[0