Comment faire de la première ligne comme en-tête lors de la lecture d'un fichier dans PySpark et de la convertir vers les Pandas Dataframe

Je suis à la lecture d'un fichier dans PySpark et formant la rdd. Je puis convertir normal dataframe puis à pandas dataframe. Le problème que je rencontre est qu'il y a en-tête de ligne dans mon fichier d'entrée, et je veux faire ce que l'en-tête de dataframe colonnes, mais elles sont lus dans une rangée supplémentaire et non comme en-tête. Voici mon code actuel:

def extract(line):
    return line


input_file = sc.textFile('file1.txt').zipWithIndex().filter(lambda (line,rownum): rownum>=0).map(lambda (line, rownum): line)

input_data = (input_file
    .map(lambda line: line.split(";"))
    .filter(lambda line: len(line) >=0 )
    .map(extract)) # Map to tuples

df_normal = input_data.toDF()
df= df_normal.toPandas()

Maintenant quand je regarde les df puis la ligne d'en-tête de fichier texte devient la première ligne de dataframe et il y a en-tête supplémentaire dans df avec 0,1,2... comme en-tête. Comment puis-je faire la première ligne comme en-tête?

sans un échantillon de la dataframe de travailler avec, je pense que vous pouvez simplement utiliser df_normal.toPandas('header'=1) . Ou quelle que soit la ligne qui contient l'en-tête que vous voulez
Pourquoi utiliser Étincelle à tous ici? Si l'on suppose que les données s'inscrit (en ignorant les lignes vides) sur une machine locale, il est juste un gaspillage de temps et de ressources.

OriginalL'auteur user2966197 | 2016-01-16

Il ya un couple de façons de le faire, en fonction de la structure exacte de vos données. Puisque vous ne lui donnez pas de détails, je vais essayer de le montrer à l'aide d'un fichier de données nyctaxicab.csv que vous pouvez télécharger.

Si votre fichier est en csv format, vous devez utiliser le spark-csv paquet, fournis par Databricks. Pas besoin de télécharger de façon explicite, il suffit d'exécuter pyspark comme suit:

$ pyspark --packages com.databricks:spark-csv_2.10:1.3.0

et puis

>>> from pyspark.sql import SQLContext
>>> from pyspark.sql.types import *
>>> sqlContext = SQLContext(sc)

>>> df = sqlContext.read.load('file:///home/vagrant/data/nyctaxisub.csv', 
                      format='com.databricks.spark.csv', 
                      header='true', 
                      inferSchema='true')

>>> df.count()
249999

Le fichier de 250 000 lignes, y compris l'en-tête, de sorte à 249 999 est le nombre exact d'enregistrements réels. Voici le schéma, tel que déduit automatiquement par le paquet:

>>> df.dtypes
[('_id', 'string'),
 ('_rev', 'string'),
 ('dropoff_datetime', 'string'),
 ('dropoff_latitude', 'double'),
 ('dropoff_longitude', 'double'),
 ('hack_license', 'string'),
 ('medallion', 'string'),
 ('passenger_count', 'int'),
 ('pickup_datetime', 'string'),
 ('pickup_latitude', 'double'),
 ('pickup_longitude', 'double'),
 ('rate_code', 'int'),
 ('store_and_fwd_flag', 'string'),
 ('trip_distance', 'double'),
 ('trip_time_in_secs', 'int'),
 ('vendor_id', 'string')]

Vous pouvez voir plus de détails dans mon blog post.

Si, pour quelque raison que ce soit, vous ne pouvez pas utiliser le spark-csv package, vous devez soustraire la première ligne de données, puis de les utiliser pour construire votre schéma. Voici l'idée générale, et de nouveau, on peut trouver un exemple complet avec les détails du code dans un autre post de blog de la mine:

