9.3 scala和Java做Spark操作

墨蓝 2022-05-20 08:37 113阅读 0赞

注意导入spark-assembly-1.6.0-hadoop2.4.0.jar

scala:

package com.scala

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

object Test \{

def main(args: Array\[String\]): Unit = \{

val conf = new SparkConf().setMaster("local").setAppName("WordCount")

val sc = new SparkContext(conf)

val lines = sc.textFile("test.txt")

val words = lines.flatMap \{ line => line.split(" ") \}

val pairs = words.map \{ word => (word, 1) \}

val results = pairs.reduceByKey( (a,b) => \{a+b\} )

results.foreach(x=>println(x))

![70][]

--------------------

java:

package com.mp;

import java.util.Arrays;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class Test \{

public static void main(String\[\] args) \{

SparkConf conf = new SparkConf(). setMaster("local"). setAppName("WordCount");

JavaSparkContext sc = new JavaSparkContext(conf);

//读取文本数据

JavaRDD<String> lines = sc. textFile("test.txt");

//每行按空格切割字符串，得到词

JavaRDD<String> words = lines. flatMap(new FlatMapFunction<String, String>() \{

private static final long serialVersionUID = 1L;

@Override

public Iterable<String> call(String line) throws Exception \{

return Arrays. asList(line.split(" "));

\});

//把每个词组合成map结构，（word,1）

JavaPairRDD<String, Integer> pairs = words. mapToPair(new PairFunction<String, String, Integer>() \{

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(String word) throws Exception \{

return new Tuple2<String, Integer>(word , 1);

\});

//得到组合好的map，进行统计

JavaPairRDD<String, Integer> results = pairs. reduceByKey(new Function2<Integer, Integer, Integer>() \{ private static final long serialVersionUID = 1L;

@Override

public Integer call(Integer v1, Integer v2) throws Exception \{

return v1 + v2;

\});

//计算结果进行排序 重新构建key value 数字 词

JavaPairRDD<Integer, String> pair2 = result. mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() \{

private static final long serialVersionUID = 1L;

@Override

public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple)

throws Exception \{

return new Tuple2<Integer, String>(tuple.\_2,tuple.\_1);

\});

//调转之后的RDD进行sortByKey排序，然后再调换回去

JavaPairRDD<String, Integer> result1 = pair2. sortByKey( false). mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() \{

private static final long serialVersionUID = 1L;

@Override

public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple)

throws Exception \{

return new Tuple2<String, Integer>(tuple.\_2,tuple.\_1);

\});

//得出的结果遍历出来

results.foreach(new VoidFunction<Tuple2<String,Integer>>() \{

private static final long serialVersionUID = 1L;

@Override

public void call(Tuple2<String, Integer> tuple) throws Exception \{

System.out.println("word: "+tuple.\_1+" count: "+tuple.\_2);

\});

sc.close();

[70]: /images/20220520/541075bb402f49528e6e267c5b583c0b.png

9.3 scala和Java做Spark操作

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Spark—通过Java、Scala API实现WordCount案例的基本操作

相关 spark实战java_Scala和Java二种方式实战Spark Streaming开发

相关 scala&spark相关

相关 inteliJ scala和spark环境配置

相关 9.3 scala和Java做Spark操作

相关 Spark常用transformation算子操作 —— Scala版

相关 Spark常用action算子操作 —— Scala版

相关 Spark-RDD转Dataset及简单的SparkSql操作（java和scala版本）

相关 scala和spark实践wordcount

相关 spark的java和scala混合工程构建记录！