问题:用Spark加载CSV文件
我是Spark的新手,正在尝试使用Spark从文件读取CSV数据。这是我在做什么:
sc.textFile('file.csv')
.map(lambda line: (line.split(',')[0], line.split(',')[1]))
.collect()
我希望此调用可以给我列出文件的前两列,但出现此错误:
File "<ipython-input-60-73ea98550983>", line 1, in <lambda>
IndexError: list index out of range
尽管我的CSV文件不止一列。
回答 0
您确定所有行都至少有2列?您可以尝试类似的方法吗?
sc.textFile("file.csv") \
.map(lambda line: line.split(",")) \
.filter(lambda line: len(line)>1) \
.map(lambda line: (line[0],line[1])) \
.collect()
或者,您可以打印罪魁祸首(如果有):
sc.textFile("file.csv") \
.map(lambda line: line.split(",")) \
.filter(lambda line: len(line)<=1) \
.collect()
回答 1
Spark 2.0.0+
您可以直接使用内置的csv数据源:
spark.read.csv(
"some_input_file.csv", header=True, mode="DROPMALFORMED", schema=schema
)
要么
(spark.read
.schema(schema)
.option("header", "true")
.option("mode", "DROPMALFORMED")
.csv("some_input_file.csv"))
不包括任何外部依赖项。
火花<2.0.0:
我建议不要手动解析,这在一般情况下是不容易的,我建议spark-csv
:
确保星火CSV包含在路径(--packages
,--jars
,--driver-class-path
)
并按以下方式加载数据:
(df = sqlContext
.read.format("com.databricks.spark.csv")
.option("header", "true")
.option("inferschema", "true")
.option("mode", "DROPMALFORMED")
.load("some_input_file.csv"))
它可以处理加载,模式推断,删除格式错误的行,并且不需要将数据从Python传递到JVM。
注意事项:
如果您知道架构,则最好避免架构推断并将其传递给DataFrameReader
。假设您有三列-整数,双精度和字符串:
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"))
回答 2
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())
回答 3
还有另一个选择,包括使用Pandas读取CSV文件,然后将Pandas DataFrame导入Spark。
例如:
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)
回答 4
只需按逗号分割也会将字段内的逗号分割(例如a,b,"1,2,3",c
),因此不建议使用。如果要使用DataFrames API,zero323的答案很好,但是如果要坚持使用基本Spark,则可以使用csv模块在基本Python中解析csvs :
# works for both python 2 and 3
import csv
rdd = sc.textFile("file.csv")
rdd = rdd.mapPartitions(lambda x: csv.reader(x))
编辑:正如@muon在评论中提到的那样,它将像其他任何行一样对待标头,因此您需要手动提取它。例如,header = rdd.first(); rdd = rdd.filter(lambda x: x != header)
(确保header
在评估过滤器之前不要进行修改)。但是在这一点上,最好使用内置的csv解析器。
回答 5
这是在PYSPARK中
path="Your file path with file name"
df=spark.read.format("csv").option("header","true").option("inferSchema","true").load(path)
那你可以检查
df.show(5)
df.count()
回答 6
如果要将csv加载为数据帧,则可以执行以下操作:
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
对我来说很好。
回答 7
这与JP Mercier最初提出的有关使用Pandas的建议是一致的,但进行了重大修改:如果将数据分块读取到Pandas中,应该更具延展性。这意味着,您可以解析比Pandas实际可处理的文件大得多的文件,并将其以较小的尺寸传递给Spark。(这也回答了有关为什么如果他们仍然可以将所有内容加载到Pandas中的人为什么要使用Spark的评论。)
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()
回答 8
现在,对于任何常规的csv文件,还有另一个选项:https : //github.com/seahboonsiew/pyspark-csv,如下所示:
假设我们具有以下上下文
sc = SparkContext
sqlCtx = SQLContext or HiveContext
首先,使用SparkContext将pyspark-csv.py分发给执行者
import pyspark_csv as pycsv
sc.addPyFile('pyspark_csv.py')
通过SparkContext读取CSV数据并将其转换为DataFrame
plaintext_rdd = sc.textFile('hdfs://x.x.x.x/blah.csv')
dataframe = pycsv.csvToDataFrame(sqlCtx, plaintext_rdd)
回答 9
如果您的csv数据恰好在任何字段中都不包含换行符,则可以使用加载textFile()
并解析数据
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)
回答 10
如果数据集中的任何一个或多个行的列数少于或多于2,则可能会出现此错误。
我也是Pyspark的新手,正在尝试读取CSV文件。以下代码为我工作:
在这段代码中,我使用来自kaggle的数据集,链接为:https ://www.kaggle.com/carrie1/ecommerce-data
1.不提架构:
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()
现在检查列:sdfData.columns
输出将是:
['InvoiceNo', 'StockCode','Description','Quantity', 'InvoiceDate', 'CustomerID', 'Country']
检查每一列的数据类型:
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)))
这将为数据框提供所有列,其数据类型为StringType
2.使用架构: 如果您知道架构或想要更改上表中任何列的数据类型,请使用此格式(假设我正在关注以下列,并希望它们具有特定的数据类型)
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)
现在检查每个列的数据类型的架构:
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)))
编辑:我们也可以使用以下代码行,而无需明确提及架构:
sdfData = scSpark.read.csv("data.csv", header=True, inferSchema = True)
sdfData.schema
输出为:
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)))
输出将如下所示:
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
回答 11
使用时spark.read.csv
,我发现使用这些选项escape='"'
并multiLine=True
为CSV标准提供最一致的解决方案,以我的经验,从Google表格中导出的CSV文件效果最好。
那是,
#set inferSchema=False to read everything as string
df = spark.read.csv("myData.csv", escape='"', multiLine=True,
inferSchema=False, header=True)