Flink学习

1、任务提交到本地Flink运行:

    首先,启动本地的Flink(见前面的启动方式),浏览器登陆:localhost:8081 。然后,将写好的程序打包,然后将打包好的 .jar 包上传:

            

 上传后,如上图,箭头所指就是刚刚上传后的 jar包,点击这个 jar 包,可以编辑参数:

            

填写启动类、启动参数、并行度....... ,提交 submit 的时候,要先启动 Flink任务所需的流,比如本文使用的socket流: nc -lk 7777 ,启动的7777这个端口。

如果程序中使用的print答应,在这里如何查看:

            

 点击红框中的任务:

            

 在上面的 Stdout 即可看见输出。

2、本地连接 Kafka

首先,按照 Kafka,通过官网下载:https://kafka.apache.org/downloads,可参考博客:https://www.cnblogs.com/zhaoshizi/p/12154518.html

            

使用:tar -xzvf kafka_2.12-3.0.0.tgz 解压,进入解压后的目录,当前下载的kafka程序里自带Zookeeper,可以直接使用其自带的Zookeeper建立集群,kafka自带的Zookeeper程序使用 bin/zookeeper-server-start.sh,以及 bin/zookeeper-server-stop.sh来启动和停止Zookeeper。

启动zookeeper : 

./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties 

注意,必须要跟后面的 -daemon....,表示守护进程,不跟报错。

然后,启动 Kafka:

./bin/kafka-server-start.sh -daemon ./config/server.properties

同理,也必须要跟后面的 -daemon,守护进程....,

最后,启动 Kafka 的 producer,注册 topic,指定端口(broker-list指定集群中的一个或者多个服务器,一般我们再使用console producer的时候,这个参数是必备参数,另外一个必备的参数是topic ):

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic sensor

然后就可以发送数据了。

            

程序接收到的数据;

            

 程序的源码为:

public class SourceTest3_ReadKafka {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");

        DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<String>("sensor",new SimpleStringSchema(), properties));

        dataStream.print();
        env.execute();
    }
}

停止 Kafka :

./bin/kafka-server-stop.sh ./config/server.properties

停止Zookeeper :

./bin/zookeeper-server-stop.sh -daemon ./config/zookeeper.properties

使用 jps 查看是否还有 Kafka的进程,没有的话就是正常停止掉了。

Kafka去消费topic的数据:

首先启动Zookeeper、Kafka后,启动消费(bootstrap-servers指的是目标集群的服务器地址,这个和broker-list功能是一样的):

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sinktest

这里,消费的是本地的localhost,topic的名称是:sinktest

            

 

原文地址:https://www.cnblogs.com/luo-c/p/15499043.html