从 PySpark 中的 s3 子目录读取数据

2024-02-18

我想从 S3 存储桶中读取所有 parquet 文件,包括子目录中的所有文件(这些实际上是前缀)。

在 S3 URL 中使用通配符 (*) 仅适用于指定文件夹中的文件。例如,使用此代码将仅读取下面的镶木地板文件target/ folder.

df = spark.read.parquet("s3://bucket/target/*.parquet")
df.show()

假设我的 s3 存储桶中有这样的结构:

"s3://bucket/target/2020/01/01/some-file.parquet"
"s3://bucket/target/2020/01/02/some-file.parquet"

上面的代码会引发异常:

pyspark.sql.utils.AnalysisException: 'Path does not exist: s3://mailswitch-extract-underwr-prod/target/*.parquet;'

如何从 s3 存储桶的子目录中读取所有 parquet 文件?

为了运行我的代码,我将 AWS Glue 2.0 与 Spark 2.4 和 python 3 结合使用。


如果您想读取目标文件夹下的所有 parquet 文件

"s3://bucket/target/2020/01/01/some-file.parquet"
"s3://bucket/target/2020/01/02/some-file.parquet"

你可以做

df = spark.read.parquet("bucket/target/*/*/*/*.parquet")

缺点是您需要知道镶木地板文件的深度。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

从 PySpark 中的 s3 子目录读取数据 的相关文章

  • Spark:shuffle操作导致GC长时间暂停

    我在跑Spark 2我正在尝试洗牌大约 5 TB 的 json 我在洗牌期间遇到了很长的垃圾收集暂停Dataset val operations spark read json inPath as MyClass operations re
  • pyspark中的稀疏向量RDD

    我一直在使用 mllib 的功能通过 Python Pyspark 实现此处描述的 TF IDF 方法 https spark apache org docs 1 3 0 mllib feature extraction html http
  • scala.collection.Seq 不适用于 Java

    Using 阿帕奇火花2 0 1 Java 7 在 Apache Spark Java API 文档中 DataSet 类出现了一个example http spark apache org docs latest api java org
  • 如何使用 Spark 执行插入覆盖?

    我正在尝试将我们的 ETL Hive 脚本之一转换为 Spark 其中 Hive ETL 脚本维护一个表 其中需要在每晚新同步之前删除部分数据 Hive ETL 使用插入覆盖的方式将主表删除超过 3 天的数据 基本上创建一个临时表 其中的数
  • 将案例类传递给函数参数

    抱歉问了一个简单的问题 我想将案例类传递给函数参数 并且想在函数内部进一步使用它 到目前为止我已经尝试过这个TypeTag and ClassTag但由于某种原因 我无法正确使用它 或者可能是我没有看到正确的位置 用例与此类似 case c
  • 如何使用 Apache Livy 设置 Spark 配置属性?

    我不知道在向 Apache Livy 提交 Spark 作业时如何以编程方式传递 SparkSession 参数 这是测试 Spark 作业 class Test extends Job Int override def call jc J
  • Spark 中的广播 Annoy 对象(对于最近邻居)?

    由于 Spark 的 mllib 没有最近邻居功能 我正在尝试使用Annoy https github com spotify annoy为近似最近邻 我尝试广播 Annoy 对象并将其传递给工人 然而 它并没有按预期运行 下面是可重复性的
  • Spark 按列重新分区,每列动态分区数

    如何根据列中的项目数对 DataFrame 进行分区 假设我们有一个包含 100 人的 DataFrame 列是first name and country 我们希望为一个国家 地区的每 10 个人创建一个分区 如果我们的数据集包含 80
  • Spark SQL / PySpark 中的逆透视

    我手头有一个问题陈述 其中我想在 Spark SQL PySpark 中取消透视表 我已经浏览了文档 我可以看到仅支持pivot 但到目前为止还不支持取消透视 有什么方法可以实现这个目标吗 让我的初始表如下所示 When I pivotPy
  • Python Spark DataFrame:用 SparseVector 替换 null

    在 Spark 中 我有以下名为 df 的数据框 其中包含一些空条目 id features1 features2 185 5 0 1 4 0 1 0 null 220 5 0 2 3 0 1 0 10 1 2 6 0 1 225 null
  • Spark 执行器 STDOUT 到 Kubernetes STDOUT

    我在 Spark Worker 中运行的 Spark 应用程序将执行程序日志输出到特定文件路径 worker home directory app xxxxxxxx 0 stdout I used log4j properties将日志从
  • 从 pandas udf 记录

    我正在尝试从 python 转换中调用的 pandas udf 进行日志记录 因为在执行器上调用的代码不会显示在驱动程序的日志中 我一直在寻找一些选项 但到目前为止最接近的选项是这个one https stackoverflow com q
  • 在 kubernetes 上安装 PySpark 软件包时出现 Spark-Submit:ivy-cache 文件未找到错误

    我一整天都在与它斗争 我能够安装并使用带有 Spark shell 或连接的 Jupiter 笔记本的包 graphframes 但我想使用 Spark Submit 将其移动到基于 kubernetes 的 Spark 环境 我的火花版
  • 使用 mlib 执行 Spark-Shell,错误:对象 jblas 不是包 org 的成员

    在spark shell中 当我执行import org jblas DoubleMatrix 它会在 RHEL 上抛出 错误 对象 jblas 不是包 org 的成员 实际上 我用谷歌搜索了 jblas 并安装了 gfortran htt
  • 如何从 Databricks Delta 表中删除列?

    我最近开始发现 Databricks 并遇到了需要删除增量表的特定列的情况 当我使用 PostgreSQL 时 它就像 ALTER TABLE main metrics table DROP COLUMN metric 1 我正在浏览 Da
  • pyspark:将多个数据帧字段传递给 udf

    我是 Spark 和 Python 的新手 任何帮助表示赞赏 我有一个 UDF 并使用 US zipcd 纬度和经度创建了一个 Spark 数据框 UDF import math def distance origin destinatio
  • Spark:并行转换多个数据帧

    了解如何在并行转换多个数据帧时实现最佳并行性 我有一系列路径 val paths Array path1 path2 我从每个路径加载数据帧 然后转换并写入目标路径 paths foreach path gt val df spark re
  • 如何通过sparkSession向worker提交多个jar?

    我使用的是火花2 2 0 下面是我在 Spark 上使用的 java 代码片段 SparkSession spark SparkSession builder appName MySQL Connection master spark ip
  • 使用 pyspark 计算所有可能的单词对

    我有一个文本文档 我需要找到整个文档中重复单词对的可能数量 例如 我有下面的word文档 该文档有两行 每行用 分隔 文档 My name is Sam My name is Sam My name is Sam My name is Sa
  • Spark (Python) 中的 Kolmogorov Smirnov 测试不起作用?

    我正在 Python Spark ml 中进行正态性测试 看到了我的结果think是一个错误 这是设置 我有一个标准化的数据集 范围 1 到 1 当我做直方图时 我可以清楚地看到数据不正常 gt gt gt prices norm hist

随机推荐