FlinkDataSet算子的作用是什么

这篇文章主要介绍“Flink DataSet算子的作用是什么”，在日常操作中，相信很多人在Flink DataSet算子的作用是什么问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Flink DataSet算子的作用是什么”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

创新互联是一家集网站建设,洪泽企业网站建设,洪泽品牌网站建设,网站定制,洪泽网站建设报价,网络营销,网络优化,洪泽网站推广为一体的创新建站企业，帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿，时刻以成就客户成长自我，坚持不断学习、思考、沉淀、净化自己，让我们为更多的企业打造出实用型网站。

Flink为了能够处理有边界的数据集和无边界的数据集，提供了对应的DataSet API和DataStream API。我们可以开发对应的Java程序或者Scala程序来完成相应的功能。下面举例了一些DataSet API中的基本的算子。

下面我们通过具体的代码来为大家演示每个算子的作用。

1、Map、FlatMap与MapPartition

//获取运行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

ArrayList data = new ArrayList();
data.add("I love Beijing");
data.add("I love China");
data.add("Beijing is the capital of China");
DataSource text = env.fromCollection(data);

DataSet> mapData = text.map(new MapFunction>() {

	public List map(String data) throws Exception {
		String[] words = data.split(" ");
		
		//创建一个List
		List result = new ArrayList();
		for(String w:words){
			result.add(w);
		}
		return result;
	}
});
mapData.print();
System.out.println("*****************************************");

DataSet flatMapData = text.flatMap(new FlatMapFunction() {

	public void flatMap(String data, Collector collection) throws Exception {
		String[] words = data.split(" ");
		for(String w:words){
			collection.collect(w);
		}
	}
});
flatMapData.print();

System.out.println("*****************************************");
/*	new MapPartitionFunction
	第一个String：表示分区中的数据元素类型
	第二个String：表示处理后的数据元素类型*/
DataSet mapPartitionData = text.mapPartition(new MapPartitionFunction() {

	public void mapPartition(Iterable values, Collector out) throws Exception {
		//针对分区进行操作的好处是：比如要进行数据库的操作，一个分区只需要创建一个Connection
		//values中保存了一个分区的数据
		 Iterator it = values.iterator();
		while (it.hasNext()) {
			String next = it.next();
			String[] split = next.split(" ");
			for (String word : split) {
				out.collect(word);
			}
		}
		//关闭链接
	}
});
mapPartitionData.print();

2、Filter与Distinct

//获取运行环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

ArrayList data = new ArrayList();
data.add("I love Beijing");
data.add("I love China");
data.add("Beijing is the capital of China");
DataSource text = env.fromCollection(data);

DataSet flatMapData = text.flatMap(new FlatMapFunction() {

	public void flatMap(String data, Collector collection) throws Exception {
		String[] words = data.split(" ");
		for(String w:words){
			collection.collect(w);
		}
	}
});

//去掉重复的单词
flatMapData.distinct().print();
System.out.println("*********************");

//选出长度大于3的单词
flatMapData.filter(new FilterFunction() {
	
	public boolean filter(String word) throws Exception {
		int length = word.length();
		return length>3?true:false;
	}
}).print();

3、Join操作

//获取运行的环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//创建第一张表：用户ID  姓名
ArrayList> data1 = new ArrayList>();
data1.add(new Tuple2(1,"Tom"));
data1.add(new Tuple2(2,"Mike"));
data1.add(new Tuple2(3,"Mary"));
data1.add(new Tuple2(4,"Jone"));
//创建第二张表：用户ID 所在的城市
ArrayList> data2 = new ArrayList>();
data2.add(new Tuple2(1,"北京"));
data2.add(new Tuple2(2,"上海"));
data2.add(new Tuple2(3,"广州"));
data2.add(new Tuple2(4,"重庆"));

//实现join的多表查询：用户ID  姓名  所在的程序
DataSet> table1 = env.fromCollection(data1);
DataSet> table2 = env.fromCollection(data2);

table1.join(table2).where(0).equalTo(0)
/*第一个Tuple2：表示第一张表
 * 第二个Tuple2：表示第二张表
 * Tuple3：多表join连接查询后的返回结果   */		                   
