Как проверить наличие файла/папки с помощью pyspark без исключения

Похоже, вы должны изменить except IOError: на except AnalysisException:.

Spark во многих случаях выдает другие ошибки/исключения, чем обычный python. Он не выполняет типичные операции ввода-вывода Python при чтении файла, поэтому имеет смысл генерировать другое исключение.

приятно видеть вас на StackOverFlow.

Я второе решение dijksterhuis, за одним исключением - Исключение анализа — это очень распространенное исключение в Spark, которое может возникать по разным причинам, а не только из-за отсутствия файла.

Если вы хотите проверить, существует ли файл или нет, вам нужно обойти абстракцию Spark FS и получить прямой доступ к системе хранения (будь то s3, posix или что-то еще). Недостатком этого решения является отсутствие абстракции: как только вы измените свою базовую ФС, вам также нужно будет изменить свой код.

fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(sc._jsc.hadoopConfiguration())
fs.exists(sc._jvm.org.apache.hadoop.fs.Path("path/to/SUCCESS.txt"))

Спасибо @Dror и @Kini. Я запускаю spark на кластере и должен добавить sc._jvm.java.net.URI.create("s3://" + path.split("/")[2]), здесь s3 префикс файловой системы вашего кластера.

  def path_exists(path):
    # spark is a SparkSession
    sc = spark.sparkContext
    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create("s3://" + path.split("/")[2]),
        sc._jsc.hadoopConfiguration(),
    )
    return fs.exists(sc._jvm.org.apache.hadoop.fs.Path(path))

Вы можете проверить наличие файла, как показано здесь:

import os

if os.path.isfile('/path/file.csv'):
     print("File Exists")
     my_df = spark.read.load("/path/file.csv")
     ...
else:            
     print("File doesn't exists")

dbutils.fs.ls(file_location)

Импортировать нет dbutils. Он уже есть, когда вы запускаете свой кластер.

Ответ, опубликованный @rosefun, сработал для меня, но мне потребовалось много времени, чтобы заставить его работать. Поэтому я даю некоторые подробности о том, как работает это решение и чего следует избегать.

def path_exists(path):
    # spark is a SparkSession
    sc = spark.sparkContext
    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create("s3://" + path.split("/")[2]),
        sc._jsc.hadoopConfiguration(),
    )
    return fs.exists(sc._jvm.org.apache.hadoop.fs.Path(path))

Функция такая же, и она отлично работает, чтобы проверить, существует ли файл или нет в указанном вами пути корзины S3.

Вам придется изменить эту функцию в зависимости от того, как вы указываете значение пути к этой функции.

path = f"s3://bucket-name/import/data/"
pathexists = path_exists(path)

если переменная пути, которую вы определяете, имеет префикс s3 в пути, тогда она будет работать.

Также часть кода, которая разделяет строку, дает вам только имя корзины следующим образом:

path.split("/")[2] will give you `bucket-name`

но если у вас нет префикса s3 в пути, вам придется использовать функцию, изменив некоторый код, как показано ниже:

def path_exists(path):
   # spark is a SparkSession
   sc = spark.sparkContext
   fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(
        sc._jvm.java.net.URI.create("s3://" + path),
        sc._jsc.hadoopConfiguration(),
   )
   return fs.exists(sc._jvm.org.apache.hadoop.fs.Path("s3://" + path))