### 使用 Scala 或 Python 实现 Spark Core 数据并行化处理 以下展示了如何使用 Scala 和 Python 编写 Spark Core 程序，以实现数据并行化处理。具体示例包括统计日志关键词频率和分析结构化数据。 #### 统计日志关键词频率（Scala 示例） 在 Scala 中，可以通过 `RDD` 的变换操作（如 `flatMap`、`map` 和 `reduceByKey`）来实现日志关键词频率的统计。 ```scala import org.apache.spark.{SparkConf, SparkContext} object LogKeywordFrequency { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("LogKeywordFrequency").setMaster("local[*]") val sc = new SparkContext(conf) // 假设日志文件路径为 "logs.txt" val logData = sc.textFile("logs.txt") // 分词并统计关键词频率 val keywordCounts = logData .flatMap(line => line.split("\\s+")) // 按空格分词 .map(word => (word, 1)) // 映射为 (关键词, 1) .reduceByKey(_ + _) // 按关键词聚合计数 // 输出结果 keywordCounts.collect().foreach(println) sc.stop() } } ``` 上述代码中，`flatMap` 将每行日志拆分为单词[^1]，`map` 将每个单词映射为键值对 `(单词, 1)`，最后通过 `reduceByKey` 聚合相同单词的计数[^1]。 #### 统计日志关键词频率（Python 示例） 在 Python 中，可以使用 PySpark 提供的 API 来实现类似的功能。 ```python from pyspark import SparkConf, SparkContext if __name__ == "__main__": conf = SparkConf().setAppName("LogKeywordFrequency").setMaster("local[*]") sc = SparkContext(conf=conf) # 假设日志文件路径为 "logs.txt" log_data = sc.textFile("logs.txt") # 分词并统计关键词频率 keyword_counts = log_data \ .flatMap(lambda line: line.split()) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a + b) # 输出结果 for word, count in keyword_counts.collect(): print(f"{word}: {count}") sc.stop() ``` 这段代码与 Scala 版本类似，通过 `flatMap`、`map` 和 `reduceByKey` 完成关键词频率统计。 #### 分析结构化数据（Scala 示例） 对于结构化数据分析，可以使用 `RDD` 的复杂变换操作。例如，假设有一个包含用户行为的日志文件，格式为 `user_id, action, timestamp`，可以统计每个用户的总行为次数。 ```scala import org.apache.spark.{SparkConf, SparkContext} object StructuredDataAnalysis { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("StructuredDataAnalysis").setMaster("local[*]") val sc = new SparkContext(conf) // 假设结构化数据文件路径为 "user_actions.csv" val data = sc.textFile("user_actions.csv") // 解析并统计每个用户的总行为次数 val userActionCounts = data .map(line => line.split(",")) // 按逗号分割 .map(parts => (parts(0), 1)) // 映射为 (user_id, 1) .reduceByKey(_ + _) // 按用户 ID 聚合计数 // 输出结果 userActionCounts.collect().foreach(println) sc.stop() } } ``` #### 分析结构化数据（Python 示例） 同样，可以使用 PySpark 来完成类似的结构化数据分析任务。 ```python from pyspark import SparkConf, SparkContext if __name__ == "__main__": conf = SparkConf().setAppName("StructuredDataAnalysis").setMaster("local[*]") sc = SparkContext(conf=conf) # 假设结构化数据文件路径为 "user_actions.csv" data = sc.textFile("user_actions.csv") # 解析并统计每个用户的总行为次数 user_action_counts = data \ .map(lambda line: line.split(",")) \ .map(lambda parts: (parts[0], 1)) \ .reduceByKey(lambda a, b: a + b) # 输出结果 for user_id, count in user_action_counts.collect(): print(f"User {user_id}: {count} actions") sc.stop() ``` 上述代码通过 `map` 将每行数据解析为 `(user_id, 1)`，然后通过 `reduceByKey` 聚合计数。 --- ###