Cargar archivo CSV con Spark

110

Soy nuevo en Spark y estoy tratando de leer datos CSV de un archivo con Spark. Esto es lo que estoy haciendo:

sc.textFile('file.csv')
    .map(lambda line: (line.split(',')[0], line.split(',')[1]))
    .collect()

Esperaría que esta llamada me diera una lista de las dos primeras columnas de mi archivo, pero recibo este error:

File "<ipython-input-60-73ea98550983>", line 1, in <lambda>
IndexError: list index out of range

aunque mi archivo CSV tiene más de una columna.

python csv apache-spark pyspark Kernael
fuente

63

¿Estás seguro de que todas las líneas tienen al menos 2 columnas? ¿Puedes probar algo como, solo para comprobar ?:

sc.textFile("file.csv") \
    .map(lambda line: line.split(",")) \
    .filter(lambda line: len(line)>1) \
    .map(lambda line: (line[0],line[1])) \
    .collect()

Alternativamente, puede imprimir el culpable (si lo hubiera):

sc.textFile("file.csv") \
    .map(lambda line: line.split(",")) \
    .filter(lambda line: len(line)<=1) \
    .collect()

G Quintana
fuente

Eso fue todo, una línea con una sola columna, gracias.

Kernael

2

Es mejor analizar usando la csvbiblioteca incorporada para manejar todos los escapes porque simplemente dividir por comas no funcionará si, digamos, hay comas en los valores.

sudo

4

Hay muchas herramientas para analizar csv, no reinventes la rueda

Stephen

2

Este código se romperá si hay una coma entre comillas. Analizar csv es más complicado que simplemente dividir en ",".

Alceu Costa

Esto se interrumpe por comas. Esto es muy malo.

rjurney

184

Spark 2.0.0+

Puede utilizar la fuente de datos csv incorporada directamente:

spark.read.csv(
    "some_input_file.csv", header=True, mode="DROPMALFORMED", schema=schema
)

o

(spark.read
    .schema(schema)
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .csv("some_input_file.csv"))

sin incluir dependencias externas.

Chispa <2.0.0 :

En lugar del análisis manual, que está lejos de ser trivial en un caso general, recomendaría spark-csv:

Asegúrese de que CSV Spark está incluido en la ruta de acceso ( --packages, --jars, --driver-class-path)

Y cargue sus datos de la siguiente manera:

(df = sqlContext
    .read.format("com.databricks.spark.csv")
    .option("header", "true")
    .option("inferschema", "true")
    .option("mode", "DROPMALFORMED")
    .load("some_input_file.csv"))

Puede manejar la carga, la inferencia del esquema, la eliminación de líneas mal formadas y no requiere pasar datos de Python a la JVM.

Nota :

Si conoce el esquema, es mejor evitar la inferencia del esquema y pasarlo a DataFrameReader. Suponiendo que tiene tres columnas: entero, doble y cadena:

from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import DoubleType, IntegerType, StringType

schema = StructType([
    StructField("A", IntegerType()),
    StructField("B", DoubleType()),
    StructField("C", StringType())
])

(sqlContext
    .read
    .format("com.databricks.spark.csv")
    .schema(schema)
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .load("some_input_file.csv"))

zero323
fuente

6

Si hace esto, no olvide incluir el paquete databricks csv cuando abra el shell pyspark o use spark-submit. Por ejemplo, pyspark --packages com.databricks:spark-csv_2.11:1.4.0(asegúrese de cambiar las versiones de databricks / spark a las que ha instalado).

Galen Long

¿Es csvContext o sqlContext en pyspark? Porque en scala necesitas csvContext

Geoffrey Anderson

28

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

df = spark.read.csv("/home/stp/test1.csv",header=True,sep="|");

print(df.collect())

y durga prasad
fuente

use 'sep no' separador 'de la siguiente manera: df = spark.read.csv ("/ home / stp / test1.csv", header = True, sep = "|")

Grant Shannon

18

Y otra opción más que consiste en leer el archivo CSV usando Pandas y luego importar el Pandas DataFrame a Spark.

Por ejemplo:

from pyspark import SparkContext
from pyspark.sql import SQLContext
import pandas as pd

sc = SparkContext('local','example')  # if using locally
sql_sc = SQLContext(sc)

