Azure Databricks - 无法将结果从数据块导出到 blob

Question

我想将数据从数据砖导出到 Azure Blob。我的数据砖命令从我的 blob 中选择一些 pdf，运行表单识别器并将输出结果导出到我的 blob 中。

这是我的代码：

    %pip install azure.storage.blob
    %pip install azure.ai.formrecognizer
    
  
    from azure.storage.blob import ContainerClient
    
    container_url = "https://mystorageaccount.blob.core.windows.net/pdf-raw"
    container = ContainerClient.from_container_url(container_url)
    
    for blob in container.list_blobs():
    blob_url = container_url + "/" + blob.name
    print(blob_url)


import requests
from azure.ai.formrecognizer import FormRecognizerClient
from azure.core.credentials import AzureKeyCredential

endpoint = "https://myendpoint.cognitiveservices.azure.com/"
key = "mykeynumber"

form_recognizer_client = FormRecognizerClient(endpoint, credential=AzureKeyCredential(key))

   
    import pandas as pd
    
    field_list = ["InvoiceDate","InvoiceID","Items","VendorName"]
    df = pd.DataFrame(columns=field_list)
    
    for blob in container.list_blobs():
        blob_url = container_url + "/" + blob.name
        poller = form_recognizer_client.begin_recognize_invoices_from_url(invoice_url=blob_url)
        invoices = poller.result()
        print("Scanning " + blob.name + "...")
    
        for idx, invoice in enumerate(invoices):
            single_df = pd.DataFrame(columns=field_list)
            
            for field in field_list:
                entry = invoice.fields.get(field)
                
                if entry:
                    single_df[field] = [entry.value]
                    
                single_df['FileName'] = blob.name
                df = df.append(single_df)
                
    df = df.reset_index(drop=True)
    df
    

    account_name = "mystorageaccount"
    account_key = "fs.azure.account.key." + account_name + ".blob.core.windows.net"
    
    try:
        dbutils.fs.mount(
            source = "wasbs://[e-mail]",
            mount_point = "/mnt/pdf-recognized",
            extra_configs = {account_key: dbutils.secrets.get(scope ="formrec", key="formreckey")} )
        
    except:
        print('Directory already mounted or error')
    
    df.to_csv(r"/dbfs/mnt/pdf-recognized/output.csv", index=False)

代码运行良好，直到最后一行。我收到以下错误消息：＜code＞目录已装入或出错。FileNotFoundError:[Errno 2]没有这样的文件或目录：'/dbfs/mnt/pdf已识别/output.csv'。

我尝试使用 /dbfs：/而不是 /dbfs/，但我不知道我做错了什么。

如何将我的数据砖结果导出到 blob？

谢谢

Answer 1

看起来您正在尝试装载已装载的存储。实际上，装载操作应该只执行一次，而不是动态执行。您有几种选择来正确实现它：

>

在使用数据库残缺不全之前卸载（“/mnt/pdf 识别”）

检查是否已装载存储

mounts = [mount for mount in dbutils.fs.mounts() 
      if mount.mountPoint == "/mnt/pdf-recognized"]
if len(mounts) == 0:
  dbutils.fs.mount(....)

• 您并不真正需要挂载-它具有“坏”属性，工作区中的任何人都可以使用它来拥有用于挂载的权限。将结果写入本地磁盘，然后使用dbutils.fs将文件复制到必要的位置可能更简单。cpwithwasb协议。类似这样：

df.to_csv(r"/tmp/my-output.csv", index=False)
spark.conf.set(account_key, dbutils.secrets.get(scope ="formrec", key="formreckey"))
dbutils.fs.cp("file:///tmp/my-output.csv"), 
   "wasbs://[e-mail]/output.csv")