Importar o conteúdo do arquivo csv para os quadros de dados pyspark

12

Como posso importar um arquivo .csv para os quadros de dados pyspark? Eu até tentei ler o arquivo csv no Pandas e convertê-lo em um dataframe spark usando createDataFrame, mas ele ainda mostra algum erro. Alguém pode me guiar por isso? Além disso, diga-me como posso importar um arquivo xlsx? Estou tentando importar conteúdo csv para quadros de dados do pandas e depois convertê-lo em quadros de dados spark, mas está mostrando o erro:

"Py4JJavaError" An error occurred while calling o28.applySchemaToPythonRDD. : java.lang.RuntimeException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient 

Meu código é:

from pyspark import SparkContext 
from pyspark.sql import SQLContext 
import pandas as pd 
sqlc=SQLContext(sc) 
df=pd.read_csv(r'D:\BestBuy\train.csv') 
sdf=sqlc.createDataFrame(df) 
neha
fonte
11
Se você tiver uma mensagem de erro, deve publicá-la; provavelmente tem informações importantes para ajudar a depurar a situação.
Jagartner #
Eu estou tentando importar o conteúdo csv em quadros de dados do pandas e depois convertê-lo em quadros de dados spark .... mas ele está mostrando um erro parecido com "Py4JJavaError" Ocorreu um erro ao chamar o28.applySchemaToPythonRDD. : Java.lang.RuntimeException: java.lang.RuntimeException: Não é possível instanciar org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient
neha
e meu código foi -> do pyspark import SparkContext do pyspark.sql import SQLContext import pandas como pd sqlc = SQLContext (sc) df = pd.read_csv (r'D: \ BestBuy \ train.csv ') sdf = sqlc.createDataFrame (df) ----> Erro
neha
11
Bem-vindo ao DataScience.SE! Edite sua postagem original em vez de adicionar comentários.
Emre
caminho do arquivo deve estar no HDFS só então u pode executar os dados
Prakash Reddy

Respostas:

13

"Como posso importar um arquivo .csv para os quadros de dados do pyspark?" -- Há muitas maneiras de fazer isso; o mais simples seria iniciar o pyspark com o módulo spark-csv da Databrick. Você pode fazer isso iniciando o pyspark com

pyspark --packages com.databricks:spark-csv_2.10:1.4.0

então você pode seguir as seguintes etapas:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('cars.csv')

O outro método seria ler o arquivo de texto como um arquivo rdd usando

myrdd = sc.textFile("yourfile.csv").map(lambda line: line.split(","))

Em seguida, transforme seus dados para que cada item esteja no formato correto para o esquema (ou seja, Ints, Strings, Floats, etc.). Você vai querer usar

>>> from pyspark.sql import Row
>>> Person = Row('name', 'age')
>>> person = rdd.map(lambda r: Person(*r))
>>> df2 = sqlContext.createDataFrame(person)
>>> df2.collect()
[Row(name=u'Alice', age=1)]
>>> from pyspark.sql.types import *
>>> schema = StructType([
...    StructField("name", StringType(), True),
...    StructField("age", IntegerType(), True)])
>>> df3 = sqlContext.createDataFrame(rdd, schema)
>>> df3.collect()
[Row(name=u'Alice', age=1)]

Referência: http://spark.apache.org/docs/1.6.1/api/python/pyspark.sql.html#pyspark.sql.Row

"Além disso, por favor me diga como posso importar o arquivo xlsx?" - arquivos do Excel não são usados ​​no "Big Data"; O Spark deve ser usado com arquivos ou bancos de dados grandes. Se você tem um arquivo do Excel com 50 GB de tamanho, está fazendo algo errado. O Excel nem seria capaz de abrir um arquivo desse tamanho; da minha experiência, qualquer coisa acima de 20 MB e Excel morre.

Jon
fonte
Eu acho que pode haver um problema com a abordagem RDD acima: os campos podem conter novas linhas (embora entre aspas duplas), a saber, tools.ietf.org/html/rfc4180#section-2 .
flow2k
você pode usar ferramentas para converter arquivos xlsx em csv (coisas como APIs numéricas ou de escritório aberto). então você pode fazer a ciência dos dados normalmente
vpathak
2

A seguir funcionou bem para mim:

from pyspark.sql.types import *
schema = StructType([StructField("name", StringType(), True),StructField("age", StringType(), True)]
pd_df = pd.read_csv("<inputcsvfile>")
sp_df = spark.createDataFrame(pd_df, schema=schema)
Saurabh Agrawal
fonte
1

Eu tenho no meu diretório local um arquivo 'temp.csv'. A partir daí, usando uma instância local, faço o seguinte:

>>> from pyspark import SQLContext
>>> from pyspark.sql import Row
>>> sql_c = SQLContext(sc)
>>> d0 = sc.textFile('./temp.csv')
>>> d0.collect()
[u'a,1,.2390', u'b,2,.4390', u'c,3,.2323']
>>> d1 = d0.map(lambda x: x.split(',')).map(lambda x: Row(label = x[0], number = int(x[1]), value = float(x[2])))
>>> d1.take(1)
[Row(label=u'a', number=1, value=0.239)]
>>> df = sql_c.createDataFrame(d1)
>>> df_cut = df[df.number>1]
>>> df_cut.select('label', 'value').collect()
[Row(label=u'b', value=0.439), Row(label=u'c', value=0.2323)]

Então d0 é o arquivo de texto bruto que enviamos para um spark RDD. Para criar um quadro de dados, você deseja dividir o csv e transformar cada entrada em um tipo de linha, como faço ao criar d1. O último passo é criar o quadro de dados do RDD.

jagartner
fonte
0

Você pode usar o pacote spark-csv do DataBricks que faz muitas coisas automaticamente, como cuidar do cabeçalho, usar caracteres de escape, inferir esquemas automáticos, etc. A partir do Spark 2.0, há uma função integrada para lidar com CSVs.

Jan van der Vegt
fonte