Spark---WC---Spark从外部读取数据之textFile

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WC {

    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("WC").setMaster("local")
        val sc: SparkContext = new SparkContext(conf)
        val path = "file:///" + System.getProperty("user.dir") + "/data4test/WC.txt"

        val outpath="file:///" + System.getProperty("user.dir") + "/data4test/WC4result.txt"

        //textFile会产生两个RDD  HadoopRDD 和 MapPartitionsRDD
        val hadoopRDD: RDD[String] = sc.textFile(path)
        //产生一个RDD==>MapPartitionsRDD
        val mapPartitionsRDD_1: RDD[String] = hadoopRDD.flatMap(line => line.split(" "))
        //产生一个RDD==>MapPartitionsRDD
        val mapPartitionsRDD_2: RDD[(String, Int)] = mapPartitionsRDD_1.map(word => (word, 1))
        //产生一个RDD==>ShuffledRDD
        val shuffleRDD: RDD[(String, Int)] = mapPartitionsRDD_2.reduceByKey((a,b)=>a+b)
        shuffleRDD.saveAsTextFile(outpath)
        sc.stop()
    }
}

1
2
3
4
5
6
7
8
9
10
11
/** 
   * Read a text file from HDFS, a local file system (available on all nodes), or any 
   * Hadoop-supported file system URI, and return it as an RDD of Strings. 
   */  
  def textFile(  
      path: String,  
      minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {  
    assertNotStopped()  
    hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],  
      minPartitions).map(pair => pair._2.toString).setName(path)  
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def hadoopFile[K, V](
      path: String,
      inputFormatClass: Class[_ <: InputFormat[K, V]],
      keyClass: Class[K],
      valueClass: Class[V],
      minPartitions: Int = defaultMinPartitions): RDD[(K, V)] = withScope {
    assertNotStopped()
    //   A Hadoop configuration can be about 10 KB, which is pretty big, so broadcast it.
    //   广播hadoop配置文件
    val confBroadcast = broadcast(new SerializableConfiguration(hadoopConfiguration))
    val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path)
    new HadoopRDD(
      this,//SparkContext
      confBroadcast,
      Some(setInputPathsFunc),
      inputFormatClass,
      keyClass,
      valueClass,
      minPartitions).setName(path)
  }

1
2
3
val path = "Current.txt"  //Current fold file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "Current1.txt，Current2.txt，"  //Current fold file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "file:///usr/local/spark/spark-1.6.0-bin-hadoop2.6/README.md"  //local file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "file:///usr/local/spark/spark-1.6.0-bin-hadoop2.6/licenses/"  //local file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "file:///usr/local/spark/spark-1.6.0-bin-hadoop2.6/licenses/LICENSE-scala.txt,file:///usr/local/spark/spark-1.6.0-bin-hadoop2.6/licenses/LICENSE-spire.txt"  //local file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "/usr/local/spark/spark-1.6.0-bin-hadoop2.6/data/*/*"  //local file
val rdd1 = sc.textFile(path,2)

1
2
3
4
for (i <- 1 to 2){
      val rdd1 = sc.textFile(s"/root/application/temp/people$i*",2)
    }

1
2
3
val path = "/usr/local/spark/spark-1.6.0-bin-hadoop2.6/data/*/*.txt"  //local file
val rdd1 = sc.textFile(path,2)

1
2
3
val path = "hdfs://master:9000/examples/examples/src/main/resources/people.txt"
val rdd1 = sc.textFile(path,2)