Day15-Python基础学习之PySpark

PySpark

安装： pip install pyspark

# pySpark入门
from pyspark import SparkConf, SparkContext
# 创建SparkConf对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
# 基于SparkConf类对象创建SparkContext对象
sc = SparkContext(conf=conf)
# 打印spark运行版本
print(sc.version)
# 停止SparkContext对象的运行
sc.stop()
​
# 加载数据，即数据输入
conf1 = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc1 = SparkContext(conf=conf1)
# 通过parallelize方法将python对象加载到spark内，成为RDD对象
rdd1 = sc1.parallelize([1, 2, 3, 4, 5])
rdd2 = sc1.parallelize((1, 2, 3, 4, 5))
rdd3 = sc1.parallelize("abcdefghijklmnop")
rdd4 = sc1.parallelize({1, 2, 3, 4, 5})
rdd5 = sc1.parallelize({"key1": "value1", "key2": "value2"})
# 要查看rdd里的内容，需要使用collect方法
print(rdd1.collect())
# 注意字典打印出来只有key
sc1.stop()
​
# 读文件通过textFile方法，成为rdd对象
conf2 = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc2 = SparkContext(conf=conf2)
rdd = sc2.textFile("D:/test.txt")
print(rdd.collect())

# 数据计算map方法
from pyspark import SparkContext, SparkConf
import os
os.environ['PYSPARK_PYTHON'] = "D:/Study/Environment/Python/python.exe"
​
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
​
rdd = sc.parallelize([1, 2, 3, 4, 5])
# 通过map方法将全部数据乘10
def func(data):
    return data*10
rdd2 = rdd.map(func)
print(rdd2.collect())
​
rdd3 = rdd2.map(lambda x: x*10).map(lambda x: x+5)
print(rdd3.collect())