Импорт содержимого файла CSV в фреймы данных pyspark

Как я могу импортировать файл .csv в фреймы данных pyspark? Я даже пытался прочитать CSV-файл в Pandas, а затем преобразовать его в искровой фрейм данных, используя createDataFrame, но он все еще показывает некоторую ошибку. Может ли кто-нибудь провести меня через это? Кроме того, скажите, пожалуйста, как я могу импортировать файл xlsx? Я пытаюсь импортировать CSV-контент в кадры данных Pandas, а затем преобразовать его в искровые кадры данных, но он показывает ошибку:

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

Мой код:

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) 

"Как я могу импортировать файл .csv в фреймы данных pyspark?" - Есть много способов сделать это; проще всего было бы запустить pyspark с помощью модуля Databrick's spark-csv. Вы можете сделать это, запустив pyspark с

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

тогда вы можете выполнить следующие шаги:

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

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

Другой метод - читать текстовый файл как rdd, используя

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

Затем преобразуйте данные так, чтобы каждый элемент имел правильный формат для схемы (т. Е. Ints, Strings, Floats и т. Д.). Вы захотите затем использовать

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

Ссылка: http://spark.apache.org/docs/1.6.1/api/python/pyspark.sql.html#pyspark.sql.Row

"Кроме того, скажите, пожалуйста, как я могу импортировать файл xlsx?" - файлы Excel не используются в «Больших данных»; Spark предназначен для использования с большими файлами или базами данных. Если у вас есть файл Excel размером 50 ГБ, значит, вы делаете что-то не так. Excel даже не сможет открыть файл такого размера; Исходя из моего опыта, все, что выше 20 МБ, и Excel умирает.

Следующее работает хорошо для меня:

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)

У меня в локальном каталоге есть файл 'temp.csv'. Оттуда, используя локальный экземпляр, я делаю следующее:

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

Таким образом, d0 - это необработанный текстовый файл, который мы отправляем в RDD. Чтобы вы могли создать фрейм данных, вы хотите разбить CSV на части и сделать каждую запись типом строки, как я делаю при создании d1. Последний шаг - сделать кадр данных из RDD.

Вы можете использовать пакет spark-csv от DataBricks, который автоматически делает для вас многое, например, заботится о заголовке, использует escape-символы, автоматически выводит схему и так далее. Начиная с Spark 2.0 существует встроенная функция для работы с CSV.