算子

算子

关注
发信
关注(28)粉丝(399)

spark算子介绍

1.spark的算子分为转换算子和Action算子，Action算子将形成一个job，转换算子RDD转换成另一个RDD,或者将文件系统的数据转换成一个RDD

2.Spark的算子介绍地址：http://spark.apache.org/docs/2.3.0/rdd-programming-guide.html

3.Spark操作基本步骤【java版本，其他语言可以根据官网的案例进行学习】

（1）创建配置文件，将集群的运行模式设置好，给作业起一个名字，可以使用set方法其他配置设入。

SparkConf sparkConf = new SparkConf().setAppName("Demo").setMaster("local");

这里使用的是local的运行模式，起的名字是Demo

（2）创建SparkContext

JavaSparkContext javaContext = new JavaSparkContext(sparkConf);

（3）使用算子，操作数据

　　　　　JavaRDD<String> javaRdd = sparkContext.textFile("logfile.txt",);

        javaRdd = javaRdd.cache();//这一句必须这样写，我们在数据计算很费时的时候，将数据缓存

        long line = javaRdd.count();

        System.out.println(line);

（4）关闭资源

sparkContext.close();

上面以一个求出数据行数的例子，看一下代码操作的流程。

4.Action算子和介绍和举例

（1）map算子;将数据读取使用map进行操作，使用foreach算子计算出 结果。 每一次读取partition中的一条数据进行分析

　　案例：将数据乘以10，在输出，测试算子。

package kw.test.demo;

import java.util.Arrays;

import java.util.List;

import org.apache.spark.SparkConf;

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

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

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

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

/*

 * 本案例：将数据 值乘以一个数，然后将数据的值返回。

 */

public class MapApp {

    public static void main(String[] args) {

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

        JavaSparkContext jsc= new JavaSparkContext(conf);

        List<Integer> list = Arrays.asList(,,,,) ;

        JavaRDD<Integer> javaRdd = jsc.parallelize(list);

        JavaRDD<Integer> result = javaRdd.map(new Function<Integer,Integer>() {

            @Override

            public Integer call(Integer list) throws Exception {

                // TODO Auto-generated method stub

                return list*;

            }

        });

        result.foreach(new  VoidFunction<Integer>() {

            @Override

            public void call(Integer result) throws Exception {

                // TODO Auto-generated method stub

                System.out.println(result);

            }

        });

        jsc.close();

    }

}

(2)MapPartition:将一整块的数据放入然后处理，他和map的区别就是，map将一部分数据放入然后计算，MapPartition将一整块的数据一起放入计算。

如果数据量小的时候，可以是Mappartition中，如果数据量比较大的时候使用Map会比较好，因为可以防止内存溢出。

package kw.test.demo;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.HashMap;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

import org.apache.hadoop.hive.metastore.api.Function;

import org.apache.spark.SparkConf;

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.VoidFunction;

public class MapPartitionApp {

    public static void main(String[] args) {

        /*

         * 创建配置文件

         * 创建出RDD

         */

        SparkConf sparkconf = new SparkConf().setMaster("local").setAppName("mapPartition");

        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkconf);

        /*

         * mapPartition的使用是将一个块一起放入到算子中操作。

         *

         * 假如说RDD上的数据不是太多的时候，可以使用mapPartition 来操作，如果一个RDD的数据比较多还是使用map好

         * 返回了大量数据，容易曹成内存溢出。

         */

    /*    准备数据集*/

        List <String> list= Arrays.asList("kangwang","kang","wang");

        JavaRDD<String> javaRDD = javaSparkContext.parallelize(list);

        final Map<String,Integer> sore = new HashMap<String ,Integer>();

        sore.put("kangwang", );

        sore.put("kang", );

        sore.put("wang", );

        JavaRDD<Integer> sRDD= javaRDD.mapPartitions(new FlatMapFunction<Iterator<String>, Integer>() {

            @Override

            public Iterator<Integer> call(Iterator<String> it) throws Exception {

                // TODO Auto-generated method stub

                List list = new ArrayList();

                while(it.hasNext())

                {

                    String name = it.next();

                    Integer so = sore.get(name);

                    list.add(so);

                }

                Iterator i =list.iterator();

                return  i;

            }

        });

        sRDD.foreach(new VoidFunction<Integer>() {

            @Override

            public void call(Integer it) throws Exception {

                // TODO Auto-generated method stub

                System.out.println("it的值是"+it);

            }

        });

    }

}

（3）MapPartitionWithIndex:

本案例：
  查看将数据的分配到具体的快上的信息。
  我们可以指定partition的个数，默认是2
  parallelize并行集合的时候，指定了并行度，也就是partition的个数是2
  具体他们的数据怎样分，我们并不知道，由spark自己分配
 如果想要知道，就可以使用此算子，将数据的值打印出来。

package kw.test.demo;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.List;

import org.apache.spark.SparkConf;

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

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

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

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

