python - pyspark 和 HDFS 命令

Question

我想在我的 Spark 程序 (Pyspark) 开始时进行一些清理。例如，我想从以前的 HDFS 运行中删除数据。在 pig 中，这可以使用诸如

之类的命令来完成

fs -copyFromLocal ....

rmf /path/to-/hdfs

或在本地使用 sh 命令。

我想知道如何用 Pyspark 做同样的事情。

Answer 1

您可以删除 PySpark 中的 hdfs 路径，而无需使用第三方依赖项，如下所示:

from pyspark.sql import SparkSession
# example of preparing a spark session
spark = SparkSession.builder.appName('abc').getOrCreate()
sc = spark.sparkContext
# Prepare a FileSystem manager
fs = (sc._jvm.org
      .apache.hadoop
      .fs.FileSystem
      .get(sc._jsc.hadoopConfiguration())
      )
path = "Your/hdfs/path"
# use the FileSystem manager to remove the path
fs.delete(sc._jvm.org.apache.hadoop.fs.Path(path), True)

---

为了进一步改进，您可以将上述想法包装到一个辅助函数中，您可以跨作业/包重复使用该函数:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('abc').getOrCreate()

def delete_path(spark, path):
    sc = spark.sparkContext
    fs = (sc._jvm.org
          .apache.hadoop
          .fs.FileSystem
          .get(sc._jsc.hadoopConfiguration())
          )
    fs.delete(sc._jvm.org.apache.hadoop.fs.Path(path), True)

delete_path(spark, "Your/hdfs/path")