Azure Databricks 上的最大消息大小

Posted 2023-04-15

【中文标题】Azure Databricks 上的最大消息大小

【英文标题】：Maximum message size on Azure Databricks

【发布时间】：2020-02-06 01:45:12

【问题描述】：

我在 Azure 上使用带有 python 的 databricks 来处理我的数据。此过程的结果将保存为 azure blob 存储上的 csv 文件。

但这就是问题所在。当结果文件超过 750 Mb 时发生错误。

在对 google 进行了一些研究之后，我知道我必须增加我的 Scala.rc.message.maxSize，我就这样做了。问题是我可以设置的最大大小只有 2Gb，而且当我使用 databricks 分析大数据时，我确实希望文件远大于 2Gb。

问题是：

2 Gb 真的是 Azure Databricks 支持的最大消息大小吗？我试图搜索并浏览微软的官方文档，但找不到任何相关信息。

有什么方法可以增加价值吗？甚至将其设置为可扩展取决于我的数据。

这是我用于这些过程的 python 代码。

#mount azure storage to my databricks
dbutils.fs.mount(
  source = "wasbs://mystoragecontainer.blob.core.windows.net",
  mount_point = "/mnt/test3",
  extra_configs = "fs.azure.account.key.mystoragecontainer.blob.core.windows.net":dbutils.secrets.get(scope = "myapps", key = "myappskey"))


#define saving process in a function
def save_data(df, savefile):
  df.coalesce(1).write.mode("overwrite").options(header="true").format("com.databricks.spark.csv").save(savefile)
  res = savefile.split('/')
  ls_target = savefile.rstrip(res[-1])
  dbutils.fs.ls(savefile+"/")
  fileList = dbutils.fs.ls(savefile+"/")
  target_name = ""
  for item in fileList:
    if item.name.endswith("csv"):
      filename= item.path
      target_parts = filename.split('/')
      target_name = filename.replace('/'+target_parts[-2]+'/', '/')
      print(target_name)
      dbutils.fs.mv(filename, ls_target)
    else:
      filename= item.path
      dbutils.fs.rm(filename, True)
  dbutils.fs.rm(savefile, True)
  dbutils.fs.mv(target_name, savefile)

# call my save function
save_data(df,"dbfs:/mnt/test3/myfolderpath/japanese2.csv")

任何信息将不胜感激。

最好的，

【问题讨论】：

到目前为止你有什么尝试？ Databricks Runtime 5.5 及以下版本的 dbfs 存在限制，仅支持小于 2GB 的文件。如果您使用本地文件 I/O API 读取或写入大于 2GB 的文件，您可能会看到损坏的文件。话虽如此，您应该能够在 Python 中将更大的文件写入已挂载的存储。你能分享一些你试过的代码吗？

嗨@AxelR。谢谢您的答复。我已经安装了我的天蓝色存储，但同样的问题仍然存在。将我的代码添加到我的第一篇文章中，请查看以供参考。

【参考方案1】：

如果我理解正确，您想合并由 :

生成的分布式 csv

df.coalesce(1).write.mode("overwrite").options(header="true").format("com.databricks.spark.csv").save(savefile) 

我建议您尝试将其转换为 pandas 数据帧并写入单个 csv，如下所示：

# call my save function
df.toPandas().to_csv("/dbfs/mnt/test3/myfolderpath/japanese2.csv")

这应该编写一个包含数据框中所有数据的 csv。 在使用 Pandas 时要小心使用/dbfs/，因为它使用文件 API 而不是 DBFS API。

另外，这是 pySpark，不是真正的 scala。