Spark 之 PySpark

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# FileName ：01_word_count.py
# Author   ：zheng xingtao
# Date     ：2022/3/11 13:35


import os

from pyspark import SparkConf, SparkContext

# 设置环境变量，解决无法找到Java
os.environ['JAVA_HOME'] = '/hadoop/software/jdk1.8.0_241'


def work_count():
    """
    单词计数，读取HDFS上的words.txt文件
    :return: 对其内部的单词统计出现的数量
    """

    conf = SparkConf().setMaster("local[*]").setAppName("word_count")

    # 通过 SparkConf 对象构建 SoarkContext 对象
    sc = SparkContext(conf=conf)

    # 读取文件
    # HDFS文件
    # file_count = sc.textFile("hdfs://node-211:8020/zheng/words.txt")
    # 本地文件
    file_count = sc.textFile("./data/input/words.txt")


    # 将单词进行切割，得到一个存储全部单词的集合对象
    words_count = file_count.flatMap(lambda line: line.split(" "))

    # 将单词转换成元组对象，key-单词，value-数字
    words_with_one_count = words_count.map(lambda x: (x, 1))

    # 将元组的value 按照key来分组，对所有的value执行聚合操作（相加）
    result_rdd = words_with_one_count.reduceByKey(lambda a, b: a + b)

    #  通过collect方法手机RDD的数据输出
    print(result_rdd.collect())


if __name__ == '__main__':
    work_count()

# 以客户端提交

# 在Local下执行
bin/spark-submit --master local[*] /root/project/PySpark/01_word_count.py

# 在yarn中执行
bin/spark-submit --master yarn /root/project/PySpark/01_word_count.py