<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Parse Json
Step1:  載入原始 RAW Data
Step2:  利用 LXML Parser 來分析文章結構
lxml.html urlparse 需在涵式內被import，以供RDD運算時使用
其他import python package的方法  Submitting Applications
Use spark-submit --py-files to add .py, .zip or .egg files to be distributed with your application. 
lxml.html.fromstring 的input為HTML string，回傳為可供 xpath 處理的物件
XPath syntax Ref_1, Ref_2
/   Selects from the root node
//   Selects all nodes in the document from the current node
@   Selects attributes
//@lang   Selects all attributes that are named lang
//title[@lang]   Selects all the title elements that have an attribute named lang
XPath usful Chrome plugin XPath Helper
Step3:  取出 Image Src 的列表
Step4:  統計 Image Src 的列表
Step5:  <span style="color
<ASSISTANT_TASK:>
Python Code:
def parseRaw(json_map):
    url = json_map['url']
    content = json_map['html']
    return (url,content)
Explanation: Parse Json
End of explanation
import json
import pprint
pp = pprint.PrettyPrinter(indent=2)
path = "./pixnet.txt"
all_content = sc.textFile(path).map(json.loads).map(parseRaw)
Explanation: 載入原始 RAW Data
End of explanation
def parseImgSrc(x):
    try:
        urls = list()
        import lxml.html
        from urlparse import urlparse
        node = lxml.html.fromstring(x)
        root =  node.getroottree()
        for src in root.xpath('//img/@src'):
            try :
                host = urlparse(src).netloc
                if '.' not in host : continue
                if host.count('.') == 1 : 
                    pass
                else: 
                    host = host[host.index('.')+1:]
                urls.append('imgsrc_'+host)
            except :
                print "Error Parse At:" , src
            
        for src in root.xpath('//input[@src]/@src'):
            try :
                host = urlparse(src).netloc
                if '.' not in host : continue
                if host.count('.') == 1 : 
                    pass
                else: 
                    host = host[host.index('.')+1:]
                urls.append('imgsrc_'+host)
            except :
                print "Error parseImgSrc At:" , src
        
    except :
        print "Unexpected error:", sys.exc_info()
    return  urls
all_content.map(lambda x: x[1]).first()[:100]
Explanation: 利用 LXML Parser 來分析文章結構
lxml.html urlparse 需在涵式內被import，以供RDD運算時使用
其他import python package的方法  Submitting Applications
Use spark-submit --py-files to add .py, .zip or .egg files to be distributed with your application. 
lxml.html.fromstring 的input為HTML string，回傳為可供 xpath 處理的物件
XPath syntax Ref_1, Ref_2
/   Selects from the root node
//   Selects all nodes in the document from the current node
@   Selects attributes
//@lang   Selects all attributes that are named lang
//title[@lang]   Selects all the title elements that have an attribute named lang
XPath usful Chrome plugin XPath Helper
End of explanation
image_list = all_content.map(lambda x :parseImgSrc(x[1]))
pp.pprint(image_list.first()[:10])
Explanation: 取出 Image Src 的列表
End of explanation
img_src_count = all_content.map(
    lambda x :parseImgSrc(x[1])).flatMap(
    lambda x: x).countByValue()
for i in img_src_count:
    print i , ':' , img_src_count[i]
Explanation: 統計 Image Src 的列表
End of explanation
from operator import add
all_content.map(
    lambda x :parseImgSrc(x[1])).flatMap(lambda x: x).map(lambda x: (x,1)).reduceByKey(add).sortBy(
    lambda x: x[1], ascending=False).collect()
Explanation: <span style="color: blue">請使用　reduceByKey , sortBy 來計算出 img src 排行榜</span>
請參照以下文件
[http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD]
幾種RDD sorting的方式
針對key值排序
使用 sortByKey
sc.parallelize(tmp).sortByKey(True, 1).collect()
使用 sortBy
sc.parallelize(tmp).sortBy(lambda x: x[0]).collect()
使用 takeOrdered
sc.parallelize(tmp).takeOrdered(10, lambda s: -1 * s[0])
針對value排序
使用 sortBy
sc.parallelize(tmp).sortBy(lambda x: x[1]).collect()
使用 takeOrdered
sc.parallelize(tmp).takeOrdered(10, lambda s: -1 * s[1])
takeOrdered()使用方式
sort by keys (ascending): RDD.takeOrdered(num, key = lambda x: x[0])
sort by keys (descending): RDD.takeOrdered(num, key = lambda x: -x[0])
sort by values (ascending): RDD.takeOrdered(num, key = lambda x: x[1])
sort by values (descending): RDD.takeOrdered(num, key = lambda x: -x[1])
End of explanation
<END_TASK>