public class MapPartionWithIndex {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setAppName("MapPartitionWithIndex").setMaster("local");

        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

        //准备数据

        List<String> list =new ArrayList<String>();

        list.add("Demo1");

        list.add("Demo2");

        list.add("Demo3");

        list.add("Demo4");

        list.add("Demo5");

        list.add("Demo6");

        list.add("Demo7");

        list.add("Demo8");

        list.add("Demo9");

        list.add("Demo10");

        list.add("Demo11");

        list.add("Demo12");

        //创建RDD,指定map的个数4

        JavaRDD<String> javaRDD = javaSparkContext.parallelize(list, );

        JavaRDD<String> javaRDD1 =javaRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {

            @Override

            public Iterator<String> call(Integer index, Iterator<String> it2) throws Exception {

                // TODO Auto-generated method stub

                //index是partition的个数

                List<String> list = new ArrayList<String>();

                while(it2.hasNext())

                {

                    String name = it2.next();

                    String info = "partition是:"+index+"数据的name是："+name;

                    list.add(info);

                }

                return list.iterator();

            }

        }, true);

        javaRDD1.foreach(new VoidFunction<String>() {

            @Override

            public void call(String infos) throws Exception {

                // TODO Auto-generated method stub

                System.out.println(infos);

            }

        });

    }

}

(4)coalesce算子，是架构RDD的partition的数量缩减

将一定数量的partition压缩到更少的partition分区中去

使用的场景，很多时候在filter算子应用之后会优化一下到使用coalesce算子。

filter算子应用到RDD上面，说白了会应用到RDD对应到里面的每个partition上

数据倾斜，换句话说就是有可能的partition里面就剩下了一条数据 建议使用coalesce算子，

从前各个partition中 数据都更加的紧凑就可以减少它的 个数

package kw.test.demo;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Iterator;

import java.util.List;

import org.apache.spark.SparkConf;

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

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

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

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

public class CoalesceOpter {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setAppName("coalesceDemo").setMaster("local");

        JavaSparkContext javaContext = new JavaSparkContext(sparkConf);

        List<String> list = Arrays.asList("kw1","djf","kw1","fgf",

                "djf","kw1","djf","sdsds","kw1","ssdu","djf");

        JavaRDD<String>javaRDD = javaContext.parallelize(list,);

        JavaRDD<String> info = javaRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {

            @Override

            public Iterator<String> call(Integer arg0, Iterator<String> arg1) throws Exception {

                // TODO Auto-generated method stub

                List<String> list =new ArrayList<String>();

                while(arg1.hasNext())

                {

                    String name = arg1.next();

                    String info = arg0 +"^^^^^……………………………………………………………………"+ name;

                    list.add(info);

                }

                return list.iterator();

            }

        }, true);

        info.foreach(new VoidFunction<String>() {

            @Override

            public void call(String arg0) throws Exception {

                // TODO Auto-generated method stub

                System.out.println(arg0);

            }

        });

        info.coalesce();

        JavaRDD<String> javaRDD1 = info.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {

            @Override

            public Iterator<String> call(Integer arg0, Iterator<String> arg1) throws Exception {

                // TODO Auto-generated method stub

                List<String> list = new ArrayList<String>();

                while(arg1.hasNext())

                {

                    String name = arg1.next();

                    String info2 ="        " +name +"………………………………" +arg0;

                    list.add(info2);

                }

                return list.iterator();

            }

        }, true);

        javaRDD1.foreach(new VoidFunction<String>() {

            @Override

            public void call(String arg0) throws Exception {

                // TODO Auto-generated method stub

                System.out.println(arg0);

            }

        });

    }

}

*

（5）filter此案例将数据的值过滤出来。使用的是filter算子

package kw.test.demo;

import java.util.ArrayList;

import java.util.List;

import org.apache.spark.SparkConf;

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

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

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

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

/*

 * 此案例将数据的值过滤出来。使用的是filter算子

 */

public class APPFilter {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("Filter");

        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        List<Integer> list = new ArrayList<Integer>();

        list.add();

        list.add();

        list.add();

        list.add();

        list.add();

        list.add();

        list.add();

        list.add();

        list.add();

        JavaRDD<Integer> javaRDD = jsc.parallelize(list,);

        //返回值  将返回true的数据返回

        JavaRDD<Integer> num= javaRDD.filter(new Function<Integer, Boolean>() {

            @Override

            public Boolean call(Integer it) throws Exception {

                // TODO Auto-generated method stub

                return it%==;

            }

        });

        num.foreach(new VoidFunction<Integer>() {

            @Override

            public void call(Integer arg0) throws Exception {

                // TODO Auto-generated method stub

                System.out.println(arg0);

            }

        });

    }

}

spark程序可以在本地运行，也可以在集群中运行，可以大成jar，放到真实的集群环境中运行程序。

 

05-23 06:23
Powered by LMLPHP ©2024 算子 0.051482