Как импортировать несколько файлов CSV за одну загрузку?

Используйте подстановочный знак, например замените 2008 на *:

df = sqlContext.read
       .format("com.databricks.spark.csv")
       .option("header", "true")
       .load("../Downloads/*.csv") // <-- note the star (*)

Искра 2.0

// these lines are equivalent in Spark 2.0
spark.read.format("csv").option("header", "true").load("../Downloads/*.csv")
spark.read.option("header", "true").csv("../Downloads/*.csv")

Примечания:

1. Вместо format("com.databricks.spark.csv") используйте метод format("csv") или csv. com.databricks.spark.csv формат был интегрирован в 2.0.

2. Используйте spark, а не sqlContext

Ex1:

Чтение одного CSV-файла. Укажите полный путь к файлу:

 val df = spark.read.option("header", "true").csv("C:spark\\sample_data\\tmp\\cars1.csv")

Ex2:

Чтение нескольких файлов CSV с передачей имен:

val df=spark.read.option("header","true").csv("C:spark\\sample_data\\tmp\\cars1.csv", "C:spark\\sample_data\\tmp\\cars2.csv")

Ex3:

Чтение нескольких файлов CSV с передачей списка имен:

val paths = List("C:spark\\sample_data\\tmp\\cars1.csv", "C:spark\\sample_data\\tmp\\cars2.csv")
val df = spark.read.option("header", "true").csv(paths: _*)

Ex4:

Чтение нескольких файлов CSV в папке без учета других файлов:

val df = spark.read.option("header", "true").csv("C:spark\\sample_data\\tmp\\*.csv")

Ex5:

Чтение нескольких файлов CSV из нескольких папок:

val folders = List("C:spark\\sample_data\\tmp", "C:spark\\sample_data\\tmp1")
val df = spark.read.option("header", "true").csv(folders: _*)

Обратите внимание, что вы можете использовать и другие приемы, например:

-- One or more wildcard:
       .../Downloads20*/*.csv
--  braces and brackets   
       .../Downloads201[1-5]/book.csv
       .../Downloads201{11,15,19,99}/book.csv

Reader's Digest: (Spark 2.x)

Например, если у вас есть 3 каталога с файлами csv:

dir1, dir2, dir3

Затем вы определяете пути как строку списка путей, разделенных запятыми, следующим образом:

пути = "dir1 / , dir2 /, dir3 / *"

Затем используйте следующую функцию и передайте ей эту переменную paths

def get_df_from_csv_paths(paths):

        df = spark.read.format("csv").option("header", "false").\
            schema(custom_schema).\
            option('delimiter', '\t').\
            option('mode', 'DROPMALFORMED').\
            load(paths.split(','))
        return df

К тому времени запустив:

df = get_df_from_csv_paths(paths)

Вы получите в df единый фрейм данных Spark, содержащий данные из всех CSV, найденных в этих трех каталогах.

=============================================== ============================

Полная версия:

Если вы хотите получить несколько CSV-файлов из нескольких каталогов, вам просто нужно передать список и использовать подстановочные знаки.

Например:

если ваш data_path выглядит так:

's3: // bucket_name / subbucket_name / 2016-09 - * / 184 / *,
s3: // bucket_name / subbucket_name / 2016-10 - * / 184 / *,
s3: // bucket_name / subbucket_name / 2016-11 - * / 184 / *,
s3: // bucket_name / subbucket_name / 2016-12 - * / 184 / *, ... '

вы можете использовать указанную выше функцию для одновременного приема всех CSV-файлов во всех этих каталогах и подкаталогах:

Это приведет к загрузке всех каталогов в s3 bucket_name / subbucket_name / в соответствии с указанными шаблонами подстановочных знаков. например первый образец заглянет в

bucket_name / subbucket_name /

для всех каталогов с именами, начинающимися с

2016-09-

и для каждого из них возьмите только каталог с именем

184

и в этом подкаталоге найдите все файлы csv.

И это будет выполняться для каждого из шаблонов в списке, разделенном запятыми.

Это работает лучше, чем союз.

Используя Spark 2.0+, мы можем загружать несколько файлов CSV из разных каталогов, используя df = spark.read.csv(['directory_1','directory_2','directory_3'.....], header=True). Дополнительную информацию см. В документации здесь

val df = spark.read.option("header", "true").csv("C:spark\\sample_data\\*.csv)

будут рассматривать файлы tmp, tmp1, tmp2, ....