代码之家 › 专栏 › 技术社区 › Idiot

Spark-处理后移动文件

delta-lake databricks azure-blob-storage pyspark apache-spark

Idiot · 技术社区 · 2 年前

我将json文件读取到数据帧中: df = spark.read.option("multiline", "true").json(f"/mnt/bronze/{something}*")

然后我进行一些处理,然后将其写入delta表: rows.write.format("delta").mode("overwrite").save(f"/mnt/silver/{something}")

我想在处理后移动json文件,这样以后就不会再重新处理它们了。但我不知道如何移动它们。

我尝试在rdd.foreach()中使用dbutils,但这是不可能的。我还在rdd.foreach()中尝试了shutil.move(),但由于我使用blob存储,所以找不到我的文件。

1 回复 | 直到 2 年前

Daniel Perez Efremova 2 年前

让我提供一些解决方案。

我想/mnt/b古铜/{something}是您着陆文件的暂存区,并且您希望防止读取以前着陆的文件。

您在当前笔记本的末尾有一些选项:

dbutils.fs.rm("/mnt/bronze/{something}*", True)

dbutils.fs.rm 是删除文件的功能, "/mnt/bronze/{something}*" 是要删除的文件的路径和模式,第二个参数为True表示删除是递归的(如果您确定只删除文件,这是可选的)。

dbutils.fs.mv("/mnt/bronze/{something}*", "/mnt/tmp/")

PD:如果你选择2),我建议在tmp/dir的路径中添加一些基于时间戳的规则,以确保它保持有序和干净。例如 f"mnt/tmp/{today()}"

推荐文章

Danylo Kuznetsov · 如何在PySpark Rancher中将DataFrame转换为整数?

1 年前

JFlo · 在PySpark笔记本中读取多个Parquet文件

1 年前

Enrique Benito Casado · Databricks Pyspark写入增量格式模式覆盖无法正常工作

1 年前

Patterson · 无法从Visual Studio代码创建用于Databricks的Python控制盘

1 年前

Aaron · Databricks SQL中行到列的动态转换

1 年前

Ajay S Pal · 当调用函数时传递参数时,PySpark没有在函数内部创建Dataframe

1 年前

Patterson · 将T-SQL重构为Databricks SQL以从子查询返回结果

1 年前

Desert Spider · 使用多个分隔符拆分sql

2 年前

Idiot · Spark-处理后移动文件

2 年前

Sarath Subramanian · 处理来自cat命令的错误消息

2 年前