project的使用

上篇：connect的使用

project [DataStream → DataStream]

project 主要用于获取 tuples 中的指定字段集

    public static void project(StreamExecutionEnvironment env) {DataStreamSource<Tuple3<String, Integer, String>> streamSource = env.fromElements(new Tuple3<>("li", 22, "2018-09-23"),new Tuple3<>("ming", 33, "2020-09-23"));streamSource.project(0, 2).print();}

运行结果：

 (ming,2020-09-23)(li,2018-09-23)

来一个实例，直接上代码：

package cn._51doit.flink.day03;import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/*** project的使用* 有界流** 直接运行，控制台打印输出：* 2> (99.99,laoduan)* 1> (9999.99,laozhao)*/
public class ProjectDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());DataStreamSource<String> lines = env.fromElements("laozhao,18,9999.99","laoduan,28,99.99");SingleOutputStreamOperator<Tuple3<String, Integer, Double>> tpDataStream = lines.map(e -> {String[] fields = e.split(",");return Tuple3.of(fields[0], Integer.parseInt(fields[1]), Double.parseDouble(fields[2]));}).returns(Types.TUPLE(Types.STRING,Types.INT,Types.DOUBLE));SingleOutputStreamOperator<Tuple> projected = tpDataStream.project(2, 0);projected.print();env.execute();}
}

代码改造后：

package cn._51doit.flink.day03;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** project的使用:有界流** 直接运行，控制台打印输出[代码改造之前]* 4> (99.99,laoduan)* 3> (9999.99,laozhao)** 直接运行，控制台打印输出[代码改造之后]* 3> (9999.99,laozhao)* 4> (99.99,laoduan)*/
public class ProjectDemo_01 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());DataStreamSource<String> lines = env.fromElements("laozhao,18,9999.99","laoduan,28,99.99");SingleOutputStreamOperator<Tuple3<String, Integer, Double>> tpDataStream = lines.map(e -> {String[] fields = e.split(",");return Tuple3.of(fields[0], Integer.parseInt(fields[1]), Double.parseDouble(fields[2]));}).returns(Types.TUPLE(Types.STRING,Types.INT,Types.DOUBLE));SingleOutputStreamOperator<Tuple2<Double, String>> projected = tpDataStream.map(new MapFunction<Tuple3<String, Integer, Double>, Tuple2<Double, String>>() {@Overridepublic Tuple2<Double, String> map(Tuple3<String, Integer, Double> value) throws Exception {return Tuple2.of(value.f2, value.f0);}});projected.print();env.execute();}
}

project的使用相关推荐

ideal如何创建dynamic web project
步骤如下 ① file -> new -> project ② 选择 Java Enterprise -> next ③ create project from template - ...
This version of Android Studio cannot open this project, please retry with Android Studio 3.5 or new
今天github 下载一个库导入 as 提示 This version of Android Studio cannot open this project, please retry with A ...
Error:The SDK Build Tools revision (23.0.3) is too low for project ':app'. Minimum required is 25.0.
导入github上项目的时候出现 Error:The SDK Build Tools revision (23.0.3) is too low for project ':app'. Minimum ...
Error:(49, 1) A problem occurred evaluating project ':guideview'. Could not read script 'https://r
出现问题如下: Error:(49, 1) A problem occurred evaluating project ':guideview'. > Could not read script ...
IntelliJ IDEA 的Project structure说明
IntelliJ IDEA 的Project structure可以在File->Project structure中打开,同时,在新建项目是IDE一般用向导的方式让你填写Project str ...
将Project的内容导出成单独的XPO文件
AX跟VSS整合的版本管理可以通过创建知识库将当前层的代码全部签入到VSS中,但是如果不是一个团队开发solution,而是针对客户的需求随时做得一些小改动,一般都希望以Project的形式组织代码和 ...
linux vim project,vim插件project的用法
用任何编辑器写代码,文件管理的方便与否对编码效率影响很大.一般的IDE都有文件管理功能,并且用来的不错.在vim中,要实现较好的文件管理功能一般都靠插件.在有米实习的第一个月,自己一直用NERDTre ...
android studio 同类,让Android Studio的Project视图和Anroid视图类似
关于AS的Project与Android视图的不同,可以看我的博文 http://blog.csdn.net/siyehuazhilian/article/details/42123563 Andro ...
invalid project description._[Project教程] 在Project软件中如何处理加班工时
在微软Project软件中有处理加班工时的功能,但是我在系统课程中没有讲,为什么呢?就像Project软件中的[进度线]功能,这个功能还不够完善,所以暂时不建议大家使用.加班工时这个问题也是如此,总体 ...
There's no Qt version assigned to this project for platform Win32/Win64
Qt5.8+ vs2010和vs2015都出现这个问题,无论是32bit还是64bit平台解决方法: 1:注意:打开main.cpp文件,否则 "Qt Project Settings&q ...

project的使用

project [DataStream → DataStream]

project的使用相关推荐

最新文章

热门文章