Charger un fichier CSV avec Spark

Je suis nouveau à l'Étincelle et je suis en train de lire des données au format CSV à partir d'un fichier avec Spark.
Voici ce que j'ai fais :

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

Je m'attends à cet appel pour me donner une liste des deux premières colonnes de mon fichier mais j'obtiens cette erreur :

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

bien que mon fichier CSV que plus d'une colonne.

InformationsquelleAutor Kernael | 2015-02-28

55

Êtes-vous sûr que tous les lignes ont au moins 2 colonnes? Pouvez-vous essayer quelque chose comme, juste pour vérifier?:
```
sc.textFile("file.csv") \
    .map(lambda line: line.split(",")) \
    .filter(lambda line: len(line)>1) \
    .map(lambda line: (line[0],line[1])) \
    .collect()
```
Alternativement, vous pouvez imprimer le coupable (le cas échéant):
```
sc.textFile("file.csv") \
    .map(lambda line: line.split(",")) \
    .filter(lambda line: len(line)<=1) \
    .collect()
```
- Qui était-il, d'une ligne avec une seule colonne, je vous remercie.
- Il est préférable d'analyser l'utilisation du haut- csv bibliothèque pour gérer tous les échapper, parce que tout simplement le fractionnement par des virgules ne fonctionnera pas si, par exemple, il y a des virgules dans les valeurs.
- Il existe de nombreux outils pour analyser csv, ne pas réinventer la roue
- Ce code ne fonctionnera plus si il y a une virgule à l'intérieur des guillemets. L'analyse csv est plus compliqué que simplement fractionnement à ",".
InformationsquelleAutor G Quintana
148

Étincelle 2.0.0+

Vous pouvez utiliser les haut-csv source de données directement:
```
spark.read.csv(
    "some_input_file.csv", header=True, mode="DROPMALFORMED", schema=schema
)
```
ou
```
(spark.read
    .schema(schema)
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .csv("some_input_file.csv"))
```
sans dépendances externes.

Spark < 2.0.0:

Au lieu de manuel d'analyse, ce qui est loin d'être négligeable dans le cas général, je recommande spark-csv:

Assurez-vous que l'Étincelle CSV est inclus dans le chemin d'accès (--packages, --jars, --driver-class-path)

Et de charger vos données comme suit:
```
(df = sqlContext
    .read.format("com.databricks.spark.csv")
    .option("header", "true")
    .option("inferschema", "true")
    .option("mode", "DROPMALFORMED")
    .load("some_input_file.csv"))
```
Il peut gérer le chargement, le schéma d'inférence, en laissant tomber la malformation de lignes et ne nécessite pas le passage des données à partir de Python à la JVM.

Note:

Si vous connaissez le schéma, il est préférable d'éviter de schéma d'inférence et de le passer à DataFrameReader. En supposant que vous avez trois colonnes - integer, double et string:
```
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"))
```
- Si vous faites cela, n'oubliez pas d'inclure le databricks csv paquet lorsque vous ouvrez le pyspark shell ou de l'utilisation de l'étincelle-soumettre. Par exemple, pyspark --packages com.databricks:spark-csv_2.11:1.4.0 (assurez-vous de changer la databricks/spark versions à ceux que vous avez installé).
- Est-il csvContext ou sqlContext dans pyspark? Parce que dans scala vous avez besoin csvContext
InformationsquelleAutor zero323
15

Et encore une autre option qui consiste à lire le fichier CSV à l'aide de Pandas, puis de l'importer les Pandas DataFrame dans Spark.

Par exemple:
```
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)
```
- Pourquoi serait-OP voudrais faire sur la spark s'il est capable de charger des données dans les pandas
- Des fins académiques
- Ne voulant pas installer ou de spécifier les dépendances sur chaque étincelle de cluster....
- Panda permet fichier de segmentation lors de la lecture de sorte qu'il est encore un cas d'utilisation ici pour avoir des Pandas de la poignée initiale de l'analyse du fichier. Voir ma réponse ci-dessous pour le code.
- Attention: les Pandas gère également le schéma de colonne façon différemment étincelle surtout quand il y a des espaces concernés. Plus sûr de charge csv dans des chaînes de caractères pour chaque colonne.
InformationsquelleAutor JP Mercier

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())

utiliser la "sep pas" séparateur " comme suit: df = étincelle.lire.csv("/home/stp/test1.csv",header=True,sep="|")

InformationsquelleAutor y durga prasad

13

