如何统计Azure Data Lake Gen2存储容器内文件的行数

本地环境可通过如下Python代码统计数据文件行数：

with open('PPPLoanHoldStatus_AprilData.txt', 'r') as fp:
    for count, line in enumerate(fp):
        pass
print('Total Lines', count + 1)

需要实现同等功能，统计存储在Azure Data Lake Gen2下名为myContainer容器内目标文件的行数。
在Azure Databricks Notebook中直接传入ABFSS路径调用原生open()方法运行以下代码时触发报错：

with open('abfss://myContainer@myAzureDLGen2.dfs.core.windows.net/MyDataFile.txt', 'r') as fp:
    for count, line in enumerate(fp):
        pass
print('Total Lines', count + 1)

报错信息如下：

No such file or directory: 'abfss://myContainer@myAzureDLGen2.dfs.core.windows.net/MyDataFile.txt'

Python原生open()函数仅支持访问本地文件系统路径，无法识别ABFSS分布式存储协议，传入ADLS Gen2的ABFSS地址时会被识别为本地文件路径，因此触发文件不存在报错。

可按运行场景选择对应实现方式：

• 若在Azure Databricks环境运行，且已完成ADLS Gen2的访问权限配置（存储挂载、服务主体授权、工作区托管身份授权均可），可选择以下两种方式：

  1. 类原生open写法（小文件适用，逻辑和本地代码完全一致）
     直接使用Databricks内置的dbutils.fs文件工具访问ABFSS路径，不需要额外引入依赖：

  file_path = "abfss://myContainer@myAzureDLGen2.dfs.core.windows.net/MyDataFile.txt"
  count = 0
  with dbutils.fs.open(file_path, 'r') as fp:
      for count, line in enumerate(fp):
          pass
  print(f'Total Lines {count + 1}')
  

  注意：该方法会把文件拉取到Driver节点逐行遍历，仅适合百MB级以下小文件，大文件使用可能导致Driver节点内存溢出。

  2. Spark分布式统计（大文件适用，性能更高）
     利用Databricks内置的Spark引擎分布式读取文件统计行数，不需要在单节点逐行遍历，GB级以上大文件统计效率远高于逐行遍历写法：

  file_path = "abfss://myContainer@myAzureDLGen2.dfs.core.windows.net/MyDataFile.txt"
  line_count = spark.read.text(file_path).count()
  print(f'Total Lines {line_count}')
  

  若文件存在特殊编码、自定义行分隔符，可在read.text()方法中传入对应参数调整读取规则。

• 若在本地非Databricks环境运行，需要先安装Azure官方ADLS Gen2 SDK，配置存储访问凭证后再读取统计：

  1. 先安装依赖包：

  pip install azure-storage-file-datalake
  

  2. 代码实现：

  from azure.storage.filedatalake import DataLakeServiceClient
  
  # 替换为自己的存储账户地址、访问凭证
  service_client = DataLakeServiceClient(
      account_url="https://myAzureDLGen2.dfs.core.windows.net",
      credential="你的存储账户访问密钥/SAS令牌/服务主体凭证"
  )
  file_client = service_client.get_file_client(
      file_system="myContainer",
      file_path="MyDataFile.txt"
  )
  count = 0
  with file_client.download_file() as stream:
      for count, line in enumerate(stream.readall().split(b'\n')):
          pass
  print(f'Total Lines {count + 1}')
  

内容的提问来源于stack exchange，提问作者nam