PySpark CSVをDataframeに読み込んで操作する方法

Question

私はpysparkにはかなり新しく、csvファイルとして保存される大きなデータセットを処理するためにこのファイルを使用しようとしています。 私はCSVファイルをsparkデータフレームに読み込み、いくつかのカラムを削除して、新しいカラムを追加したいと思います。 どうすればいいですか？

このデータをデータフレームに取り込む際に問題があります。これは、削減ステップでエラーTypeError: 'JavaPackage' object is not callableを生成

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

：これは私がこれまで持っているもののストリップダウンバージョンです。

これは可能ですか？データフレームを減らすというアイデアは、得られたデータをデータベース（Redshift、spark-redshiftパッケージを使用して）に書き込むことができるようにすることです。

unionAll()とmap()をpartial()と一緒に使ってみましたが、動作させることはできません。

私はAmazonのEMRで、spark-redshift_2.10:2.0.0、AmazonのJDBCドライバRedshiftJDBC41-1.1.17.1017.jarを使っています。

Answer 1

アップデート - 応答のコメントでもご質問：データフレームへのCSVから

読み取りデータを： あなただけCSVスパークデータフレームの中にファイルを読み込もうと思われます。

もしそうなら - 私の答えはここにあります：https://stackoverflow.com/a/37640154/5088142これをカバーしてください。

次のコードを使用して、 "ドロップ（COL）" https://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html

ドロップを使用して列を削除することができ

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 
spark.read.format("csv").option("header", "true").load("/path/to_csv.csv") 
spark.read.option("header", "true").csv("/path/to_csv.csv") 

ドロップカラム

火花データフレームにCSVを読んでください（列）

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

追加カラムあなたが "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)] 

注：スパークを使用することができる他の多くの機能を有している（例えば