spark教程(五)-实战

1,108.99564,34.33999999,1
2,108.99564,34.3399087138298,1
3,108.99564,34.3398174376596,1
4,108.99564,34.3397261614894,1
5,108.99564,34.3396348853192,1
6,108.99564,34.3395436091489,1
7,108.99564,34.3394523329787,1
8,108.99564,34.3393610568085,1

from __future__ import division
from pyspark import SparkContext

max_lng = 136
min_lng = 73
max_lat = 54
min_lat = 3
lng_stage = 1000
lat_stage = 1000

lng_step=(max_lng - min_lng) / (lng_stage)
lat_step=(max_lat - min_lat) / (lat_stage)

def mymap(data):
    # print(data, 1111111111111111111111111111111111)
    return data.split(‘,‘)

def mygroup(data):
    # 对经纬度分区，打标签
    # print(data[1], type(data[1]))
    label_lng = round(float(data[1]) / lng_step, 1)
    label_lat = round(float(data[2]) / lat_step, 1)
    return (label_lng, label_lat)

def mapkey(data):
    # 把标签还原成经纬度
    return data[0][0] * lng_step, data[0][1] * lat_step, data[1]


# sc = SparkContext(‘local‘, ‘gpsfreq‘)       # 本地模式
sc = SparkContext(‘spark://hadoop10:7077‘, ‘gpsfreq‘)     # standalone 模式
rdd = sc.textFile(‘dwd.csv‘, 100)       # standalone 模式 读取本地文件，必须在每个节点上都有这个文件，且路径一致

# print rdd.map(mymap).collect()
print rdd.map(mymap).groupBy(mygroup).mapValues(len).map(mapkey).collect()

[(111.8943, 32.629799999999996, 11), (109.2861, 34.2006, 42), (109.0467, 35.2155, 77), (111.72420000000001, 33.1755, 11), (111.636, 32.8542, 7)]