python - 在 pyspark 中使用 S3a 保存文件时获取 S3 响应代码（仅限 HTTP 代码，如 200,300,400,403,500 等）

Question

我正在尝试获取 HTTP 代码并将其存储在 RDS 表中，以便以后分析 pyspark 作业，该作业将使用 S3a 将文件以 AVRO 格式保存到 S3。保存文件后，我知道 S3 将返回状态代码，但我不确定如何在代码中记录它。请找到代码片段。

def s3_load(df, row):
    df.write.\
       format("com.databricks.spark.avro").\
       save("s3a://Test-" + row["PARTNER"].lower() + "/" + row["TABLE_NAME"] + "/" +
            datetime.datetime.today().strftime('%Y%m%d'))

在上面的代码中，我希望 o 以状态码的形式返回。注意：我可以将 S3 中的文件保存为 AVRO 格式。

谢谢

标签： pythonamazon-web-servicesamazon-s3pysparkapache-spark-sql

Answer 1

这是此问题中讨论的类似概念，获取包装 s3 API 的库或函数的状态代码：Amazon S3 POST，完成时的事件？

---

最终，如果 databricks 是处理上传的库，则函数调用的结果响应代码df.write.save(...)将在 databricks 函数调用的结果中的某处找到。

Databricks 支持s3并s3a作为保存文件的目标目的地（如他们的文档所示），但似乎 databricks 并没有显示来自底层操作的响应代码（也许他们这样做，我在任何文档中都找不到它） ）。

前进的一些选择：

• 假设 databricks 会为该上传引发“某种”错误，一个简单的try/except将允许您正确捕获此错误（尽管任何非 databricks 级别的错误仍然会通过）。

• 在 AWS 上，s3 存储桶上传是一个事件源，可用作其他操作的触发器，例如调用 AWS Lambda，您可以使用它来调用任意云托管函数。本教程中提供了大量有关此架构的信息。

• 根据并行上传的需要，您可以使用boto3官方 AWS python 库重写您的小型上传函数。讨论如何处理此处讨论的那些错误/响应代码。

• Databricks 似乎在其企业产品中的某处也具有审计日志记录功能。