pandas_df = pd.read_csv('file.csv')  # assuming the file contains a header
# pandas_df = pd.read_csv('file.csv', names = ['column 1','column 2']) # if no header
s_df = sql_sc.createDataFrame(pandas_df)

JP Mercier
fuente

7

¿Por qué OP le gustaría hacer en Spark si puede cargar datos en pandas?

WoodChopper

No quiero instalar o especificar dependencias en cada clúster de chispa ....

SummerEla

Panda permite la fragmentación de archivos al leer, por lo que todavía hay un caso de uso aquí para que Pandas maneje el análisis inicial de archivos. Vea mi respuesta a continuación para obtener el código.

abby sobh

Precaución: Pandas también maneja el esquema de columna de manera diferente a Spark, especialmente cuando hay espacios en blanco involucrados. Es más seguro cargar csv como cadenas para cada columna.

AntiPawn79

@WoodChopper Puedes usar Pandas como UDF en Spark, ¿no?

flow2k

16

Simplemente dividir por comas también dividirá las comas que están dentro de los campos (por ejemplo a,b,"1,2,3",c), por lo que no se recomienda. La respuesta de zero323 es buena si desea utilizar la API de DataFrames, pero si desea ceñirse a la base Spark, puede analizar csvs en base Python con el módulo csv :

# works for both python 2 and 3
import csv
rdd = sc.textFile("file.csv")
rdd = rdd.mapPartitions(lambda x: csv.reader(x))

EDITAR: Como @muon mencionó en los comentarios, esto tratará el encabezado como cualquier otra fila, por lo que deberá extraerlo manualmente. Por ejemplo, header = rdd.first(); rdd = rdd.filter(lambda x: x != header)(asegúrese de no modificar headerantes de que se evalúe el filtro). Pero en este punto, probablemente sea mejor que utilice un analizador csv incorporado.

Galen Long
fuente

1

No necesita Hive para usar DataFrames. En cuanto a su solución: a) No es necesario StringIO. csvpuede usar cualquier iterable b) __next__no debe usarse directamente y fallará en la línea vacía. Eche un vistazo a flatMap c) Sería mucho más eficiente de usar en mapPartitionslugar de inicializar el lector en cada línea :)

zero323

¡Muchas gracias por las correcciones! Antes de editar mi respuesta, quiero asegurarme de que la entiendo completamente. 1) ¿Por qué rdd.mapPartitions(lambda x: csv.reader(x))funciona mientras rdd.map(lambda x: csv.reader(x))arroja un error? Esperaba que ambos lanzaran lo mismo TypeError: can't pickle _csv.reader objects. También parece que mapPartitionsllama automáticamente a algunos equivalentes a "readlines" en el csv.readerobjeto, donde con map, necesitaba llamar __next__explícitamente para sacar las listas de csv.reader. 2) ¿Dónde flatMapentra? Llamar mapPartitionssolo funcionó para mí.

Galen Long

1

rdd.mapPartitions(lambda x: csv.reader(x))funciona porque mapPartitionsespera un Iterableobjeto. Si quieres ser explícito, puedes comprensión o expresión generadora. mapsolo no funciona porque no itera sobre el objeto. De ahí mi sugerencia de uso, flatMap(lambda x: csv.reader([x]))que iterará sobre el lector. Pero mapPartitionsaquí es mucho mejor.

zero323

1

tenga en cuenta que esto leerá el encabezado como una fila de datos, no como encabezado

muon

7

Esto está en PYSPARK

path="Your file path with file name"

df=spark.read.format("csv").option("header","true").option("inferSchema","true").load(path)

Entonces puedes comprobar

df.show(5)
df.count()

grano de amarnath
fuente

6

Si desea cargar csv como un marco de datos, puede hacer lo siguiente:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

df = sqlContext.read.format('com.databricks.spark.csv') \
    .options(header='true', inferschema='true') \
    .load('sampleFile.csv') # this is your csv file

Funcionó bien para mí.

Jeril
fuente

@GalenLong, si no le importa, ¿puede compartir la respuesta ya existente

Jeril

Extraño, juro que hubo otra respuesta con esta solución. Quizás lo confundí con otra pregunta. Culpa mía.

Galen Long

5