Simplement fractionnement par la virgule sera également diviser les virgules qui sont dans les champs (par exemple,a,b,"1,2,3",c), de sorte qu'il n'est pas recommandé. zero323 réponse est bon si vous voulez utiliser le DataFrames API, mais si vous voulez coller à la base de l'Étincelle, vous pouvez analyser ces volumes dans la base de Python, avec la csv module:
```
# works for both python 2 and 3
import csv
rdd = sc.textFile("file.csv")
rdd = rdd.mapPartitions(lambda x: csv.reader(x))
```
EDIT: Comme @muon mentionné dans les commentaires, cela permettra de traiter l'en-tête comme n'importe quel autre ligne de sorte que vous aurez besoin de l'extraire manuellement. Par exemple, header = rdd.first(); rdd = rdd.filter(lambda x: x != header) (assurez-vous de ne pas modifier header avant le filtre évalue). Mais à ce stade, vous êtes probablement mieux d'utiliser un haut-csv analyseur.
- Vous n'avez pas besoin de la Ruche pour utiliser DataFrames. Quant à votre solution: a) Il n'est pas nécessaire pour StringIO. csv pouvez utiliser n'importe quel itérable b) __next__ ne devrait pas être utilisé directement et échoue sur la ligne vide. Jetez un oeil à flatMap c) Il serait beaucoup plus efficace d'utiliser mapPartitions au lieu d'initialisation du lecteur sur chaque ligne 🙂
- Merci beaucoup pour les corrections! Avant je modifier ma réponse, je veux m'assurer que je comprends totalement. 1) Pourquoi ne rdd.mapPartitions(lambda x: csv.reader(x)) travail tout en rdd.map(lambda x: csv.reader(x)) renvoie une erreur? Je m'attendais à la fois de jeter le même TypeError: can't pickle _csv.reader objects. Il semble aussi que mapPartitions appelle automatiquement l'équivalent de "readlines" sur le csv.reader l'objet, le cas avec map, j'ai besoin d'appeler __next__ explicitement pour obtenir la liste des csv.reader. 2) d'Où vient flatMap venir? Juste appeler mapPartitions seul a fonctionné pour moi.
- rdd.mapPartitions(lambda x: csv.reader(x)) fonctionne parce que mapPartitions s'attend à une Iterable objet. Si vous voulez être explicite, vous pourriez vous de compréhension ou de générateur d'expression. map seul ne fonctionne pas car il n'a pas d'itérer sur l'objet. D'où ma suggestion d'utiliser flatMap(lambda x: csv.reader([x])) qui va se répéter sur le lecteur. Mais mapPartitions est beaucoup mieux ici.
- notez qu'il sera lu comme une ligne de données, pas comme en-tête
InformationsquelleAutor Galen Long
4

Maintenant, il y a également une autre option pour tout fichier csv: https://github.com/seahboonsiew/pyspark-csv comme suit:

Supposer que nous avons le contexte suivant
```
sc = SparkContext
sqlCtx = SQLContext or HiveContext
```
D'abord, distribuer pyspark-csv.py pour les liquidateurs de l'aide SparkContext
```
import pyspark_csv as pycsv
sc.addPyFile('pyspark_csv.py')
```
Lire des données au format csv via SparkContext et la convertir en DataFrame
```
plaintext_rdd = sc.textFile('hdfs://x.x.x.x/blah.csv')
dataframe = pycsv.csvToDataFrame(sqlCtx, plaintext_rdd)
```
InformationsquelleAutor optimist

Ceci est en ligne avec ce que JP Mercier initialement suggéré sur l'utilisation des Pandas, mais avec une modification majeure: Si vous lisez des données dans les Pandas en morceaux, il devrait être plus malléable. Ce qui signifie que vous pouvez analyser un fichier assez volumineux que les Pandas peuvent effectivement gérer en une seule pièce, et le transmettre à l'Étincelle dans les petites tailles. (Cela répond également le commentaire à propos de pourquoi vouloir utiliser Étincelle si on peut charger le tout dans les Pandas de toute façon.)

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()

InformationsquelleAutor abby sobh

3

Si vous souhaitez charger csv comme un dataframe ensuite, vous pouvez effectuer les opérations suivantes:
```
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
```
Il a bien fonctionné pour moi.
- Je downvoted car cette réponse existe déjà.
- si vous n'avez pas l'esprit, pouvez-vous partager le déjà existant répondre
- Bizarre, je vous jure, il y avait une autre réponse à cette solution. Peut-être que j'ai eu ce confondue avec une autre question. Mon mauvais.
InformationsquelleAutor Jeril

Si vos données csv arrive pas contenir des sauts de ligne dans l'un des champs, vous pouvez charger vos données avec textFile() et l'analyse

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)

InformationsquelleAutor iec2011007

Si vous rencontrez un ou plusieurs ligne(s) avec plus ou moins de nombre de colonnes que 2 dans le jeu de données puis cette erreur peut se produire.

Je suis aussi nouvelle pour Pyspark et en essayant de lire un fichier CSV. Suivant le code a fonctionné pour moi:

Dans ce code, je suis en utilisant des données de kaggle le lien est: https://www.kaggle.com/carrie1/ecommerce-data

1. Sans mentionner le schéma:

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()

Maintenant vérifier les colonnes:
sdfData.les colonnes

Sortie sera:

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

Vérifier le type de données pour chaque colonne:

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

Ce que va donner la trame de données avec toutes les colonnes de type de données que Chaînetapez

2. Avec le schéma:
Si vous connaissez le schéma ou vous voulez changer le type de données de chaque colonne dans le tableau ci-dessus puis utiliser cette (disons que je vais avoir des colonnes suivantes et en un type de données particulier pour chacun d'eux)

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)

Maintenant vérifier le schéma pour le type de données de chaque colonne:

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

Édité: Nous pouvons utiliser la ligne de code suivante ainsi sans mentionner schéma explicite:

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

La sortie est:

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 sortie ressemble à ceci:

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

InformationsquelleAutor Yogesh

-6
```
import pandas as pd
data1 = pd.read_csv("test1.csv")
data2 = pd.read_csv("train1.csv")
```
- C'est les Pandas, pas d'Étincelle. L'auteur veut que les données soient chargées dans une mémoire distribuée Étincelle de cluster, pas une seule machine.
InformationsquelleAutor hey kay

Vous devez vous connecter pour publier un commentaire.