Apache Flink是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。

<properties>
    <encoding>UTF-8</encoding>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <java.version>1.8</java.version>
    <scala.version>2.12</scala.version>
    <flink.version>1.13.1</flink.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.16</version>
    </dependency>
    <!--flink客户端-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--scala版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-scala_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--flink-java版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--streamingጱscala版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--streamingጱjava版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--日志输出-->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.7</version>
        <scope>runtime</scope>
        </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
        <scope>runtime</scope>
    </dependency>
    <!--json依赖包-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.44</version>
    </dependency>
</dependencies>

新建log4j.properties，日志文件
### 配置appender名称
log4j.rootLogger = debugFile, errorFile
### debug级别以上的日志到：src/logs/debug.log
log4j.appender.debugFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.debugFile.File = src/logs/flink.log
log4j.appender.debugFile.Append = true
### Threshold属性指定输出等级
log4j.appender.debugFile.Threshold = info
log4j.appender.debugFile.layout = org.apache.log4j.PatternLayout
log4j.appender.debugFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n
### error级别以上的日志 src/logs/error.log
log4j.appender.errorFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.errorFile.File = src/logs/error.log
log4j.appender.errorFile.Append = true
log4j.appender.errorFile.Threshold = error
log4j.appender.errorFile.layout = org.apache.log4j.PatternLayout
log4j.appender.errorFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n

/**
 * @author lixiang
 */
public class Test {
    private static List<String> list = new ArrayList<>();
    static {
        list.add("SpringBoot,Docker");
        list.add("Netty,SpringCloud");
        list.add("Flink,Linux");
    }
    public static void test1() {
        Tuple3<Integer, String, Integer> tuple3 = Tuple3.of(1, "lixiang", 23);
        System.out.println(tuple3.f0);
        System.out.println(tuple3.f1);
        System.out.println(tuple3.f2);
    }
    public static void test2() {
        List<String> collect = list.stream().map(obj -> obj + "拼接").collect(Collectors.toList());
        System.out.println(collect);
    }
    public static void test3() {
        List<String> collect = list.stream().flatMap(obj -> Arrays.stream(obj.split(","))).collect(Collectors.toList());
        System.out.println(collect);
    }
}

public static void main(String[] args) {
     test1();
     test2();
     test3();
}

/**
 * @author lixiang
 */
public class FlinkDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口，存储全局相关的参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        //相同类型元素的数据流
        DataStream<String> stringDataStream = env.fromElements("Java,SpringBoot,SpringCloud", "Java,Linux,Docker");
        stringDataStream.print("处理前");
        //FlatMapFunction<Strings,String>，key是输入类型，value是Collector响应的收集的类型，看源码的注释也是DataStream<String>里面泛型类型
        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] arr = value.split(",");
                for (String s : arr) {
                    out.collect(s);
                }
            }
        });
        flatMapDataStream.print("处理后");
        //DataStream需要execute，可以取个名称
        env.execute("data stream job");
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

<!--Flink web ui-->
<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-runtime-web_${scala.version}</artifactId>
     <version>${flink.version}</version>
</dependency>

/**
 * flink UI demo
 * @author lixiang
 */
public class FlinkUI {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //设置并行度
        env.setParallelism(1);
        //监听192.168.139.80:8888输送过来的数据流
        DataStreamSource<String> stream = env.socketTextStream("192.168.139.80", 8888);
        //流处理
        SingleOutputStreamOperator<String> streamOperator = stream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    out.collect(s);
                }
            }
        });
        streamOperator.print("处理后");
        //执行任务
        env.execute("data stream job");
    }
}

/**
 * @author lixiang
 */
public class FlinkSourceDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> ds1 = env.fromElements("java,springboot", "kafka,redis", "openstack,k8s,docker");
        ds1.print("ds1");
        DataStreamSource<String> ds2 = env.fromCollection(Arrays.asList("hive", "hadoop", "hbase", "rabbitmq", "java"));
        ds2.print("ds2");
        DataStreamSource<Long> ds3 = env.fromSequence(0, 10);
        ds3.print("ds3");
        //执行任务
        env.execute("data job");
    }
}

/**
 * @author lixiang
 */
public class FlinkSourceDemo2 {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动的入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> textDS = env.readTextFile("E:\\软件资料\\log.txt");
        //DataStream<String> hdfsDS = env.readTextFile("hdfs://lixiang:8010/file/log/words.txt");
        textDS.print("textDS");
        env.execute("text job");
    }
}

/**
 * flink UI demo
 * @author lixiang
 */
public class FlinkUI {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //设置并行度
        env.setParallelism(1);
        //监听192.168.139.80:8888输送过来的数据流
        DataStreamSource<String> stream = env.socketTextStream("192.168.139.80", 8888);
        //流处理
        SingleOutputStreamOperator<String> streamOperator = stream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    out.collect(s);
                }
            }
        });
        streamOperator.print("处理后");
        //执行任务
        env.execute("data stream job");
    }
}

/**
 * 订单实体类
 * @author lixiang
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class VideoOrder
{
    private String tradeNo;
    private String title;
    private int money;
    private int userId;
    private Date createTime;
}

/**
 * @author lixiang
 */
public class VideoOrderSource extends RichParallelSourceFunction<VideoOrder>
{
    private volatile Boolean flag = true;
    private Random random = new Random();
    private static List<String> list = new ArrayList<>();
    static
    {
        list.add("SpringBoot2.x课程");
        list.add("Linux入到到精通");
        list.add("Flink流式技术课程");
        list.add("Kafka流式处理消息平台");
        list.add("微服务SpringCloud教程");
    }
    @Override
    public void run(SourceContext<VideoOrder> sourceContext) throws Exception {
        int x = 0;
        while (flag)
        {
            Thread.sleep(1000);
            String id = UUID.randomUUID().toString();
            int userId = random.nextInt(10);
            int money = random.nextInt(100);
            int videoNum = random.nextInt(list.size());
            String title = list.get(videoNum);
            sourceContext.collect(new VideoOrder(id, title, money, userId, new Date()));
            x++;
            if (x == 10)
            {
                cancel();
            }
        }
    }
    /**
     * 取消任务
     */
    @Override
    public void cancel()
    {
        flag = false;
    }
}

/**
 * @author lixiang
 */
public class FlinkMainSource {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<VideoOrder> source = env.addSource(new VideoOrderSource());
        source.print("接入的数据");
        SingleOutputStreamOperator<Integer> streamOperator = source.flatMap(new FlatMapFunction<VideoOrder, Integer>() {
            @Override
            public void flatMap(VideoOrder value, Collector<Integer> out) throws Exception {
                out.collect(value.getMoney());
            }
        });
        streamOperator.print("处理后");
        //流程启动
        env.execute("custom source job");
    }
}

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());