这篇文章主要为大家展示了“本地jvm如何执行flink程序带web ui的操作”,内容简而易懂,条理清晰,希望能够帮助大家解决疑惑,下面让小编带领大家一起研究并学习一下“本地jvm如何执行flink程序带web ui的操作”这篇文章吧。
StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
可以获取flink执行环境。但是本地jvm执行的时候是不带web ui的。有时候出于监控的考虑,需要带着监控页面查看。任务运行状况,可以使用下面方式获取flink本地执行环境,并带有web ui。
Configuration config = new Configuration(); config.setInteger(RestOptions.PORT,9998); StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
<properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <flink.version>1.6.3</flink.version> <java.version>1.8</java.version> <scala.version>2.11.8</scala.version> <hbase.version>1.2.4</hbase.version> <scala.binary.version>2.11</scala.binary.version> <maven.compiler.source>${java.version}</maven.compiler.source> <maven.compiler.target>${java.version}</maven.compiler.target> </properties> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency>
import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.common.JobExecutionResult; import org.apache.flink.api.java.ExecutionEnvironment; public class FlinkReadTextFile { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment(); DataSet<String> data = env.readTextFile("file:///Users/***/Documents/test.txt"); data.filter(new FilterFunction<String>() { @Override public boolean filter(String value) throws Exception { return value.startsWith("五芳斋美"); } }) .writeAsText("file:///Users/***/Documents/test01.txt"); JobExecutionResult res = env.execute(); } }
import org.apache.flink.streaming.api.windowing.time.Time import org.apache.flink.streaming.api.scala._ object SocketWindowWordCount { /** Main program method */ def main(args: Array[String]): Unit ={ // the port to connect to // val port: Int = try { // ParameterTool.fromArgs(args).getInt("port") // } catch { // case e: Exception => { // System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'") // return // } // } // get the execution environment val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // get input data by connecting to the socket val text = env.socketTextStream("localhost", 9000, '\n') // parse the data, group it, window it, and aggregate the counts val windowCounts = text .flatMap { w => w.split("\\s") } .map { w => WordWithCount(w, 1) } .keyBy("word") .timeWindow(Time.seconds(5), Time.seconds(1)) .sum("count") // print the results with a single thread, rather than in parallel windowCounts.print().setParallelism(1) env.execute("Socket Window WordCount") } // Data type for words with count case class WordWithCount(word: String, count: Long) }
免责声明:本站发布的内容(图片、视频和文字)以原创、来自本网站内容采集于网络互联网转载等其它媒体和分享为主,内容观点不代表本网站立场,如侵犯了原作者的版权,请告知一经查实,将立刻删除涉嫌侵权内容,联系我们QQ:712375056,同时欢迎投稿传递力量。
Copyright © 2009-2022 56dr.com. All Rights Reserved. 特网科技 特网云 版权所有 特网科技 粤ICP备16109289号
域名注册服务机构:阿里云计算有限公司(万网) 域名服务机构:烟台帝思普网络科技有限公司(DNSPod) CDN服务:阿里云计算有限公司 百度云 中国互联网举报中心 增值电信业务经营许可证B2
建议您使用Chrome、Firefox、Edge、IE10及以上版本和360等主流浏览器浏览本网站