>>> taxiFile = sc.textFile("file:///home/ctsats/datasets/BDU_Spark/nyctaxisub.csv")
>>> taxiFile.count()
250000
>>> taxiFile.take(5)
[u'"_id","_rev","dropoff_datetime","dropoff_latitude","dropoff_longitude","hack_license","medallion","passenger_count","pickup_datetime","pickup_latitude","pickup_longitude","rate_code","store_and_fwd_flag","trip_distance","trip_time_in_secs","vendor_id"',
 u'"29b3f4a30dea6688d4c289c9672cb996","1-ddfdec8050c7ef4dc694eeeda6c4625e","2013-01-11 22:03:00",+4.07033460000000E+001,-7.40144200000000E+001,"A93D1F7F8998FFB75EEF477EB6077516","68BC16A99E915E44ADA7E639B4DD5F59",2,"2013-01-11 21:48:00",+4.06760670000000E+001,-7.39810790000000E+001,1,,+4.08000000000000E+000,900,"VTS"',
 u'"2a80cfaa425dcec0861e02ae44354500","1-b72234b58a7b0018a1ec5d2ea0797e32","2013-01-11 04:28:00",+4.08190960000000E+001,-7.39467470000000E+001,"64CE1B03FDE343BB8DFB512123A525A4","60150AA39B2F654ED6F0C3AF8174A48A",1,"2013-01-11 04:07:00",+4.07280540000000E+001,-7.40020370000000E+001,1,,+8.53000000000000E+000,1260,"VTS"',
 u'"29b3f4a30dea6688d4c289c96758d87e","1-387ec30eac5abda89d2abefdf947b2c1","2013-01-11 22:02:00",+4.07277180000000E+001,-7.39942860000000E+001,"2D73B0C44F1699C67AB8AE322433BDB7","6F907BC9A85B7034C8418A24A0A75489",5,"2013-01-11 21:46:00",+4.07577480000000E+001,-7.39649810000000E+001,1,,+3.01000000000000E+000,960,"VTS"',
 u'"2a80cfaa425dcec0861e02ae446226e4","1-aa8b16d6ae44ad906a46cc6581ffea50","2013-01-11 10:03:00",+4.07643050000000E+001,-7.39544600000000E+001,"E90018250F0A009433F03BD1E4A4CE53","1AFFD48CC07161DA651625B562FE4D06",5,"2013-01-11 09:44:00",+4.07308080000000E+001,-7.39928280000000E+001,1,,+3.64000000000000E+000,1140,"VTS"']

# Construct the schema from the header 
>>> header = taxiFile.first()
>>> header
u'"_id","_rev","dropoff_datetime","dropoff_latitude","dropoff_longitude","hack_license","medallion","passenger_count","pickup_datetime","pickup_latitude","pickup_longitude","rate_code","store_and_fwd_flag","trip_distance","trip_time_in_secs","vendor_id"'
>>> schemaString = header.replace('"','')  # get rid of the double-quotes
>>> schemaString
u'_id,_rev,dropoff_datetime,dropoff_latitude,dropoff_longitude,hack_license,medallion,passenger_count,pickup_datetime,pickup_latitude,pickup_longitude,rate_code,store_and_fwd_flag,trip_distance,trip_time_in_secs,vendor_id'
>>> fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(',')]
>>> schema = StructType(fields)

# Subtract header and use the above-constructed schema:
>>> taxiHeader = taxiFile.filter(lambda l: "_id" in l) # taxiHeader needs to be an RDD - the string we constructed above will not do the job
>>> taxiHeader.collect() # for inspection purposes only
[u'"_id","_rev","dropoff_datetime","dropoff_latitude","dropoff_longitude","hack_license","medallion","passenger_count","pickup_datetime","pickup_latitude","pickup_longitude","rate_code","store_and_fwd_flag","trip_distance","trip_time_in_secs","vendor_id"']
>>> taxiNoHeader = taxiFile.subtract(taxiHeader)
>>> taxi_df = taxiNoHeader.toDF(schema)  # Spark dataframe
>>> import pandas as pd
>>> taxi_DF = taxi_df.toPandas()  # pandas dataframe

Pour des raisons de concision, ici, toutes les colonnes finissent par être de type string, mais dans le post de blog je montre en détail et expliquer comment vous pouvez affiner souhaité types de données (et noms) pour des domaines spécifiques.

OriginalL'auteur desertnaut

La réponse la plus simple serait de définir header='true'

Par exemple:

df = spark.read.csv('housing.csv', header='true')

df = spark.read.option("header","true").format("csv").schema(myManualSchema).load("maestraDestacados.csv")

ou header=True

OriginalL'auteur naren

Vous devez vous connecter pour publier un commentaire.