测试版本

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>javabook.apache</artifactId>
  7.         <groupId>org.example</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>javabook.apache.flink</artifactId>
  14.     <properties>
  15.         <maven.compiler.source>8</maven.compiler.source>
  16.         <maven.compiler.target>8</maven.compiler.target>
  17.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.apache.flink</groupId>
  22.             <artifactId>flink-java</artifactId>
  23.             <version>1.13.1</version>
  24.         </dependency>
  26.         <dependency>
  27.             <groupId>org.apache.flink</groupId>
  28.             <artifactId>flink-clients_2.12</artifactId>
  29.             <version>1.13.1</version>
  30.         </dependency>
  32.         <dependency>
  33.             <groupId>org.apache.flink</groupId>
  34.             <artifactId>flink-streaming-java_2.12</artifactId>
  35.             <version>1.13.1</version>
  36.             <scope>provided</scope>
  37.         </dependency>
  38.     </dependencies>
  39. </project>

测试代码

package online.javabook.flink.quickstart;

import online.javabook.flink.quickstart.flatmapper.MyFlatMapper;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * nc -lkv -p 7777
 */
public class SocketWordCount {
    public static void main(String[] args) throws Exception{

        // env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        ParameterTool parameterTool = ParameterTool.fromArgs(args);

        // paras
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");

        // inputDataStream
        DataStream<String> inputDataStream = env.socketTextStream(host, port);

        // outputDataStream
        DataStream<Tuple2<String, Integer>> outputDataStream = inputDataStream.flatMap( new MyFlatMapper())
                .keyBy(0)
                .sum(1);

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

上传代码

image.png

监听数据

[java@node101 ~]$ nc -lkv -p 7777
Ncat: Version 7.70 ( https://nmap.org/ncat )
Ncat: Listening on :::7777
Ncat: Listening on 0.0.0.0:7777
hello world

image.png

Flink命令

命令行提交job

[java@node101 bin]$ ./flink run -c online.javabook.flink.quickstart.SocketWordCount -p 3 /target/javabook.apache.flink-1.0-SNAPSHOT.jar --host node101 --port 7777

取消运行中的iob

[java@node101 bin]$ ./flink cancle [jobId]

查看运行中的iob

## 运行状态的job
[java@node101 bin]$ ./flink list

## 所有状态的job
[java@node101 bin]$ ./flink list -a