继续上一篇的内容,本文将为大家介绍将 ChangeFeed 抽取逻辑通过 Azure Function 服务实现,架构图如下,通过 Function 服务来执行 ChangeFeed 的读取,并将其转存至 DataLake 存储中。
在前文后我们介绍了通过 Function 服务可以简化整个逻辑的代码开发,Azure Function 服务中原生已经内置了很多与 Azure 其他服务原生集成的连接器,可以帮助客户实现与上下游服务的对接,用户无需关注连接器的实现,通过框架的调用,用户可以直接可以通过对象访问到上下游服务中的数据,用户只需要关注业务逻辑即可。Cosmos Database 也在 Azure Function 的支持之中,并且内置的连接器采用的也是 ChangeFeed 来实现的,用户可以直接开箱即用实现 ChangeFeed 数据的读取,无需自己维护抽取逻辑代码。Azure Function 原生支持的连接器如下:
Type | 1.x | 2.x and higher1 | Trigger | Input | Output |
---|---|---|---|---|---|
Blob storage | ✔ | ✔ | ✔ | ✔ | ✔ |
Cosmos DB | ✔ | ✔ | ✔ | ✔ | ✔ |
Event Grid | ✔ | ✔ | ✔ | ✔ | |
Event Hubs | ✔ | ✔ | ✔ | ✔ | |
HTTP & webhooks | ✔ | ✔ | ✔ | ✔ | |
IoT Hub | ✔ | ✔ | ✔ | ✔ | |
Microsoft Graph Excel tables | ✔ | ✔ | ✔ | ||
Microsoft Graph OneDrive files | ✔ | ✔ | ✔ | ||
Microsoft Graph Outlook email | ✔ | ✔ | |||
Microsoft Graph events | ✔ | ✔ | ✔ | ✔ | |
Microsoft Graph Auth tokens | ✔ | ✔ | |||
Mobile Apps | ✔ | ✔ | ✔ | ||
Notification Hubs | ✔ | ✔ | |||
Queue storage | ✔ | ✔ | ✔ | ✔ | |
SendGrid | ✔ | ✔ | ✔ | ||
Service Bus | ✔ | ✔ | ✔ | ✔ | |
SignalR | ✔ | ✔ | ✔ | ||
Table storage | ✔ | ✔ | ✔ | ✔ | |
Timer | ✔ | ✔ | ✔ | ||
Twilio | ✔ | ✔ | ✔ |
下面我们来开始操作:
1. 准备开发环境,建议使用 Visual Studio Code,其内置的 Azure Function 的开发扩展,可以方便开发
参考:https://docs.microsoft.com/en-us/azure/developer/python/tutorial-vs-code-serverless-python-01
2. 创建 Azure Function 项目,注意在选择 trigger 部分,选择 cosmos db
参考:https://docs.microsoft.com/en-us/azure/developer/python/tutorial-vs-code-serverless-python-02
3. 准备 function.json 配置文件,function.json 主要描述 function 与上下游数据连接的参数,其中下述配置中 type:cosmosDBTrigger 部分定义了 cosmos 的连接信息, databaseName,collectionName 替换为前面所创建的 cosmos db的名称,leaseCollectionName 是 function 用来维护租约和 CheckPoint。connectionStringSetting 参见后续 local.settings.json。type:blob 部分定义了 function 下游存储 Data Lake 的连接信息,其中 path 参数定义了 function 抽取增量变化数据在 Data Lake 中的存储路径,connection 参数参见后续 local.settings.json。另外 feedPollDelay 参数表示 function 服务轮询 ChangeFeed 数据的间隔,其单位为毫秒,在演示中建议大家可以设置为 60000,实际根据数据水线处理时间和数据更新需求来决定。
{ "scriptFile": "__init__.py", "bindings": [ { "type": "cosmosDBTrigger", "name": "documents", "direction": "in", "leaseCollectionName": "<leasecollectionname>", "connectionStringSetting": "Cosmos_DOCUMENTDB", "databaseName": "<databasename>", "collectionName": "<collectionname>", "createLeaseCollectionIfNotExists": "true", "feedPollDelay": <pollinterval> }, { "name": "outputblob", "type": "blob", "path": "<filesystemname>/{DateTime}.csv", "connection": "ChangeFeedResultStorage", "direction": "out" } ] }
4. 准备 local.settings.json,该配置文件中定义了在上述 function.json 中所引用的连接密钥参数, 其中 Cosmos_DOCUMENTDB 和 ChangeFeedResultStorage 内分别填入,Cosmos 和 Data Lake 的连接字符串,可以在 portal 中对应资源的 access 信息部分获取。
{ "IsEncrypted": false, "Values": { "AzureWebJobsStorage": "<connectionstring>", "FUNCTIONS_WORKER_RUNTIME": "python", "Cosmos_DOCUMENTDB": "<cosmosconnectionstring>", "ChangeFeedResultStorage": "<datalakeconnectionstring>" } }
5. 准备业务逻辑代码 init.py,init.py 是 function 函数被拉起后的 entrypoint 函数,我们将前面介绍的抽取 changefeed 增量变化数据的代码逻辑定义其中,下述演示代码中通过调用 documents 和 outputblob 借助 function 内置的连接器实现对上下游数据访问,无需再自己开发集成代码。用户只需要开发自己的数据处理逻辑即可,演示中是将增量变化数据转存到 Data Lake 存储中。
import logging import json import csv import io import azure.functions as func # read changefeed from cosmos, store as CSV file def main(documents: func.DocumentList, outputblob: func.Out[str]) : if documents: count = 0 outputresult = io.StringIO() csv_writer = csv.writer(outputresult, quoting=csv.QUOTE_NONNUMERIC) for document in documents: if count == 0: header = json.loads(document.to_json()).keys() csv_writer.writerow(header) count += 1 csv_writer.writerow(json.loads(document.to_json()).values()) outputblob.set(outputresult.getvalue())
6. 通过 VS Code Function 本地调试工具进行测试,可以仿真通过前面 blog 中的数据插入函数在 cosmos db 内插入一些新的数据,然后在 Data Lake 中是否转存成功。
7. 将 Function 服务打包发布至 Azure Function 服务中,上述开发测试均在本地完成,测试无误后将代码正式发布至 Azure Function 服务
参考:https://docs.microsoft.com/en-us/azure/developer/python/tutorial-vs-code-serverless-python-05
至此通过 Function 服务完成 ChangeFeed 读取及转存至 DataLake 已经完成。整个 Function 代码中 function.json 中针对不同连接器的参数说明大家可以参阅:https://docs.microsoft.com/en-us/azure/azure-functions/functions-bindings-cosmosdb?tabs=csharp