尝试使用 Apache Spark Databricks 将消息从 Azure 服务总线加载到 ADLS

Question

我正在尝试使用 Databricks 将 Azure 服务总线队列中的消息移动到 ADLS。

我使用的代码如下：

from azure.servicebus import ServiceBusClient, ServiceBusMessage

CONNECTION_STR = "Endpoint=sb://carlsbus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=2BIOtFdooJFIkhZF0qxtYEoK32hJa39DseStw="

QUEUE_NAME = "myqueue"

servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR, logging_enable=True)

receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME, max_wait_time=5)

当我执行以下命令时：

mystream2 = spark.readStream.format(receiver).load("/mnt/lake/RAW/FormulaClassification/F1Area/F1Domain/flightStream/1")

但我一直收到错误消息：

'ServiceBusReceivedMessage' object has no attribute '_get_object_id'

知道我为什么会收到错误消息吗？

Answer 1

Error - 'ServiceBusReceivedMessage' object has no attribute '_get_object_id'

当您尝试在函数内部使用数据框时会出现上述错误。在您的情况下，您在 format() 函数中使用了 receiver。

您不能在函数内部使用数据框。

更多信息请关注 SO thread and this article

尝试使用 Apache Spark Databricks 将消息从 Azure 服务总线加载到 ADLS

Trying to load messages from Azure Service Bus to ADLS with Apache Spark Databricks

azureservicebus

apache-spark

azure-databricks