.with(new JoinFunction, Tuple2, Tuple3>() {
	public Tuple3 join(Tuple2 table1,
			Tuple2 table2) throws Exception {
		return new Tuple3(table1.f0,table1.f1,table2.f1);
	} }).print();

4、笛卡尔积

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//创建第一张表：用户ID  姓名
ArrayList> data1 = new ArrayList>();
data1.add(new Tuple2(1,"Tom"));
data1.add(new Tuple2(2,"Mike"));
data1.add(new Tuple2(3,"Mary"));
data1.add(new Tuple2(4,"Jone"));

//创建第二张表：用户ID 所在的城市
ArrayList> data2 = new ArrayList>();
data2.add(new Tuple2(1,"北京"));
data2.add(new Tuple2(2,"上海"));
data2.add(new Tuple2(3,"广州"));
data2.add(new Tuple2(4,"重庆"));

//实现join的多表查询：用户ID  姓名  所在的程序
DataSet> table1 = env.fromCollection(data1);
DataSet> table2 = env.fromCollection(data2);

//生成笛卡尔积
table1.cross(table2).print();

5、First-N

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//这里的数据是：员工姓名、薪水、部门号
DataSet> grade = 
		env.fromElements(new Tuple3("Tom",1000,10),
						 new Tuple3("Mary",1500,20),
						 new Tuple3("Mike",1200,30),
						 new Tuple3("Jerry",2000,10));

//按照插入顺序取前三条记录
grade.first(3).print();
System.out.println("**********************");

//先按照部门号排序，在按照薪水排序
grade.sortPartition(2, Order.ASCENDING).sortPartition(1, Order.ASCENDING).print();
System.out.println("**********************");

//按照部门号分组，求每组的第一条记录
grade.groupBy(2).first(1).print();

6、外链接操作

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//创建第一张表：用户ID  姓名
ArrayList> data1 = new ArrayList>();
data1.add(new Tuple2(1,"Tom"));
data1.add(new Tuple2(3,"Mary"));
data1.add(new Tuple2(4,"Jone"));

//创建第二张表：用户ID 所在的城市
ArrayList> data2 = new ArrayList>();
data2.add(new Tuple2(1,"北京"));
data2.add(new Tuple2(2,"上海"));
data2.add(new Tuple2(4,"重庆"));

//实现join的多表查询：用户ID  姓名  所在的程序
DataSet> table1 = env.fromCollection(data1);
DataSet> table2 = env.fromCollection(data2);

//左外连接
table1.leftOuterJoin(table2).where(0).equalTo(0)
	  .with(new JoinFunction, Tuple2, Tuple3>() {

		public Tuple3 join(Tuple2 table1,
				Tuple2 table2) throws Exception {
			// 左外连接表示等号左边的信息会被包含
			if(table2 == null){
				return new Tuple3(table1.f0,table1.f1,null);
			}else{
				return new Tuple3(table1.f0,table1.f1,table2.f1);
			}
		}
	}).print();

System.out.println("***********************************");
//右外连接
table1.rightOuterJoin(table2).where(0).equalTo(0)
	  .with(new JoinFunction, Tuple2, Tuple3>() {

		public Tuple3 join(Tuple2 table1,
				Tuple2 table2) throws Exception {
			//右外链接表示等号右边的表的信息会被包含
			if(table1 == null){
				return new Tuple3(table2.f0,null,table2.f1);
			}else{
				return new Tuple3(table2.f0,table1.f1,table2.f1);
			}
		}
	}).print();

System.out.println("***********************************");

//全外连接
table1.fullOuterJoin(table2).where(0).equalTo(0)
.with(new JoinFunction, Tuple2, Tuple3>() {

	public Tuple3 join(Tuple2 table1, Tuple2 table2)
			throws Exception {
		if(table1 == null){
			return new Tuple3(table2.f0,null,table2.f1);
		}else if(table2 == null){
			return new Tuple3(table1.f0,table1.f1,null);
		}else{
			return new Tuple3(table1.f0,table1.f1,table2.f1);
		}
	}
	
}).print();

到此，关于“Flink DataSet算子的作用是什么”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！

名称栏目：FlinkDataSet算子的作用是什么
分享URL：http://scjbc.cn/article/jocedd.html