spark

Spark RDD Scala实现单词计数示例

1. 使用本地文件读取

在使用Spark进行本地文件读取并执行单词计数时，首先确保你的环境已配置好Scala和Spark。以下是一个基本的示例代码：

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

object WordCountLocal {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("WordCountLocal")
    val sc = new SparkContext(conf)

    // 假设本地文件路径为"data/words.txt"
    val textFile: RDD[String] = sc.textFile("file:///path/to/your/local/file/words.txt")

    val counts: RDD[(String, Int)] = textFile.flatMap(line => line.split("\\s+"))
                                       .map(word => (word, 1))
                                       .reduceByKey(_ + _)

    counts.saveAsTextFile("file:///path/to/output/directory/wordCountsLocal")
    sc.stop()
  }
}

2. 使用HDFS读取

若要从HDFS读取数据并执行单词计数，你需要确保Hadoop环境已经正确配置，并且Spark能够访问到HDFS。以下是相应的示例代码：

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object WordCountHDFS {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("WordCountHDFS")
                               .setMaster("yarn") // 或其他Spark集群管理器配置
    val sc = new SparkContext(conf)

    // 假设HDFS文件路径为"hdfs://namenode:port/path/to/hdfs/file/words.txt"
    val textFile: RDD[String] = sc.textFile("hdfs://namenode:port/path/to/hdfs/file/words.txt")

    val counts: RDD[(String, Int)] = textFile.flatMap(line => line.split("\\s+"))
                                           .map(word => (word, 1))
                                           .reduceByKey(_ + _)

    counts.saveAsTextFile("hdfs://namenode:port/path/to/output/directory/wordCountsHDFS")
    sc.stop()
  }
}

注意： - 在实际应用中，namenode:port、/path/to/your/local/file/words.txt 和 /path/to/hdfs/file/words.txt 需要替换为你具体的主机名、端口和文件路径。 - 确保Spark作业提交前，Scala和Spark环境已正确安装与配置，且对于HDFS读取，Hadoop配置已完成验证。 - 如果你是在阿里云环境中操作，考虑使用OSS作为数据源时，可以参考文档配置Spark支持OSS读写以加速数据查询。