Cargar archivo CSV con Spark

Chispa 2.0.0+

Puede utilizar directamente la fuente de datos csv integrada:

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 ninguna dependencia externa.

Chispa :

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

Asegúrese de que Spark CSV esté incluido en la ruta (--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 carga, inferencia de esquemas, caída de líneas malformadas y no requiere pasar datos de Python a la JVM.

Nota:

Si conoce el esquema, es mejor evitar la inferencia de 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"))

Simplemente dividir por coma 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 quieres usar la API de DataFrames, pero si quieres apegarte a base Spark, puedes analizar csv 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))

EDIT: Como @muon mencionó en los comentarios, esto tratará el encabezado como cualquier otra fila, por lo que necesitará extraerlo manualmente. Por ejemplo, header = rdd.first(); rdd = rdd.filter(lambda x: x != header) (asegúrese de no modificar header antes el filtro evalúa). Pero en este punto, probablemente sea mejor usar un analizador CSV integrado.

Y otra opción que consiste en leer el archivo CSV usando Pandas y luego importar el DataFrame de Pandas en 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)

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,separator="|");

print(df.collect())

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. Es decir, que puedes analizar un archivo mucho más grande de lo que los Pandas realmente pueden 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 pueden cargar todo en Pandas de todos modos.)

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

Ahora, también hay otra opción para cualquier archivo csv general: https://github.com/seahboonsiew/pyspark-csv como sigue:

Supongamos que tenemos el siguiente contexto

sc = SparkContext
sqlCtx = SQLContext or HiveContext

Primero, distribuir pyspark-csv.py a ejecutores usando SparkContext

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

Lea los datos csv a través de SparkContext y conviértalos en DataFrame

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

Si sus datos csv no contienen nuevas líneas en ninguno de los campos, puede cargar sus datos con 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)

Si desea cargar csv como un dataframe, 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í.

import pandas as pd

data1 = pd.read_csv("test1.csv")
data2 = pd.read_csv("train1.csv")