scala读取本地hdfs_Spark本地文件读写

Spark 对很多种文件格式的读取和保存方式都很简单。从诸如文本文件的非结构化的文件， 到诸如 JSON 格式的半结构化的文件，再到诸如 SequenceFile 这样的结构化的文件， Spark 都可以支持（见下表） 。 Spark 会根据文件扩展名选择对应的处理方式。这一过程是封装 好的，对用户透明。

一、读取

1.1 读取txt文件

只需要使用文件路径作为参数调用 SparkContext 中的 textFile() 函数，就可以读取一个文本文件，读取的一行作为一个元素。如果要控制分区数的话，可以指定 minPartitions 。

read_txt = sc.textFile("file:///home/xiaohuzi/spark_exercise/data/testfile.txt")
print read_txt.collect()
[u'itemtratio', u'longitudet96.94%', u'latitudet96.94%', u'district_codet100.00%', u'district_namet99.94%', u'city_codet100.00%', u'city_namet100.00%', u'bizcircle_idt89.12%', u'bizcircle_namet74.75%']

1.2 读取csv文件

csv 文件的读取和 txt 相同。

read_csv = sc.textFile("file:///home/xiaohuzi/spark_exercise/data/testfile.csv")
print read_csv.collect()
[u'item,ratio', u'longitude,96.94%', u'latitude,96.94%', u'district_code,100.00%', u'district_name,99.94%', u'city_code,100.00%', u'city_name,100.00%', u'bizcircle_id,89.12%', u'bizcircle_name,74.75%']

1.3 读取json文件

将数据作为文本文件读取， 然后对 JSON 数据进行解析， 这样的方法可以在所有支持的 编程语言中使用。 这种方法假设文件中的每一行都是一条 JSON 记录。 如果你有跨行的 JSON 数据，你就只能读入整个文件，然后对每个文件进行解析。

import json
read_json = sc.textFile("file:///home/xiaohuzi/spark_exercise/data/testfile.json")
data = read_json.map(lambda x: json.loads(x))
print data.collect()
[{u'id': u'1234', u'name': u'xiaohuzi'}, {u'id': u'001123', u'name': u'hustand'}, {u'id': u'97213', u'name': u'stanford'}]

1.4 读取HDFS文件

读取HDFS文件和前面的格式相同，只是路径有一些变化

read_hdfs = sc.textFile("hdfs/testfile")
print read_hdfs.take(3)

二、保存

2.1保存到本地txt

保存到本地一般是比较小的数据，可以先 collect 再保存到本地，如果直接保存存储的会是各个分区的数据。

# 直接保存，会保存成文件夹
read_txt.saveAsTextFile("file:///home/xiaohuzi/spark_exercise/data/save_testfile.txt")

# 先 collect 在保存为 txt
save_to_csv(read_txt.collect())

2.2 保存到本地json

这个时候需要先对 json 进行 dumps 处理。

save_json = read_json.map(lambda x : json.dumps(x))
save_json.saveAsTextFile("file:///home/xiaohuzi/spark_exercise/data/save_testfile.json")