RDD(resilient distributed dataset, 彈性分布式數(shù)據(jù)集)

RDD的特點(diǎn)

1、分布式
2福荸、只讀
3蕴坪、血緣關(guān)系
4、緩存（默認(rèn)緩存到內(nèi)存）
5、從外部文件創(chuàng)建
6背传、從父RDD生成子RDD
7呆瞻、支持本地磁盤文件
8、支持整個(gè)目錄径玖、多文件痴脾、通配符
9、支持壓縮文件
10挺狰、支持HDFS
11明郭、容錯(cuò)

RDD核心屬性

調(diào)度和計(jì)算都依賴于這五個(gè)屬性

1. 分區(qū)列表
   分布在哪臺(tái)機(jī)器上
2. 依賴列表
   即血緣關(guān)系，需要記錄下來丰泊，因?yàn)樯婕暗饺蒎e(cuò)

• 寬窄依賴

• 窄依賴：一（父）對(duì)一（子）映射

  
  
  image.png

• 寬依賴：多（父）對(duì)一（子）映射（會(huì)影響到后面的執(zhí)行計(jì)劃）
  會(huì)發(fā)生一種Shuffle操作----一種數(shù)據(jù)的傳遞（由于下游的RDD需要把上游的好幾個(gè)task的結(jié)果拿過來會(huì)涉及到數(shù)據(jù)的傳輸）

  
  
  image.png

• Shuffle

  
  
  image.png

3. Compute函數(shù)薯定，用于計(jì)算RDD各分區(qū)的值
4. 分區(qū)策略（可選）
   數(shù)據(jù)是如何分布到各個(gè)節(jié)點(diǎn)上的
5. 優(yōu)先位置列表（可選，HDFS實(shí)現(xiàn)數(shù)據(jù)本地化瞳购，避免數(shù)據(jù)移動(dòng)）

例

sc = spark.sparkContext
rdd = sc.textFile("d:/app.log")   #新建rdd
rdd.collect()  #查看rdd內(nèi)容

rdd = sc.textFile("d:/app.log,d:/Sid_SAS_20180430.txt")  #讀取多個(gè)文件的話话侄，用逗號(hào)隔開
rdd.collect()

還可以讀取壓縮文件

rdd = sc.textFile("d:/1-1G113144611.zip")
rdd.count() #查看有多少行

集合并行化

x = [1,2,3,4,5]
type(x)
rdd2 = sc.parallelize(x)  #將數(shù)據(jù)轉(zhuǎn)化為RDD
rdd2.collect()
type(rdd2)   #查看是否轉(zhuǎn)化為RDD

RDD支持哪些運(yùn)算

1 Transformations

• 輸入RDD，輸出RDD
• 延遲執(zhí)行
• 常用的Transformation

2 Actions

• 輸入RDD地粪，輸出非RDD
• 立即執(zhí)行
• 常用的Action

from pyspark.sql import SparkSession
master = 'local'
spark = SparkSession.builder. \
        appName('test'). \
        master(master). \
        getOrCreate()
sc = spark.sparkContext
filename = 'file:///tmp/README.md'
logData = sc.textFile(filename)
wordsRDD = logData. \
    flatMap(lambda x:x.split(" ")). \
    map(lambda x:(x,1)). \
    reduceByKey(lambda x,y:x+y)

out_filenmae = 'result'
wordsRDD.saveAsTextFile(out_filename)
words = wordsRDD.collect()
spark.stop()

作業(yè)

1.從身份證號(hào)中提取年齡

2.從身份證號(hào)中提取性別

3.從身份證號(hào)中提取星座

啟動(dòng)Pyspark

from pyspark.sql import SparkSession
form datetime import date
spark = SparkSession.builder.appName('CardAnalysis').getOrCreate()
sd = spark.sparkContext

###讀取文件
logData = sc.textFile('F:/pyspark/data/cardnum.txt')
###讀取身份證號(hào)
cardRDD = logData.map(lambda x:x.split(',')).map(lambda x:x[1])
###1 取身份證7-14位并計(jì)算年齡
birthdayRDD = cardRDD.map(lambda x:date(int(x[6:10]), int(x[10:12]), int(x[12:14])))
ageRDD = birthdayRDD.map(lambda x:int((date.today()-x).days/365))
age_list = ageRDD.collect()
print(age_list)

###2 取身份證第17位并計(jì)算性別
genderRDD = cardRDD.map(lambda x: 'Male' if int(x[17])%2 else 'Female')
gender_list = genderRDD.collect()
print(gender_list)

###3 取身份證11-14位并計(jì)算星座
定義計(jì)算星座函數(shù)
def getConstellation(month,day):
    n = (u'摩羯座', u'水瓶座', u'雙魚座', u'白羊座', u'金牛座', u'雙子座', u'巨蟹座', 
         u'獅子座', u'處女座', u'天秤座', u'天蝎座', u'射手座')
    d = ((1,20), (2,19), (3,21), (4,21), (5,21), (6,22), (7,23),(8,23), (9,23),(10,23),(11,23),(12,23))
    return n[len(list(filter(lambda y: y<=(month,day),d)))%12]

constellationRDD = cardRDD.map(lambda x:getConstellation(int(x[10:12]),int(x[12:14])))
constellation_list = constellationRDD.collect()
print(constellation_list)