<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 采用Spark处理OpenStreetMap的osm文件。
Spark DataFrame参考
Step1:  配置环境SparkConf和创建SparkContext运行环境对象。
Step2:  显示Spark的配置信息。
Step3:  Spark的文本RDD操作。
按照文本方式读取osm的json格式文件，将JSON字符串转为dict对象。
Step4:  从RDD中按照文本方式进行关键词查询。
Step5:  Spark的DataFrame操作。
使用SQL引擎直接生成Spark的DataFrame对象，支持查询等操作。
读取osm的node数据表。
Step6:  Spark DataFrame的 select() 操作。show()方法可以指定最多显示的记录数。
Step7:  读取osm的way表。
Step8:  查看way表中的数据。
Step9:  构建way的几何对象。
从way中的每一条记录生成NodeID的字符串列表，用于下一步查询node的坐标信息表。
Step10:  将多个way的node信息查询出来。
Step11:  将经纬度坐标转换为一个GeoJSON的几何对象表示，并保存回way的geometry字段。
Step12:  查找指定关键词。
自定义函数处理。
<ASSISTANT_TASK:>
Python Code:
from pprint import *
import pyspark
from pyspark import SparkConf, SparkContext
sc = None
print(pyspark.status)
Explanation: 采用Spark处理OpenStreetMap的osm文件。
Spark DataFrame参考: https://spark.apache.org/docs/1.3.0/sql-programming-guide.html#interoperating-with-rdds  
  by openthings@163.com，2016-4-23.      License: GPL, MUST include this header.  
说明：
使用sc.read.json()读取json文件（osm-all2json从osm转换而来），生成Spark的DataFrame对象。
查询从json文件创建的DataFrame对象，创建新的DataFrame。
读取way的nd索引（Node的ID），并构建way的geometry对象。
后续：
将数据保存到MongoDB/Hbase/HDFS等其它存储系统。
将数据进行分块，保存为分区域的DataFrame数据集合。
将DataFrame转换为GeoPandas.DataFrame，然后保存为shape files。
将DataFrame直接转换为GIScript.Dataset，然后保存为UDB files。
End of explanation
conf = (SparkConf()
         .setMaster("local")
         .setAppName("MyApp")
         .set("spark.executor.memory", "1g"))
if sc is None:
    sc = SparkContext(conf = conf)
print(type(sc))
print(sc)
print(sc.applicationId)
Explanation: 配置环境SparkConf和创建SparkContext运行环境对象。
End of explanation
print(conf)
conf_kv = conf.getAll()
pprint(conf_kv)
Explanation: 显示Spark的配置信息。
End of explanation
fl = sc.textFile("../data/muenchen.osm_node.json")
for node in fl.collect()[0:2]:
    node_dict = eval(node)
    pprint(node_dict)
Explanation: Spark的文本RDD操作。
按照文本方式读取osm的json格式文件，将JSON字符串转为dict对象。
End of explanation
lines = fl.filter(lambda line: "soemisch" in line)
print(lines.count())
print(lines.collect()[0])
Explanation: 从RDD中按照文本方式进行关键词查询。
End of explanation
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
nodeDF = sqlContext.read.json("../data/muenchen.osm_node.json")
#print(nodeDF)
nodeDF.printSchema()
Explanation: Spark的DataFrame操作。
使用SQL引擎直接生成Spark的DataFrame对象，支持查询等操作。
读取osm的node数据表。
End of explanation
nodeDF.select("id","lat","lon","timestamp").show(10,True)
#help(nodeDF.show)
Explanation: Spark DataFrame的 select() 操作。show()方法可以指定最多显示的记录数。
End of explanation
wayDF = sqlContext.read.json("../data/muenchen.osm_way.json")
wayDF.printSchema()
Explanation: 读取osm的way表。
End of explanation
wayDF.select("id","tag","nd").show(10,True)
Explanation: 查看way表中的数据。
End of explanation
def sepator():
    print("===============================================================")
#### 将给定way的nd对象的nodeID列表提取出来，并生成一个查询的过滤字符串。
def nodelist_way(nd_list):
    print("WayID:",nd_list["id"],"\tNode count:",len(nd_list["nd"]))
    ndFilter = "("
    for nd in  nd_list["nd"]:
        ndFilter = ndFilter + nd["ref"] + ","
    ndFilter = ndFilter.strip(',') + ")"
    print(ndFilter)
    return ndFilter
#### 根据way的节点ID从nodeDF中提取node信息，包含经纬度等坐标域。
def nodecoord_way(nodeID_list):
    nodeDF.registerTempTable("nodeDF")
    nodeset = sqlContext.sql("select id,lat,lon,timestamp from nodeDF where nodeDF.id in " + nodeID_list)
    nodeset.show(10,True)
Explanation: 构建way的几何对象。
从way中的每一条记录生成NodeID的字符串列表，用于下一步查询node的坐标信息表。
End of explanation
for wayset in wayDF.select("id","nd").collect()[4:6]:
    ndFilter = nodelist_way(wayset)
    nodecoord_way(ndFilter)
#pprint(nd_list["nd"])
#sepator()
Explanation: 将多个way的node信息查询出来。
End of explanation
relationDF = sqlContext.read.json("../data/muenchen.osm_relation.json")
#print(relationDF)
relationDF.printSchema()
relationDF.show(10,True)
Explanation: 将经纬度坐标转换为一个GeoJSON的几何对象表示，并保存回way的geometry字段。
End of explanation
def myFunc(s):
        words = s.split()
        return len(words)
#wc = fl.map(myFunc).collect()
wc = fl.map(myFunc).collect()
wc
#df = sqlContext.read.format("com.databricks.spark.xml").option("rowTag", "result").load("../data/muenchen.osm")
#df
Explanation: 查找指定关键词。
自定义函数处理。
End of explanation
<END_TASK>