Laden Sie die CSV-Datei mit Spark

Ich bin neu in Spark und versuche, CSV-Daten aus einer Datei mit Spark zu lesen. Folgendes mache ich:

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

Ich würde erwarten, dass dieser Aufruf mir eine Liste der beiden ersten Spalten meiner Datei gibt, aber ich erhalte folgende Fehlermeldung:

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

obwohl meine CSV-Datei als mehr als eine Spalte.

Sind Sie sicher, dass alle Zeilen mindestens 2 Spalten haben? Können Sie so etwas versuchen, nur um es zu überprüfen?:

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

Alternativ können Sie den Täter ausdrucken (falls vorhanden):

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

Spark 2.0.0+

Sie können die integrierte CSV-Datenquelle direkt verwenden:

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

oder

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

ohne Berücksichtigung externer Abhängigkeiten.

Funke <2.0.0 :

Anstelle einer manuellen Analyse, die im Allgemeinen alles andere als trivial ist, würde ich empfehlen spark-csv:

Stellen Sie sicher , dass Funken CSV in dem Pfad enthalten ( --packages, --jars, --driver-class-path)

Und laden Sie Ihre Daten wie folgt:

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

Es kann das Laden, das Ableiten von Schemas und das Löschen fehlerhafter Zeilen verarbeiten und erfordert keine Übergabe von Daten von Python an die JVM.

Hinweis :

Wenn Sie das Schema kennen, ist es besser, Schema-Inferenzen zu vermeiden und an zu übergeben DataFrameReader. Angenommen, Sie haben drei Spalten - Integer, Double und 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"))

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

Und noch eine Option, die darin besteht, die CSV-Datei mit Pandas zu lesen und dann den Pandas DataFrame in Spark zu importieren.

Beispielsweise:

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)

Durch einfaches Teilen durch Komma werden auch Kommas innerhalb von Feldern (z. B. a,b,"1,2,3",c) geteilt, daher wird dies nicht empfohlen. Die Antwort von zero323 ist gut, wenn Sie die DataFrames-API verwenden möchten, aber wenn Sie sich an Base Spark halten möchten, können Sie CSVs in Basis-Python mit dem CSV- Modul analysieren :

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

BEARBEITEN: Wie in den Kommentaren bei @muon erwähnt, wird der Header wie jede andere Zeile behandelt, sodass Sie ihn manuell extrahieren müssen. Beispiel: header = rdd.first(); rdd = rdd.filter(lambda x: x != header)(Stellen Sie sicher, dass Sie keine Änderungen vornehmen, headerbevor der Filter ausgewertet wird.) Aber an diesem Punkt ist es wahrscheinlich besser, einen integrierten CSV-Parser zu verwenden.

Dies ist in PYSPARK

path="Your file path with file name"

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

Dann können Sie überprüfen

df.show(5)
df.count()

Wenn Sie csv als Datenrahmen laden möchten, können Sie Folgendes tun:

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

Es hat gut für mich funktioniert.

Dies steht im Einklang mit dem, was JP Mercier ursprünglich zur Verwendung von Pandas vorgeschlagen hatte, jedoch mit einer wesentlichen Änderung: Wenn Sie Daten in Blöcken in Pandas einlesen, sollten diese formbarer sein. Das bedeutet, dass Sie eine viel größere Datei analysieren können, als Pandas tatsächlich als Einzelstück verarbeiten kann, und sie in kleineren Größen an Spark übergeben können. (Dies beantwortet auch den Kommentar, warum man Spark verwenden möchte, wenn sie sowieso alles in Pandas laden können.)

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

Jetzt gibt es auch eine andere Option für jede allgemeine CSV-Datei: https://github.com/seahboonsiew/pyspark-csv wie folgt:

Angenommen, wir haben den folgenden Kontext

sc = SparkContext
sqlCtx = SQLContext or HiveContext

Verteilen Sie zunächst pyspark-csv.py mithilfe von SparkContext an Ausführende

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

Lesen Sie CSV-Daten über SparkContext und konvertieren Sie sie in DataFrame

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

Wenn Ihre CSV-Daten in keinem der Felder Zeilenumbrüche enthalten, können Sie Ihre Daten mit laden textFile()und analysieren

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)

Wenn Sie eine oder mehrere Zeilen mit weniger oder mehr Spalten als 2 im Dataset haben, kann dieser Fehler auftreten.

Ich bin auch neu in Pyspark und versuche, eine CSV-Datei zu lesen. Der folgende Code hat bei mir funktioniert:

In diesem Code verwende ich einen Datensatz von kaggle. Der Link lautet: https://www.kaggle.com/carrie1/ecommerce-data

1. Ohne das Schema zu erwähnen:

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

Überprüfen Sie nun die Spalten: sdfData.columns

Ausgabe wird sein:

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

Überprüfen Sie den Datentyp für jede Spalte:

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

Dadurch wird der Datenrahmen mit allen Spalten mit dem Datentyp StringType angegeben

2. Mit Schema: Wenn Sie das Schema kennen oder den Datentyp einer Spalte in der obigen Tabelle ändern möchten, verwenden Sie diese (Angenommen, ich habe folgende Spalten und möchte sie in einem bestimmten Datentyp für jede Spalte haben).

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)

Überprüfen Sie nun das Schema für den Datentyp jeder Spalte:

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

Bearbeitet: Wir können auch die folgende Codezeile verwenden, ohne das Schema explizit zu erwähnen:

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

Die Ausgabe ist:

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

Die Ausgabe sieht folgendermaßen aus:

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

Bei der Verwendung spark.read.csvstelle ich fest, dass die Verwendung der Optionen escape='"'und multiLine=Truedie konsistenteste Lösung für den CSV-Standard und meiner Erfahrung nach am besten mit aus Google Sheets exportierten CSV-Dateien funktioniert.

Das ist,

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