目錄

一、安裝Spark

二、Spark基本操作

一、安裝Spark

pip3 install pyspark

二、Spark基本操作

# 導入spark的SparkContext,SparkConf模塊
from pyspark import SparkContext, SparkConf
# 導入os模塊
import os
# 設(shè)置PYSPARK的python環(huán)境
os.environ['PYSPARK_PYTHON'] = "C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python310\\python.exe"# 構(gòu)建SparkConf()設(shè)置AppName和Master
conf = SparkConf().setAppName("myapp").setMaster("local")
# 構(gòu)建入口對象SparkContext對象sc
sc = SparkContext(conf=conf)# 定義列表對象
list1 = [1, 2, 3, 4, 5]
# 構(gòu)建list對象的RDD對象
rdd1 = sc.parallelize(list1)
print(rdd1.collect())# 定義元祖tuple對象
tuple1 = (1, 2, 3, 4, 5)
# 構(gòu)建tuple對象的RDD對象
rdd2 = sc.parallelize(tuple1)
print(rdd2.collect())# 定義字典對象
dict1 = {'a': 1, 'b': 2, 'c': 3}
# 構(gòu)建dict對象的RDD對象
rdd3 = sc.parallelize(dict1)
print(rdd3.collect())# 定義集合對象
set1 = {1, 2, 3, 4, 5}
# 構(gòu)建集合對象的RDD對象
rdd4 = sc.parallelize(set1)
print(rdd4.collect())# 定義字符串對象
string1 = "Hello, World!"
# 構(gòu)建字符串對象的RDD對象
rdd5 = sc.parallelize(string1)
print(rdd5.collect())# 通過textFile方法構(gòu)建RDD對象
# data.txt文件內(nèi)容如下：
# python java python python java
# java python C pascal java
# C java python pascal basic
# basic C java python python
# C C java basic java
rdd6 = sc.textFile('./data.txt')
print(rdd6.collect())# 使用flatMap算子將rdd6中的元素進行解構(gòu)拆分
rdd7 = rdd6.flatMap(lambda x: x.split(' '))
print(rdd7.collect())# 使用map算子將rdd7中的元素進行結(jié)構(gòu)轉(zhuǎn)換
# 構(gòu)造成(x,1)的元祖
rdd8 = rdd7.map(lambda x:(x,1))
print(rdd8.collect())# 使用reduceByKey算子對rdd8中的元素進行聚合
rdd9 = rdd8.reduceByKey(lambda x,y:x+y)
print(rdd9.collect())# 使用sortBy算子對rdd9中的元素進行排序
rdd10 = rdd9.sortBy(lambda x:x[1],ascending=False,numPartitions=1)
print(rdd10.collect())# 使用filter算子對rdd10中的元素進行過濾
rdd11 = rdd10.filter(lambda x:x[0] == 'java')
print(rdd11.collect())# def add(x):
#     return x + 1
#
# rdd = rdd1.map(add).map(lambda x: x * 10)
#
# print(rdd.collect())# rdd = sc.parallelize(['a', 'b', 'd', 'd'])
#
# rdd2 = rdd.map(lambda x:(x,1))
#
# print(rdd2.collect())
#
# rdd3 = rdd2.reduceByKey(lambda x,y:x+y)
#
# print(rdd3.collect())
#
# rdd4 = rdd2.countByKey()
#
# print(rdd4)sc.stop()