Esto está en línea con lo que JP Mercier sugirió inicialmente sobre el uso de Pandas, pero con una modificación importante: si lee datos en Pandas en trozos, debería ser más maleable. Lo que significa que puede analizar un archivo mucho más grande de lo que Pandas realmente puede manejar como una sola pieza y pasarlo a Spark en tamaños más pequeños. (Esto también responde al comentario sobre por qué uno querría usar Spark si de todos modos pueden cargar todo en Pandas).

from pyspark import SparkContext
from pyspark.sql import SQLContext
import pandas as pd

sc = SparkContext('local','example')  # if using locally
sql_sc = SQLContext(sc)

Spark_Full = sc.emptyRDD()
chunk_100k = pd.read_csv("Your_Data_File.csv", chunksize=100000)
# if you have headers in your csv file:
headers = list(pd.read_csv("Your_Data_File.csv", nrows=0).columns)

for chunky in chunk_100k:
    Spark_Full +=  sc.parallelize(chunky.values.tolist())

YourSparkDataFrame = Spark_Full.toDF(headers)
# if you do not have headers, leave empty instead:
# YourSparkDataFrame = Spark_Full.toDF()
YourSparkDataFrame.show()

abby sollozo
fuente

5

Ahora, también hay otra opción para cualquier archivo csv general: https://github.com/seahboonsiew/pyspark-csv de la siguiente manera:

Supongamos que tenemos el siguiente contexto

sc = SparkContext
sqlCtx = SQLContext or HiveContext

Primero, distribuya pyspark-csv.py a los ejecutores usando SparkContext

import pyspark_csv as pycsv
sc.addPyFile('pyspark_csv.py')

Leer datos csv a través de SparkContext y convertirlos a DataFrame

plaintext_rdd = sc.textFile('hdfs://x.x.x.x/blah.csv')
dataframe = pycsv.csvToDataFrame(sqlCtx, plaintext_rdd)

optimista
fuente

3

Si sus datos csv no contienen líneas nuevas en ninguno de los campos, puede cargar sus datos textFile()y analizarlos

import csv
import StringIO

def loadRecord(line):
    input = StringIO.StringIO(line)
    reader = csv.DictReader(input, fieldnames=["name1", "name2"])
    return reader.next()

input = sc.textFile(inputFile).map(loadRecord)

iec2011007
fuente

2

Si tiene una o más filas con menos o más columnas que 2 en el conjunto de datos, puede surgir este error.

También soy nuevo en Pyspark y estoy tratando de leer un archivo CSV. El siguiente código funcionó para mí:

En este código, estoy usando un conjunto de datos de kaggle, el enlace es: https://www.kaggle.com/carrie1/ecommerce-data

1. Sin mencionar el esquema:

from pyspark.sql import SparkSession  
scSpark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example: Reading CSV file without mentioning schema") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

sdfData = scSpark.read.csv("data.csv", header=True, sep=",")
sdfData.show()

Ahora verifique las columnas: sdfData.columns

La salida será:

['InvoiceNo', 'StockCode','Description','Quantity', 'InvoiceDate', 'CustomerID', 'Country']

Verifique el tipo de datos para cada columna:

sdfData.schema
StructType(List(StructField(InvoiceNo,StringType,true),StructField(StockCode,StringType,true),StructField(Description,StringType,true),StructField(Quantity,StringType,true),StructField(InvoiceDate,StringType,true),StructField(UnitPrice,StringType,true),StructField(CustomerID,StringType,true),StructField(Country,StringType,true)))

Esto dará el marco de datos con todas las columnas con tipo de datos como StringType

2. Con esquema: si conoce el esquema o desea cambiar el tipo de datos de cualquier columna en la tabla anterior, use esto (digamos que tengo las siguientes columnas y las quiero en un tipo de datos particular para cada una de ellas)

from pyspark.sql import SparkSession  
from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import DoubleType, IntegerType, StringType
    schema = StructType([\
        StructField("InvoiceNo", IntegerType()),\
        StructField("StockCode", StringType()), \
        StructField("Description", StringType()),\
        StructField("Quantity", IntegerType()),\
        StructField("InvoiceDate", StringType()),\
        StructField("CustomerID", DoubleType()),\
        StructField("Country", StringType())\
    ])

scSpark = SparkSession \
    .builder \
    .appName("Python Spark SQL example: Reading CSV file with schema") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

sdfData = scSpark.read.csv("data.csv", header=True, sep=",", schema=schema)

Ahora verifique el esquema para el tipo de datos de cada columna:

sdfData.schema

