循环遍历数据块中的文件失败

Question

继续Managing huge zip files in dataBricks

Databricks 在 30 个文件后挂起。怎么办？

我已将巨大的 32Gb zip 拆分为 100 个独立的部分。我已经从文件中拆分了标题，因此可以像处理任何 CSV 文件一样处理它。我需要根据列过滤数据。文件位于 Azure Data Lake Storage Gen1 中，必须存储在那里。

尝试一次读取单个文件（或所有 100 个文件）在工作约 30 分钟后失败。 （请参阅上面的链接问题。）

我做了什么：

def lookup_csv(CR_nro, hlo_lista =[], output = my_output_dir ): 

  base_lib = 'adl://azuredatalakestore.net/<address>'
  all_files = pd.DataFrame(dbutils.fs.ls(base_lib + f'CR{CR_nro}'), columns = ['full', 'name', 'size'])
  done = pd.DataFrame(dbutils.fs.ls(output), columns = ['full', 'name', 'size'])
  all_files = all_files[~all_files['name'].isin(tehdyt['name'].str.replace('/', ''))]
  all_files = all_files[~all_files['name'].str.contains('header')]

  my_scema = spark.read.csv(base_lib + f'CR{CR_nro}/header.csv', sep='\t', header=True, maxColumns = 1000000).schema
  tmp_lst = ['CHROM', 'POS', 'ID', 'REF', 'ALT', 'QUAL', 'FILTER', 'INFO', 'FORMAT'] + [i for i in hlo_lista  if i in my_scema.fieldNames()]

  for my_file in all_files.iterrows(): 
    print(my_file[1]['name'], time.ctime(time.time()))
    data = spark.read.option('comment', '#').option('maxColumns', 1000000).schema(my_scema).csv(my_file[1]['full'], sep='\t').select(tmp_lst)
    data.write.csv( output + my_file[1]['name'], header=True, sep='\t')

这行得通……有点。它的工作原理是大约 30 个文件，然后挂断

Py4JJavaError：调用 o70690.csv 时出错。​​​ 原因：org.apache.spark.SparkException：作业因阶段失败而中止：阶段 154.0 中的任务 0 失败 4 次，最近一次失败：阶段 154.0 中丢失任务 0.3（TID 1435、10.11.64.46、执行程序 7）：com .microsoft.azure.datalake.store.ADLException：创建文件 CR03_pt29.vcf.gz/_started_1438828951154916601 时出错 操作 CREATE 因 HTTP401 失败：null 2 次尝试后最后遇到的异常抛出。 [HTTP401(null),HTTP401(null)]

我尝试添加一些删除并休眠：

   ​data.unpersist()
   ​data = []
   ​time.sleep(5)

还有一些 try-exception 尝试。

for j in range(1,24): 
    for i in range(4): 
        try: 
            lookup_csv(j, hlo_lista =FN_list, output = blake +f'<my_output>/CR{j}/' )
        except Exception as e:
            print(i, j, e)
            time.sleep(60)

这些都不走运。一旦失败，就会一直失败。

知道如何处理这个问题吗？我认为与 ADL 驱动器的连接会在一段时间后失败，但如果我将命令排队：

lookup_csv(<inputs>) 
<next cell> 
lookup_csv(<inputs>) 

它可以正常工作，失败并且可以在下一个单元格中正常工作。我可以忍受这一点，但非常烦人的是基本循环无法在这种环境中工作。

Answer 1

最好的解决方案是永久安装 ADSL 存储并为此使用 azure 应用程序。

在 Azure 中，请转到应用注册 - 使用名称注册应用，例如“databricks_mount”。在 delta Lake 存储中为该应用添加 IAM 角色“Storage Blob Data Contributor”。

configs = {"fs.azure.account.auth.type": "OAuth",
      "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider", 
      "fs.azure.account.oauth2.client.id": "<your-client-id>",
      "fs.azure.account.oauth2.client.secret": "<your-secret>",
      "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<your-endpoint>/oauth2/token"}

dbutils.fs.mount(
 source = "abfss://delta@yourdatalake.dfs.core.windows.net/",
 mount_point = "/mnt/delta",
 extra_configs = configs)

您无需挂载即可访问，但仍需要注册应用并通过笔记本中的 spark 设置应用配置才能访问 ADLS。由于 azure 应用程序，它应该在整个会话中是永久性的：

spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider"), 
spark.conf.set("fs.azure.account.oauth2.client.id", "<your-client-id>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<your-secret>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<your-endpoint>/oauth2/token")

这个解释是最好的https://docs.databricks.com/data/data-sources/azure/adls-gen2/azure-datalake-gen2-sp-access.html#access-adls-gen2-directly 虽然我记得第一次我也有这个问题。该页面还解释了如何注册应用程序。也许这符合您的公司政策。