PySpark Comment lire CSV dans Dataframe, et les manipuler

Je suis assez nouveau à pyspark et suis en train de l'utiliser pour traiter un grand jeu de données qui est enregistré en tant que fichier csv.
J'aimerais lire fichier CSV dans spark dataframe, supprimer des colonnes, et d'ajouter de nouvelles colonnes.
Comment dois-je faire?

J'ai de la difficulté à obtenir ces données dans un dataframe. C'est une version allégée de ce que j'ai à ce jour:

def make_dataframe(data_portion, schema, sql):
    fields = data_portion.split(",")
    return sql.createDateFrame([(fields[0], fields[1])], schema=schema)

if __name__ == "__main__":
    sc = SparkContext(appName="Test")
    sql = SQLContext(sc)

    ...

    big_frame = data.flatMap(lambda line: make_dataframe(line, schema, sql))
                .reduce(lambda a, b: a.union(b))

    big_frame.write \
        .format("com.databricks.spark.redshift") \
        .option("url", "jdbc:redshift://<...>") \
        .option("dbtable", "my_table_copy") \
        .option("tempdir", "s3n://path/for/temp/data") \
        .mode("append") \
        .save()

    sc.stop()

Cela produit une erreur TypeError: 'JavaPackage' object is not callable à la réduire étape.

Est-il possible de faire cela? L'idée de réduire à un dataframe est d'être capable d'écrire les données à une base de données (Redshift, à l'aide de la spark-décalage vers le rouge).

J'ai aussi essayé d'utiliser unionAll(), et map() avec partial(), mais ne peut pas le faire fonctionner.

Je suis en cours d'exécution ce sur Amazon EMR, avec spark-redshift_2.10:2.0.0, et Amazon pilote JDBC RedshiftJDBC41-1.1.17.1017.jar.

Qu'est-ce que l'entrée (champs CSV), et ce qui devrait être la sortie? Vous devrez peut-être revoir votre code. Il me semble que vous pouvez éviter la nécessité de créer des bases de données-cadres et de l'union....
Le csv est juste des numéros qui peuvent être organisées dans le schéma de la dataframe. Je sais que je peux facilement l'enregistrer comme un EDR agréable et efficace, mais si je fais ça je ne peux pas écrire à un redshift de la base de données (autant que je sache), ce qui est le but ultime.
ce que j'ai essayé de faire allusion - je suppose que vous pouvez le résoudre en utilisant une étincelle de données-image, sans avoir besoin de plusieurs données-images + de l'union. encore une fois - ce que l'algorithme sont que vous essayez d'utiliser? quel est le résultat attendu?
J'ai ajouté l'étincelle redshift la fonction d'écriture, je l'utiliserais avec le dataframe. Cela permettrait d'ajouter les données à une table existante. Vous avez peut-être raison, je ne sais pas si il y a une meilleure façon de renvoyer les données à partir de la fonction map (make_dataframe) pour me permettre de créer un cadre?

OriginalL'auteur Tim B | 2016-10-30

  1. 9

    Mise à jour - répondre aussi à votre question dans les commentaires:

    Lire des données à partir d'un fichier CSV à dataframe:
    Il semble que vous essayez seulement de lire fichier CSV dans une étincelle dataframe.

    Si soi - ma réponse ici: https://stackoverflow.com/a/37640154/5088142.

    Le code suivant devrait lire CSV dans une étincelle de données-cadre

    import pyspark
sc = pyspark.SparkContext()
sql = SQLContext(sc)

df = (sql.read
         .format("com.databricks.spark.csv")
         .option("header", "true")
         .load("/path/to_csv.csv"))

//these lines are equivalent in Spark 2.0 - using [SparkSession][1]
from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

spark.read.format("csv").option("header", "true").load("/path/to_csv.csv") 
spark.read.option("header", "true").csv("/path/to_csv.csv")

    chute de la colonne

    vous pouvez déposer colonne à l'aide de "drop(col)"
    https://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html

    drop(col)

    Returns a new DataFrame that drops the specified column.
Parameters: col – a string name of the column to drop, or a Column to drop.

>>> df.drop('age').collect()
[Row(name=u'Alice'), Row(name=u'Bob')]

>>> df.drop(df.age).collect()
[Row(name=u'Alice'), Row(name=u'Bob')]

>>> df.join(df2, df.name == df2.name, 'inner').drop(df.name).collect()
[Row(age=5, height=85, name=u'Bob')]

>>> df.join(df2, df.name == df2.name, 'inner').drop(df2.name).collect()
[Row(age=5, name=u'Bob', height=85)]

    ajouter une colonne
    Vous pouvez utiliser "withColumn"
    https://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html

    withColumn(colName, col)

    Returns a new DataFrame by adding a column or replacing the existing column that has the same name.
Parameters: 

    colName – string, name of the new column.
    col – a Column expression for the new column.

>>> df.withColumn('age2', df.age + 2).collect()
[Row(age=2, name=u'Alice', age2=4), Row(age=5, name=u'Bob', age2=7)]

    Note: spark a beaucoup d'autres fonctions qui peuvent être utilisées (par exemple, vous pouvez utiliser le bouton "select" au lieu de "drop")

    Merci pour cette. Dans le cadre du traitement j'ai besoin de supprimer une colonne à partir de la trame de données (qui est inclus dans le fichier CSV en cours de lecture), et d'ajouter une nouvelle colonne à l'image contenant un ID avant d'écrire les données de redshift. Pourrais-je utiliser la méthode que vous avez ici pour lire les données, et ensuite effectuer ce traitement sur le dataframe?
    J'ai mis à jour ma réponse , pour répondre à vous questions dans votre commentaire
    si j'ai répondu à votre question, merci de l'accepter.
    C'est super, merci. Dois-je appeler en pcv()? Je voudrais éviter de soumettre toutes les données vers le conducteur si possible. Je me demandais si je pouvais ajouter la fonction write() de commande, par exemple, la withColumn commande?
    mise à jour de ma réponse

    OriginalL'auteur Yaron