01｜Spark：从“大数据的Hello World”开始

在Python中运行: ``` from pyspark import SparkContext textFile = SparkContext().textFile("./wikiOfSpark.txt") wordCount = ( textFile.flatMap(lambda line: line.split(" ")) .filter(lambda word: word != "") .map(lambda word: (word, 1)) .reduceByKey(lambda x, y: x + y) .sortBy(lambda x: x[1], False) .take(5) ) print(wordCount) #显示: [('the', 67), ('Spark', 63), ('a', 54), ('and', 51), ('of', 50)] ```

我是一个大数据小白，我想咨询下spark和hadoop在大数据体系下的关系？

老师好！wordCounts.map{case (k, v) => (v, k)}.sortByKey(false).take(5)这行代码我还存在疑问，为什么这里的map函数使用了花括号{ }而不是上面一些算子的( )，同时这个case又是什么意思？这一行代码非常像我曾经在Python中使用字典数据结构，然后根据字典值的升序排序。最后，貌似Scala语言本身就可以实现wordcount案例，那么它本身的实现和spark实现相比，spark有什么优势呢？

遇到这个问题 scala> val rootPath: String = _ <console>:24: error: unbound placeholder parameter val rootPath: String = _ 网上搜一下，说这是汇编错误。 要把val 改成var , 但会遇到"_"这default值是null。 org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:/Users/vic/src/data/null/wikiOfSpark.txt 这一段就先跳过root_path，直接给file一个路径，是可以成功运行"word count",得到和老师一样的结果: [Stage 0:> (0 + 2) / res0: Array[(Int, String)] = Array((67,the), (63,Spark), (54,a), (51,and), (50,of))

老师我可以再问一下，如果我是用IDEA创建Spark项目，是不是只要配置好Scala的SDK，然后在pom文件中加入对应版本号的spark依赖，就会自动下载spark包了？这个时候不需要再去官网下载spark了吗，同时也不再需要使用spark-shell了吗？

前排占座！三年前接触过 Spark 今天从头再学！

Java实现： SparkConf sparkConf = new SparkConf().setAppName("Test").setMaster("local[*]"); JavaSparkContext JSC = new JavaSparkContext(sparkConf); // 读取文件内容 JavaRDD<String> lineRDD = JSC.textFile("wikiOfSpark.txt"); // 以行为单位做分词 JavaRDD<String> wordRDD = lineRDD.flatMap(new FlatMapFunction<String, String>() { @Override public Iterator<String> call(String s) throws Exception { return Arrays.asList(s.split(" ")).iterator(); } }); JavaRDD<String> cleanWordRDD = wordRDD.filter(new Function<String, Boolean>() { @Override public Boolean call(String s) throws Exception { return !s.equals(""); } }); // 把RDD元素转换为（Key，Value）的形式 JavaPairRDD<String, Integer> kvRDD = cleanWordRDD.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { return new Tuple2<String, Integer>(s,1); } }); // 按照单词做分组计数 JavaPairRDD<String, Integer> wordCounts = kvRDD.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer integer, Integer integer2) throws Exception { return integer+integer2; } }); // 打印词频最高的5个词汇(先将元组的key value交换一下顺序，然后在调用sortByKey()) wordCounts.mapToPair((row)-> new Tuple2<>(row._2,row._1)).sortByKey(false).foreach(new VoidFunction<Tuple2<Integer, String>>() { @Override public void call(Tuple2<Integer, String> stringIntegerTuple2) throws Exception { System.out.println(stringIntegerTuple2._1 + ":" + stringIntegerTuple2._2); } }); //关闭context JSC.close();

Python版代码为： file='~~~/wikiOfSpark.txt' lineRDD=sc.textFile(file) lineRDD.first() # 会打印出lineRDD的第一行： u'Apache Spark'，如果出错则不打印 wordRDD=lineRDD.flatMap(lambda line: line.split(" ")) cleanWordRDD=wordRDD.filter(lambda word: word!='') kvRDD=cleanWordRDD.map(lambda word:(word,1)) wordCounts=kvRDD.reduceByKey(lambda x,y:x+y) wordCounts.map(lambda (k,v):(v,k)).sortByKey(False).take(5)

问下执行 val lineRDD: RDD[String] = spark.sparkContext.textFile(file) 报错error: not found: value spark是怎么回事？

2022年2月5号学习打卡记录 机器环境：ROG14 操作系统：win11 + wsl Ubuntu20.04 环境变量： ---------------------------- export SPARK_HOME=/mnt/c/spark/spark-2.4.8-bin-hadoop2.7 export JAVA_HOME=/mnt/c/linux_environment/jdk/jdk1.8.0_321 export M2_HOME=/mnt/c/linux_environment/apache-maven-3.8.4 export SCALA_HOME=/mnt/c/linux_environment/scala3-3.1.1 export PATH=$SPARK_HOME/bin:$SCALA_HOME/bin:$M2_HOME/bin:$JAVA_HOME/bin:$PATH --------------------------- 希望帮助和我一样从零开始一起学习的同学躲避一些坑： 坑1：jdk版本不兼容： 一开始使用jdk17版本，在启动过程中一直报错，降为1.8后启动成功； 坑2：hadoop版本问题： hadoop3.2.1 逐步使用Dataset，报错类型转换异常； 由于scala经验不足，暂时无法大规模改写老师的代码，降低版本为spark2.4.8 下载地址：https://dlcdn.apache.org/spark/ 可以选择适合的版本下载 原理性的还没有搞懂，目前在第一阶段，读懂，简单改写为主； 感谢吴磊老师的课