如何从 Azure Databricks 将 JSON 写入 Azure 队列答案

【问题标题】：How to write a JSON to Azure queue from Azure Databricks如何从 Azure Databricks 将 JSON 写入 Azure 队列
【发布时间】：2019-10-30 04:53:34
【问题描述】：

我正在尝试从 BLOB 读取 JSON 文件并将该文件写入 Azure 队列。读取部分工作正常，但在写入时会引发错误。

我已经尝试过要写入的队列文件夹的 URL，作为 .save() 的参数

这是我的代码：

storage_account_name="mrktmabcdestaaue"
storage_account_access_key="myurl=="
file_location="wasbs://myfolder@mrktmabcdestaaue.blob.core.windows.net/input.json"
file_type="json"
spark.conf.set(
        "fs.azure.account.key."+storage_account_name+".blob.core.windows.net",
  storage_account_access_key)
df = spark.read.option("multiline", "true").format(file_type).load(file_location)

df.write.mode("overwrite").format("com.databricks.spark.json").save("wasbs://myqueue@mrktmabcdestaaue.queue.core.windows.net")

我的输入 Json：

{
"Name": "Abc",
"Age": 18,
"City": "def"
}

我得到的错误信息是：

"shaded.databricks.org.apache.hadoop.fs.azure.AzureException: shaded.databricks.org.apache.hadoop.fs.azure.AzureException: 无法访问帐户 mrktmabcdestaaue.queue.core.windows 中的容器 myqueue .net 使用匿名凭据，但在配置中找不到它们的凭据。”

【问题讨论】：

标签： azure azure-servicebus-queues azure-databricks

【解决方案1】：

听起来您的集群未附加到该存储帐户。重新创建您的集群并确保该帐户已附加到集群。

您的存储位置应为wasbs://myfolder@mrktmabcdestaaue.blob.core.windows.net/input.json。

更多详情可以参考这个article。

【讨论】：

集群已附加。您的建议是针对 Azure BLOB，而不是针对 Azure 队列。

【解决方案2】：

不支持此方案。您可以写入 Blob 存储容器，但不能写入存储队列。

Databricks Azure 队列 (AQS) 连接器使用 Azure 队列存储 (AQS) 提供优化的文件源，让您可以查找写入 Azure Blob 存储 (ABS) 容器的新文件，而无需重复列出所有文件。有关详细信息，请参阅Documentation。所以也只能用于读取文件。

如果您想将阅读的内容传递给消费者，您可以使用 Azure EventHubs 或 Apache Kafka（在 Azure HDInsight 或 Confluent 上）作为消息代理。在这种情况下，您将使用结构化流。所以你必须有一个流数据框。写回流看起来像这样：

df \
.writeStream \
.format("eventhubs") \
.options(**ehConf) \
.option("checkpointLocation", checkploc) \
.start()

其他选项可能是使用 Azure 事件网格。最后，这取决于您想要实现的具体场景。

【讨论】：

【解决方案3】：

我有同样的场景，尝试了同样的事情，结果来到了这里。

我也尝试了dbutils.fs.mount，但无法像使用 blob 存储那样挂载存储队列。

我最终使用了存储队列客户端：
https://pypi.org/project/azure-storage-queue/

然后我必须阅读每个 json 消息并为每个消息调用 queue.send_message。不太好，但找不到更好的解决方案。

【讨论】：