1. #!/usr/bin/env python
    2. # coding: utf-8
    4. # In[1]:
    7. # rdd算子
    10. # In[2]:
    13. from pyspark import SparkConf,SparkContext
    14. conf = SparkConf().setAppName("test").setMaster("local[*]")
    17. # In[3]:
    20. sc = SparkContext(conf = conf)
    23. # In[4]:
    26. data = [1,2,3,4,5,6,7,8,9]
    27. rdd =sc.parallelize(data, numSlices=3)
    28. print(rdd.collect())
    31. # In[6]:
    34. print(rdd.getNumPartitions())  # 获取分区数
    37. # In[7]:
    40. # rdd 读取文件
    41. #小文件较多用wholeTextFile
    42. rdd = sc.textFile("hdfs://bipcluster04/bip/external_data/vipreco/tess_official/entity/20220408/ads_rec_mer_itemRT_v4_v3")
    44. # rdd算子
    45. transformation算子 ：返回类型还是rdd
    46. &
    47. action算子：返回类型不是rdd
    48. # In[ ]:
    51. ## wordcount 计算实例
    54. # In[13]:
    57. localFile = sc.textFile("file:///home/mlp/notebooks/202204/helloWorld.txt")
    60. # In[14]:
    63. wordRdd = localFile.flatMap(lambda x: x.split(' '))
    66. # In[15]:
    69. wordsWithOneRdd = wordRdd.map(lambda x: (x,1))
    72. # In[16]:
    75. reduceRdd = wordsWithOneRdd.reduceByKey(lambda a,b: a + b).collect()
    78. # In[17]:
    81. print(reduceRdd)
    84. # In[ ]:
    90. # In[18]:
    93. #groupBy 算子：将rdd分组；
    96. # In[23]:
    99. rdd2= sc.parallelize([1,2,3,2,4,5,6,7,8])
    102. # In[28]:
    105. rdd3 =rdd2.groupBy(lambda num: 'even' if num % 2== 0 else 'odd')
    106. print(rdd3.map(lambda x: (x[0],list(x[1]))).collect())
    109. # In[30]:
    112. # filter过滤
    113. rdd2.filter(lambda x: x % 2 ==1).collect()
    114. #rdd3.filter(lambda x: x % 2 ==1).collect()  被使用
    117. # In[35]:
    120. #distinct 算子
    121. rdd2.distinct().collect()
    124. # In[36]:
    127. # union 将2个rdd合并成一个
    128. # 只合并不去重，不同类型也可以合并
    129. rdd4 = rdd2.distinct()
    130. rdd4.union(rdd2).collect()
    133. # In[41]:
    136. # join 对两个rdd进行join连接,类sql；
    137. # 只能用于二元数组；join leftOuterJoin RightOuterJoin
    139. x = sc.parallelize([(20220410,100),(20220411,120),(20220412,140)])
    140. y = sc.parallelize([(20220410,69123),(20220411,69124)])
    143. # In[42]:
    146. x.join(y).collect()
    149. # In[43]:
    152. x = sc.parallelize([(20220410,100,1),(20220411,120,1),(20220412,140,2)])
    153. y = sc.parallelize([(20220410,69123,2),(20220411,69124,1)])
    154. x.join(y).collect()
    157. # In[45]:
    160. # 两个集合的交集intersection
    161. # 返回两个rdd的交集，一个新的rdd
    162. #rdd.inserection(rdd2)
    164. rdd1 = sc.parallelize([('a',1),('b',1)])
    165. rdd2 = sc.parallelize([('a',1),('c',1)])
    166. inser_rdd = rdd1.intersection(rdd2)
    167. inser_rdd.collect()
    170. # In[46]:
    173. # glom() 将rdd加上嵌套，这个嵌套按照分区来进行
    174. # glom的作用是将同一个分区里的元素合并到一个array里
    175. data = [1,2,3,4,5,6,7,8,9]
    176. rdd =sc.parallelize(data, numSlices=2)
    177. rdd.glom().collect()
    180. # In[49]:
    183. # groupByKey() 针对kv型rdd，按照key进行分组
    184. rdd = sc.parallelize([1,2,2,4,3])
    185. rdd1 = rdd.map(lambda x: (x,1))
    186. rdd2 = rdd1.groupByKey()
    187. rdd2.map(lambda x: (x[0],list(x[1]))).collect()
    190. # In[50]:
    193. # sortBy 
    194. # 对rdd进行排序
    195. rdd1.collect()
    198. # In[51]:
    201. rdd1.sortBy(lambda x:x[0]).collect()
    204. # In[52]:
    207. rdd1.sortBy(lambda x:x[1]).collect()
    210. # In[53]:
    213. # sortByKey() 
    214. # 针对kv性rdd， 按照key进行排序
    215. #。RDD.sortByKey(ascending=True, numPartitions=None, keyfunc=<function RDD.<lambda>>)     
    218. # In[67]:
    221. rdd = sc.parallelize([('A',1),('b',1),('a',1),('D',1),('F',1),('b',1),('a',2),('b',3)],4)
    224. # In[68]:
    227. rdd.sortByKey().collect()
    230. # In[69]:
    233. rdd.sortByKey(ascending=False, numPartitions=3).collect()
    236. # In[70]:
    239. rdd.sortByKey(ascending=True, numPartitions=1, keyfunc= lambda key:str(key).lower()).collect()
    242. # In[ ]: