'''

构建sparkSession 和练习数据（RDD 和 KV rdd）

'''

spark = SparkSession.builder.appName("rdd_api_test") \

    .master("local[2]") \

    .getOrCreate()

sc = spark.sparkContext

rdd1 = sc.parallelize([1, 5, 60, 'a', 9, 'c', 4, 'z', 'f'])

rdd2 = sc.parallelize([('a', 6),

                       ('a', 1),

                       ('b', 2),

                       ('c', 5),

                       ('c', 8),

                       ('c', 11)])

'''

查看rdd元素 , 元素个数， KV对RDD中key的出现次数， 分区个数等常用api

'''

print(rdd2.collect())

print (rdd2.take(2))

print('amount of elements:', rdd2.count())

print('RDD count of key:', rdd2.countByKey())

print('RDD output as map:', rdd2.collectAsMap())

print('RDD number of partitions:', rdd2.getNumPartitions())

'''

数值型rdd ,常用统计函数， 最小，最大 ，平均 ， 标准差，方差

'''

rdd5 = sc.parallelize(range(100))

print('RDD Min:', rdd5.min())  # rdd 最小值

print('RDD Max:', rdd5.max())

print('RDD Mean:', rdd5.mean())

print('RDD Standard deviation:', rdd5.stdev())

print('RDD Variance:', rdd5.variance())

'''

从文件读取数据，并且去掉第一行列名，进行显示

数据源：nba.csv

'''

full_csv = sc.textFile('nba.csv')

header = full_csv.first()

print(full_csv.filter(lambda line: line != header).take(4))  # 去掉头行后，看效果

'''

从文档中读出文件头部，设置rdd模式，然后把RDD转化为df

数据源：本地文件：customerheaders.txt

数据样例：

id:string

full_name:string

    ...

'''

from pyspark.sql.types import IntegerType, DoubleType, StringType, StructType, StructField

header_list = sc.textFile('customerheaders.txt') \

    .map(lambda line: line.split(":")).collect()  # 返回数组，元素为[id string]...

def strToType(str):  # string 映射到 DataType

    if str == 'int':

        return IntegerType()

    elif str == 'double':

        return DoubleType()

    else:

        return StringType()

schema = StructType([StructField(t[0], strToType(t[1]), True) for t in header_list])  # 列表构造器 构造 StructType

for item in schema:

    print(item)

'''

对原始文件rdd中每一行进行规范化处理

数据举例：

7,Mohandas MacRierie,mmacrierie0@xrea.com,11/24/1990,Fliptune,-7.1309871,111.591546,

254,Rita Slany,rslany1@ucla.edu,8/7/1961,Yodoo,48.7068855,2.3371075,

'''

customers_rdd = sc.textFile('customers.txt')

'''

一行字符串的解析函数

逻辑：

    1，取出

'''

def parseLine(line):

    tokens = zip(line.split(","), header_list)  # 打包为元组的列表， [(7, [id string]),(Mohandas, [name string]), ... ]

    parsed_tokens = []

    for token in tokens:

        token_type = token[1][1]  # 取到数据的类型，然后转化该类型，并放入parsed_tokens

        print('token_type = ', token_type)

        if token_type == 'double':

            parsed_tokens.append(float(token[0]))

        elif token_type == 'int':

            parsed_tokens.append(int(token[0]))

        else:

            parsed_tokens.append(token[0])

    return parsed_tokens

records = customers_rdd.map(parseLine)  # 把文本字符串根据模式中对应的类型，转化为对象

for item in records.take(4):

    print(item)

df = spark.createDataFrame(records, schema)  # rdd --> df

print (df)

'''

其他一些API介绍：

rdd.foreach([FUNCTION]): 对每个元素执行函数

rdd.groupBy([CRITERA]):  分组聚合 like: ('a', 1) ('b',2) ('a', 3) --> ('a',Iterable(1,3)) ('b', 2)

rdd.subtract(rdd2):  做差集计算，元素在rdd中出现，没有在rdd2中出现

rdd.subtractByKey(rdd2): 同上，适用于KV rdd

rdd.sortBy([FUNCTION]): 自定义RDD元素排序

rdd.sortByKey():  按照key 进行排序，其中key的类型必须实现了排序逻辑

rdd.join(rdd2):   like : ('a', 1) ('b',2) ('a', 3) --> ('a',(1,3)) ('b', 2)

'''