0

我正在使用 ADP 的 API 构建集成。已经有一个我将使用的端点 - 这与一个 azure 服务总线相关联。我知道该主题的名称，但我无法理解需要在 Azure 中创建的下一个工件。我希望传入的消息到达 EDW 或只是数据湖作为数据（它可能以 XML 格式出现，我可能需要将其转换为 azure sql 数据库）：

  • 我是否应该创建另一个与 EDW 绑定的订阅，以便从服务总线主题中获取消息？（我以前没有尝试过）
  • 或者我应该创建一个逻辑应用程序来直接读取服务总线或服务总线主题吗？（我也没有尝试过）

我需要设计一个可扩展的解决方案 - 任何见解将不胜感激

谢谢

4

1 回答 1

0

您可以扩展此答案还是有文档可以查看如何将服务总线主题链接到 azure 函数然后转储到数据湖中？

我不确定你使用什么语言，这里是用python实现的。

您可以使用服务总线触发器来侦听来自服务总线队列或主题的消息。然后您可以使用数据湖 SDK 来保存消息：

使用 azure function service bus trigger 监听消息：

import logging

import azure.functions as func


def main(msg: func.ServiceBusMessage):

    #put the logic of process the message here

    logging.info('Python ServiceBus queue trigger processed message: %s',
                 msg.get_body().decode('utf-8'))

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msg",
      "type": "serviceBusTrigger",
      "direction": "in",
      "queueName": "queuename",
      "connection": "bowman1012_SERVICEBUS"
    }
  ]
}

并使用如下代码将消息附加到数据湖：

from azure.storage.filedatalake import DataLakeServiceClient 
connect_str = "DefaultEndpointsProtocol=https;AccountName=0730bowmanwindow;AccountKey=xxxxxx;EndpointSuffix=core.windows.net"
datalake_service_client = DataLakeServiceClient.from_connection_string(connect_str)
myfilesystem = "test"
myfolder     = "test"
myfile       = "FileName.txt"

file_system_client = datalake_service_client.get_file_system_client(myfilesystem)            
directory_client = file_system_client.create_directory(myfolder)         
directory_client = file_system_client.get_directory_client(myfolder)
print("11111")
try:
    file_client = directory_client.get_file_client(myfile)
    file_client.get_file_properties().size
    data = "Test2"   
    print("length of data is "+str(len(data)))
    print("This is a test123")
    filesize_previous = file_client.get_file_properties().size
    print("length of currentfile is "+str(filesize_previous))
    file_client.append_data(data, offset=filesize_previous, length=len(data))
    file_client.flush_data(filesize_previous+len(data))
except:
    file_client = directory_client.create_file(myfile)
    data = "Test2"   
    print("length of data is "+str(len(data)))
    print("This is a test")
    filesize_previous = 0
    print("length of currentfile is "+str(filesize_previous))
    file_client.append_data(data, offset=filesize_previous, length=len(data))
    file_client.flush_data(filesize_previous+len(data))

如果你需要在本地开发azure function，你需要'azure function core tools'，'language environment'，'VS Code and azure function extension'。

有关更多信息，请查看以下内容：

https://docs.microsoft.com/en-us/azure/azure-functions/functions-run-local?tabs=windows%2Ccsharp%2Cbash

https://docs.microsoft.com/en-us/azure/azure-functions/functions-reference-python

https://docs.microsoft.com/en-us/azure/azure-functions/functions-bindings-service-bus-trigger?tabs=python

这是数据湖SDK的API参考（在这个网页中，您可以找到与基于python和azure的各种服务交互的所有方法）：

https://docs.microsoft.com/en-us/python/api/azure-storage-file-datalake/azure.storage.filedatalake.datalakeserviceclient?view=azure-python

于 2021-02-09T01:32:47.167 回答