リスト項目を定義済みのデータ型RDD

Question

に変換します。実際には、clouderaのdataricksでApache Spark Pythonの作業領域で作業しています。アイデアはcsvを読み込み、各フィールドをフォーマットすることです。

したがって、最初のステップは、CSVファイルを読み込むことであった。

uber_parsed = uber.map(lambda lin:lin.split(",")) 
print (uber_parsed.first()) 

結果であった：

uber = sc.textFile("dbfs:/mnt/uber/201601/pec2/uber_curated.csv") 

次のステップは、値のリストに各行を変換することでした

[u'B02765', u'2015-05-08 19:05:00', u'B02764', u'262', u'Manhattan',u'Yorkville East'] 

しかし、次の値リストの各項目を次の形式に変換する必要があります。String、Date、String、Integer、Strin g、文字列。

[[u'B02765', u'2015-05-08 19:05:00', u'B02764', u'262', u'Manhattan', u'Yorkville East'], 
[u'B02767', u'2015-05-08 19:05:00', u'B02789', u'400', u'New York', u'Yorkville East']] 

誰かがそれを行う方法を知っていますか？

Answer 1

csvリーダーを使用できます。 Spark 1.xでは、外部依存関係（spark-csv）が必要です。

from pyspark.sql.types import * 

sqlContext.read.format("csv").schema(StructType([ 
    StructField("_1", StringType()), 
    StructField("_2", TimestampType()), 
    StructField("_3", StringType()), 
    StructField("_4", IntegerType()), 
    StructField("_5", StringType()), 
    StructField("_6", StringType()), 
])).load("dbfs:/mnt/uber/201601/pec2/uber_curated.csv").rdd 

または

sqlContext.read.format("csv").schema(StructType([ 
    StructField("_1", StringType()), 
    StructField("_2", DateType()), 
    StructField("_3", StringType()), 
    StructField("_4", IntegerType()), 
    StructField("_5", StringType()), 
    StructField("_6", StringType()), 
])).option("dateFormat", "yyyy-dd-MM HH:mm:ss").load(
    "dbfs:/mnt/uber/201601/pec2/uber_curated.csv" 
).rdd 

あなたは、記述フィールド名と（_1、_2 ... _n）を置き換えることができます。