本地jvm执行flink程序带web ui的操作

本地jvm执行flink程序带web ui的操作

目录本地jvm执行flink带web ui使用Flink 本地执行入门一、maven依赖二、本地执行三、实例

本地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);

Flink 本地执行入门

一、maven依赖

UTF-8

1.8

http://2.11.8

1.2.4

2.11

${java.version}

${java.version}

org.apache.flink

flink-clients_${scala.binary.version}

${flink.version}

二、本地执行

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 data = env.readTextFile("file:///Users/***/Documents/test.txt");

data.filter(new FilterFunction() {

@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 '")

// 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)

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。