Как написать JSON в очередь Azure из блоков данных Azure

Question

Я пытаюсь прочитать файл JSON из BLOB и записать этот файл в очередь 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: невозможно получить доступ к myqueue контейнера в учетной записи mrktmabcdestaaue.queue.core.windows.net с использованием анонимных учетных данных, и учетные данные не найденыдля них в конфигурации. "

Answer 1

Этот сценарий не поддерживается. Вы можете писать в контейнер хранения BLOB-объектов, но не в очереди хранения.

Коннектор Databricks Azure Queue (AQS) использует хранилище очередей Azure (AQS) для обеспечения оптимизированного источника файлов, который позволяет находить новые файлы, записанные в контейнер хранилища BLOB-объектов Azure, без повторного перечисления всех файлов. ,См. Документация для более подробной информации. Также можно использовать только чтение файлов.

Если вы хотите передать прочитанное содержимое потребителям, вы можете использовать Azure EventHubs или Apache Kafka (в Azure HDInsight или Confluent) в качестве посредника сообщений. В этом сценарии вы бы использовали структурированную потоковую передачу. Таким образом, у вас должен быть фрейм потоковых данных. Запись обратного потока будет выглядеть следующим образом:

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

В качестве опции можно использовать Azure Event Grid. Наконец, это зависит от конкретного сценария, которого вы хотели бы достичь.

Answer 2

Похоже, ваш кластер не привязан к этой учетной записи хранения. Создайте заново кластер и убедитесь, что учетная запись подключена к кластеру.

Ваше хранилище должно быть wasbs://myfolder@mrktmabcdestaaue.blob.core.windows.net/input.json.

Для получения более подробной информации вы можете обратиться к этой статье .