apache-spark - unir - spark read csv

¿Cómo importar varios archivos csv en una sola carga?

Question

Considere que tengo un esquema definido para cargar 10 archivos csv en una carpeta. ¿Hay una manera de cargar tablas automáticamente usando Spark SQL? Sé que esto se puede realizar utilizando un marco de datos individual para cada archivo [que se muestra a continuación], pero ¿se puede automatizar con un solo comando en lugar de apuntar a un archivo? ¿Puedo apuntar una carpeta?

df = sqlContext.read
       .format("com.databricks.spark.csv")
       .option("header", "true")
       .load("../Downloads/2008.csv")

Answer 1

Tenga en cuenta que puede utilizar otros trucos como:

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

Answer 2

Usando Spark 2.0+, podemos cargar múltiples archivos CSV de diferentes directorios usando df = spark.read.csv([''directory_1'',''directory_2'',''directory_3''.....], header=True) . Para más información, consulte la documentación here

Answer 3

Utilice comodín, por ejemplo, reemplace 2008 con * :

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

Spark 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")

Notas:

1. Reemplace el format("com.databricks.spark.csv") utilizando el format("csv") o el método csv . com.databricks.spark.csv formato com.databricks.spark.csv se ha integrado a 2.0.

2. Utilice spark no sqlContext

Answer 4

Ex1 :

Leyendo un solo archivo CSV. Proporcione la ruta completa del archivo:

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

Ex2 :

Leyendo múltiples archivos CSV pasando nombres:

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

Ex3 :

Leyendo múltiples archivos CSV pasando lista de nombres:

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 :

Leyendo múltiples archivos CSV en una carpeta ignorando otros archivos:

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

Ex5 :

Leyendo múltiples archivos CSV desde múltiples carpetas:

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

Answer 5

Reader''s Digest: (Spark 2.x)

Por ejemplo, si tiene 3 directorios que contienen archivos csv:

dir1, dir2, dir3

A continuación, define las rutas como una cadena de rutas delimitadas por comas de la siguiente manera:

paths = "dir1 / , dir2 / , dir3 / *"

Luego usa la siguiente función y pásale esta variable de caminos.

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

Para entonces correr:

df = get_df_from_csv_paths(paths)

Obtendrá en df un único marco de datos de chispa que contiene los datos de todos los csvs que se encuentran en estos 3 directorios.

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

Versión completa:

En caso de que desee ingerir varios CSV de varios directorios , simplemente necesita pasar una lista y usar comodines.

Por ejemplo :

si su data_path se ve así:

''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 / *, ... ''

puede utilizar la función anterior para ingerir todos los csvs en todos estos directorios y subdirectorios a la vez:

Esto ingeriría todos los directorios en s3 bucket_name / subbucket_name / de acuerdo con los patrones de comodín especificados. por ejemplo, el primer patrón se vería en

bucket_name / subbucket_name /

para todos los directorios con nombres que comiencen con

2016-09-

y para cada uno de ellos tome solo el directorio nombrado

184

y dentro de ese subdirectorio busque todos los archivos csv.

Y esto se ejecutaría para cada uno de los patrones en la lista delimitada por comas.

Esto funciona mucho mejor que la unión ...