下面我将详细讲解如何使用 Spark SQL 分别通过 SQL 模式和 DSL（Domain Specific Language）模式实现 WordCount 功能。
WordCount 是大数据处理中的经典案例，主要功能是统计文本中每个单词出现的次数。
准备工作
首先需要初始化 SparkSession，这是 Spark SQL 的入口点：
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, col# 创建 SparkSession
spark = SparkSession.builder \.appName("WordCountExample") \.master("local[*]") # 本地模式运行，生产环境可去掉.getOrCreate()
示例数据
我们使用一段简单的文本作为示例数据：
# 创建包含文本数据的 DataFrame
data = [("Hello Spark SQL",), ("Hello DSL",), ("Spark SQL is powerful",), ("DSL is flexible",)]
df = spark.createDataFrame(data, ["sentence"])
方法一：SQL 模式实现 WordCount
SQL 模式的核心是将数据注册为临时视图，然后通过编写 SQL 语句来实现单词计数。
步骤如下：
- 将 DataFrame 注册为临时视图
df.createOrReplaceTempView("sentences")
- 编写 SQL 语句实现单词计数
# 使用 SQL 进行单词拆分、过滤和计数
word_count_sql = spark.sql("""SELECT word, COUNT(*) as countFROM (-- 拆分句子为单词SELECT explode(split(sentence, ' ')) as wordFROM sentences) tempWHERE word != '' -- 过滤空字符串GROUP BY wordORDER BY count DESC
""")# 显示结果
word_count_sql.show()
- 输出结果
+--------+-----+
| word|count|
+--------+-----+
| Hello| 2|
| Spark| 2|
| SQL| 2|
| DSL| 2|
| is| 2|
|powerful| 1|
|flexible| 1|
+--------+-----+
方法二：DSL 模式实现 WordCount
DSL 模式（DataFrame API）通过调用 DataFrame 的方法链来实现功能，不需要编写 SQL 语句。
步骤如下：
# 使用 DataFrame API (DSL) 实现单词计数
word_count_dsl = df.select(# 拆分句子并展开为多行explode(split(col("sentence"), " ")).alias("word")
).filter(col("word") != "" # 过滤空字符串
).groupBy(col("word") # 按单词分组
).count(
).orderBy(col("count").desc() # 按计数降序排列
)# 显示结果
word_count_dsl.show()
输出结果与 SQL 模式完全相同。
两种模式的对比分析
| 特点 | SQL 模式 | DSL 模式 |
|---|---|---|
| 语法风格 | 使用标准 SQL 语句 | 使用方法链调用（如 select、filter、groupBy） |
| 适用人群 | 熟悉 SQL 的数据分析师、数据工程师 | 熟悉编程的开发者 |
| 灵活性 | 适合复杂查询（如窗口函数、子查询） | 适合程序式数据处理流程 |
| 可读性 | 对于复杂业务逻辑，SQL 结构更清晰 | 对于数据处理流水线，方法链更直观 |
| 类型安全 | 运行时检查 | 部分支持编译时检查（Scala/Java） |
完整代码示例
下面是两种模式的完整代码，可以直接运行：
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, col# 初始化 SparkSession
spark = SparkSession.builder \.appName("WordCount with Spark SQL") \.master("local[*]") \.getOrCreate()# 准备示例数据
data = [("Hello Spark SQL",),("Hello DSL",),("Spark SQL is powerful",),("DSL is flexible",),("Spark and SQL and DSL",)
]
df = spark.createDataFrame(data, ["sentence"])
print("原始数据:")
df.show(truncate=False)# 方法1: SQL 模式实现 WordCount
print("\n=== SQL 模式结果 ===")
df.createOrReplaceTempView("sentences")
word_count_sql = spark.sql("""SELECT word, COUNT(*) as countFROM (SELECT explode(split(sentence, ' ')) as wordFROM sentences) tempWHERE word != ''GROUP BY wordORDER BY count DESC
""")
word_count_sql.show()# 方法2: DSL 模式实现 WordCount
print("\n=== DSL 模式结果 ===")
word_count_dsl = df.select(explode(split(col("sentence"), " ")).alias("word")
).filter(col("word") != ""
).groupBy("word"
).count(
).orderBy(col("count").desc()
)
word_count_dsl.show()# 停止 SparkSession
spark.stop()
关键函数解释
split()：将字符串按指定分隔符拆分，返回数组
split(col("sentence"), " ") # 按空格拆分句子explode()：将数组中的每个元素转换为一行，实现 "行转列"
explode(array_column) # 将数组列展开为多行groupBy() + count()：按指定列分组并计数
groupBy("word").count() # 按单词分组并计算出现次数
通过这两种方式，我们可以灵活地利用 Spark SQL 处理文本数据并实现单词计数，根据实际场景和个人习惯选择合适的方式即可。
：Memory)
