PySpark实现文本去重计数：两种方法详解

本文将介绍如何使用PySpark对文本文件进行去重计数，分别使用SparkContext和SparkSession两种方法实现。

方法一：使用SparkContextpythonfrom pyspark import SparkContext

创建SparkContextsc = SparkContext.getOrCreate()

读取文本文件lines = sc.textFile('data01.txt')

按逗号分割每一行，并取第二个元素res = lines.map(lambda x: x.split(',')).map(lambda x: x[1])

去重distinct_res = res.distinct()

统计去重后的元素个数print(distinct_res.count())

关闭SparkContextsc.stop()

代码解释:

1. 首先，导入SparkContext类，并创建一个SparkContext实例。2. 使用textFile()方法读取文本文件data01.txt，并将每一行作为字符串存储在RDD中。3. 使用map()方法对RDD进行两次转换： - 第一次map()使用lambda表达式将每一行按逗号,分割成一个列表。 - 第二次map()同样使用lambda表达式，从分割后的列表中取出第二个元素（索引为1）。4. 使用distinct()方法对RDD进行去重操作，得到只包含唯一元素的新RDD。5. 使用count()方法统计去重后的RDD中元素个数，并将结果打印输出。6. 最后，使用stop()方法关闭SparkContext。

方法二：使用SparkSessionpythonfrom pyspark.sql import SparkSession

创建SparkSessionspark = SparkSession.builder.getOrCreate()

读取文本文件lines = spark.read.text('data01.txt').rdd

按逗号分割每一行，并取第二个元素res = lines.map(lambda x: x.value.split(',')).map(lambda x: x[1])

去重distinct_res = res.distinct()

统计去重后的元素个数print(distinct_res.count())

关闭SparkSessionspark.stop()

代码解释:

1. 首先导入 SparkSession 类，并创建一个 SparkSession 实例。2. 使用 spark.read.text() 方法读取文本文件 data01.txt，并使用 .rdd 属性将其转换为 RDD。3. 后续步骤与方法一相同，都是对 RDD 进行操作：使用 map() 方法分割字符串、提取元素、使用 distinct() 方法去重以及使用 count() 方法计数。4. 最后关闭 SparkSession。

总结:

两种方法都可以实现对文本文件去重计数的功能。方法一使用 SparkContext，更加底层；方法二使用 SparkSession，更加简洁易用。建议根据实际情况选择合适的方法。