使用Spark加载CSV文件

Question

我是Spark的新手，我正在尝试使用Spark从文件中读取CSV数据。以下是我正在做的：

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

我希望这个调用会给我一个文件前两列的列表，但我遇到了以下错误：

索引器中第1行的文件“”：列表索引超出范围

虽然我的CSV文件不止一列。

Answer 1

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

Answer 2

你确定所有的行都至少有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()

Answer 3

火花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")
)

不包括任何外部依赖项。

火花

我建议使用spark csv进行手动解析，而不是一般情况下的简单解析：

确保Spark CSV包含在路径中（--程序包、--jars、--驱动程序类路径）

并按如下方式加载数据：

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