下面我將詳細講解如何使用 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)
