一、
以下环境在pycharm中运行
确保jdk,hadoop,spark,scala 软件在电脑端安装完毕，环境搭建
pycharm 环境配置
环境变量配置：
PYTHONUNBUFFERED=1; -u 不缓冲stdin、stdout和stderr 默认是缓冲的。同PYTHONUNBUFFERED=1 不用buffer的意思
PYTHONPATH=C:spark-2.2.0-bin-hadoop2.7python;
SPARK_HOME=C:spark-2.2.0-bin-hadoop2.7
from pyspark import SparkContext,SparkConf
conf = SparkConf().setMaster("local").setAppName("liujinjie")
sc = SparkContext(conf=conf)
textFile = sc.textFile("D:hello.txt")
wordCount = textFile.flatMap(lambda line: line.split(",")).map(lambda word:(word,1)).reduceByKey(lambda a,b :a + b)
wordCount.foreach(print) 其中flatMap,Map,reduceByKey 含义讲解 link
二、
scala 代码
collect: 收集一个弹性分布式数据集的所有元素到一个数组中,这样便于我们观察，毕竟分布式数据集比较抽象。Spark的collect方法，是Action类型的一个算子，会从远程集群拉取数据到driver端。最后，将大量数据
汇集到一个driver节点上，将数据用数组存放，占用了jvm堆内存，非常用意造成内存溢出，只用作小型数据的观察。
val arr = res.collect();
import org.apache.spark.{SparkContext, SparkConf}
object WordCount {
def main(args: Array[String]) {
/*
创建Spark配置对象SparkConf,设置Spark程序运行时的配置信息,
SetMaster 来设置程序要链接的Spark集群Master的URL，如果设定为local 则代表在本地运行，
setAppName 设定应用程序名称，在程序运行监控界面可看到名称
*/
val conf = new SparkConf().setMaster("local").setAppName("testRdd")
val sc = new SparkContext(conf) // 等价与 val sc = new SparkContext("local","testRdd")
/**
* 创建SparkContext对象，SparkContext是所有程序功能的唯一入口，无论用Scala、Java、Python、R等都必须要有个SparkContext
* SparkContext核心作用是初始化spark应用程序所需核心组件，包含DAGScheduler,TaskScheduler,SchedulerBackend
* 还负责 spark 程序往master 注册程序等， 是整个应用程序最重要的一个对象， 通过传入conf 定制spark运行的具体参数跟配置信息
*/
/**
* 3.根据具体数据的来源（HDFS,HBase,Local,FS,DB,S3等）通过SparkContext来创建RDD；
* RDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其他的RDD操作；
* 数据会被RDD划分成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴；
*/
/**
* 4.对初始的RDD进行Transformation级别的处理，例如map,filter等高阶函数的变成，来进行具体的数据计算
* 4.1.将每一行的字符串拆分成单个单词
*/
//对每一行的字符串进行拆分并把所有行的拆分结果通过flat合并成一个大的集合
val lines = sc.textFile("d://hello.txt")
val words = lines.flatMap(_.split(","))
val pairs = words.map{word =>(word,1)}
val wordCounts = pairs.reduceByKey(_ + _)
wordCounts.foreach(pair => println(pair._1 + ":" + pair._2))
sc.stop()
// data.flatMap(_.split(","))//下划线是占位符，flatMap是对行操作的方法，对读入的数据进行分割
// .collect()//将分布式的RDD返回一个单机的scala array，在这个数组上运用scala的函数操作，并返回结果到驱动程序
// .foreach(println)//循环打印
}
}scala> List(1, 2, 3, 4, 5, 6) collect { case i if i % 2 == 0 => i * i }
res0: List[Int] = List(4, 16, 36)
scala> List(1, 2, 3, 4, 5, 6) collect { case i if i % 2 == 0 => i+1 }
res1: List[Int] = List(3, 5, 7)
scala> List(1, 2, 3, 4, 5, 6) collect { case i => i+1 }
res2: List[Int] = List(2, 3, 4, 5, 6, 7)
scala> List(1, 2, 3, 4, 5, 6) collect { case i =>(i, i+1 )}
res3: List[(Int, Int)] = List((1,2), (2,3), (3,4), (4,5), (5,6), (6,7))
scala> List(1, 2, 7, 4, 9, 6) collect { case i =>(i, i+1 )}
res4: List[(Int, Int)] = List((1,2), (2,3), (7,8), (4,5), (9,10), (6,7))
scala> List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;i)=> i * i }
<console>:1: error: ')' expected but ';' found.
List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;i)=> i * i }
^
<console>:1: error: ';' expected but ')' found.
List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;i)=> i * i }
^
scala> List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;true)=> i * i }
<console>:1: error: ')' expected but ';' found.
List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;true)=> i * i }
^
<console>:1: error: ';' expected but ')' found.
List(1, 2, 3, 4, 5, 6) collect { case i if (i=i+1;true)=> i * i }
^
scala> List(1, 2, 3, 4, 5, 6) collect { case i if {i=i+1;true}=> i * i }
<console>:8: error: reassignment to val
List(1, 2, 3, 4, 5, 6) collect { case i if {i=i+1;true}=> i * i }
^
scala> List(1, 2, 3, 4, 5, 6) collect { case i if {i+1;true}=> i * i }
res6: List[Int] = List(1, 4, 9, 16, 25, 36)