spark-day1

Posted 2020-11-06 漫漫长路

#!/usr/bin/python
# -*- coding: utf_8 -*-
from pyspark import SparkConf, SparkContext
import os, time


if __name__ == ‘__main__‘:
    # Create SparkConf
    # sparkConf:是一个SparkConf的对象，用来初始化程序，配置应用，传递初始参数
    # setAppName：指定rdd名称；setMaster：指定运行的线程
    sparkConf = SparkConf()         .setAppName("python Spark WordCount")         .setMaster("local")

    # Create SparkContext
    # sc:是一个SparkContext的对象，代表了对计算集群的连接，用来访问Spark，创建RDD
    # 将sprkConf作为参数传递给SparkContext
    sc =SparkContext(conf=sparkConf)

    ‘‘‘
        创建RDD：
            方式一：从本地集和并行化创建
            方式二：从外部文件系统读取数据（HDFS）
    ‘‘‘
    # ----------第一种方式：从集合并行化创建RDD----------
    datas = ["hadoop spark", "spark hive spark sql", "spark hadoop sql spark"]

    # parallelize方法:接收一个list数据集作为参数，可以快速创建RDD；缺点：需要将数据集都加载到内存中
    rdd1 = sc.parallelize(datas)


    # ----------第二种方式：从外部存储中读取数据创建RDD----------

    # textfile方法：将文本文件读取为一个存储字符串的RDD
    rdd2 = sc.textFile(‘T_CAR_20150401.csv‘)

    # 转化操作：
    car_rdd = rdd2.filter(lambda x:‘FB5018‘ in x)
    car2_rdd = rdd2.filter(lambda x:‘FB5093‘in x)
    total_rdd = car_rdd.union(car2_rdd)

    # 行动操作：
    # count:获取个数
    # first:获取第一条数据
    print rdd1.count()
    print rdd1.first()
    print total_rdd.count()
    print total_rdd.first()

    # 休眠一段时间，为WEB UI进行监控
    time.sleep(100000)

    # SparkContext stop
    sc.stop()