StructType(List(StructField(InvoiceNo,IntegerType,true),StructField(StockCode,StringType,true),StructField(Description,StringType,true),StructField(Quantity,IntegerType,true),StructField(InvoiceDate,StringType,true),StructField(CustomerID,DoubleType,true),StructField(Country,StringType,true)))

Editado: También podemos usar la siguiente línea de código sin mencionar el esquema explícitamente:

sdfData = scSpark.read.csv("data.csv", header=True, inferSchema = True)
sdfData.schema

La salida es:

StructType(List(StructField(InvoiceNo,StringType,true),StructField(StockCode,StringType,true),StructField(Description,StringType,true),StructField(Quantity,IntegerType,true),StructField(InvoiceDate,StringType,true),StructField(UnitPrice,DoubleType,true),StructField(CustomerID,IntegerType,true),StructField(Country,StringType,true)))

La salida se verá así:

sdfData.show()

+---------+---------+--------------------+--------+--------------+----------+-------+
|InvoiceNo|StockCode|         Description|Quantity|   InvoiceDate|CustomerID|Country|
+---------+---------+--------------------+--------+--------------+----------+-------+
|   536365|   85123A|WHITE HANGING HEA...|       6|12/1/2010 8:26|      2.55|  17850|
|   536365|    71053| WHITE METAL LANTERN|       6|12/1/2010 8:26|      3.39|  17850|
|   536365|   84406B|CREAM CUPID HEART...|       8|12/1/2010 8:26|      2.75|  17850|
|   536365|   84029G|KNITTED UNION FLA...|       6|12/1/2010 8:26|      3.39|  17850|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|12/1/2010 8:26|      3.39|  17850|
|   536365|    22752|SET 7 BABUSHKA NE...|       2|12/1/2010 8:26|      7.65|  17850|
|   536365|    21730|GLASS STAR FROSTE...|       6|12/1/2010 8:26|      4.25|  17850|
|   536366|    22633|HAND WARMER UNION...|       6|12/1/2010 8:28|      1.85|  17850|
|   536366|    22632|HAND WARMER RED P...|       6|12/1/2010 8:28|      1.85|  17850|
|   536367|    84879|ASSORTED COLOUR B...|      32|12/1/2010 8:34|      1.69|  13047|
|   536367|    22745|POPPY'S PLAYHOUSE...|       6|12/1/2010 8:34|       2.1|  13047|
|   536367|    22748|POPPY'S PLAYHOUSE...|       6|12/1/2010 8:34|       2.1|  13047|
|   536367|    22749|FELTCRAFT PRINCES...|       8|12/1/2010 8:34|      3.75|  13047|
|   536367|    22310|IVORY KNITTED MUG...|       6|12/1/2010 8:34|      1.65|  13047|
|   536367|    84969|BOX OF 6 ASSORTED...|       6|12/1/2010 8:34|      4.25|  13047|
|   536367|    22623|BOX OF VINTAGE JI...|       3|12/1/2010 8:34|      4.95|  13047|
|   536367|    22622|BOX OF VINTAGE AL...|       2|12/1/2010 8:34|      9.95|  13047|
|   536367|    21754|HOME BUILDING BLO...|       3|12/1/2010 8:34|      5.95|  13047|
|   536367|    21755|LOVE BUILDING BLO...|       3|12/1/2010 8:34|      5.95|  13047|
|   536367|    21777|RECIPE BOX WITH M...|       4|12/1/2010 8:34|      7.95|  13047|
+---------+---------+--------------------+--------+--------------+----------+-------+
only showing top 20 rows

Yogesh
fuente

1

Cuando lo uso spark.read.csv, encuentro que usar las opciones escape='"'y multiLine=Truebrindar la solución más consistente para el estándar CSV , y en mi experiencia, funciona mejor con archivos CSV exportados desde Google Sheets.

Es decir,

#set inferSchema=False to read everything as string
df = spark.read.csv("myData.csv", escape='"', multiLine=True,
     inferSchema=False, header=True)

flow2k
fuente

de donde viene la chispa ¿es import pyspark as spark?

Luk Aron

@LukAron En un shell pyspark, sparkya está inicializado. En un script enviado por spark-submit, puede crear una instancia como from pyspark.sql import SparkSession; spark = SparkSession.builder.getOrCreate().

flow2k

Cargar archivo CSV con Spark

Respuestas: