一官网介绍

1 Flink是什么

官网地址：https://flink.apache.org/

官网Flink培训地址：https://training.ververica.com/

image

Apache Flink 是一个框架和分布式处理引擎，用于在无界和有界数据流上进行有状态计算。Flink被设计为在所有常见的集群环境中运行，以内存中的速度和任意的规模执行计算。

下面是Flink技术的一个总览

image.png

2 为什么要学Flink

在大数据领域，批处理任务（有界数据流）与流处理任务（无界数据流）一般被认为是两种不同的任务，一个大数据的框架或计算引擎会被设计为只能处理其中一种任务。

例如Storm只支持流处理任务，而MapReduce、Spark只支持批处理任务，Spark Streaming是Apache Spark之上支持流处理任务的子系统，看似支持流处理，其实Spark Streaming 采用的是一种 micro-batch（微批）的架构，即把输入的数据流切分成细粒度的batch，并为梅一个batch提交一个批处理的Spark任务，所以Spark Streaming本质上还是基于Spark 的批处理系统对流式数据的进行处理，与Storm的完全流式处理方式是不一样的。

Flink 通过灵活的执行引擎，能够同时支持批处理任务和流处理任务。

二 Flink的流处理

流是数据的载体，例如web服务器的事件、证券交易所的交易、机器传感器的读数，这些数据都是作为流的一部分创建的。当我们分析数据时，我们可以围绕着有界或无界数据流。

image

1、批处理

是当我们处理有界数据流时起作用的范例。在这种操作模式中，我们可以选择在生成任何结果之前摄取整个数据集，这意味着可以对数据进行排序，计算全局统计数据，或者生成总结所有输入的最终报告。

2 流处理

即无界数据流，至少在概念上，输入可能永远不会结束，因此我们被迫在数据到达时继续处理。

在Flink中，应用程序由可由用户定义的操作转换的数据流组成。这些数据流形成有向图，从一个或多个源开始，以一个或多个接收器结束。

image

应用程序可能从流源(如消息队列或分布式日志)(如Apache Kafka或Kinesis)中使用实时数据。但是Flink也可以使用来自各种数据源的有界的历史数据。类似地，Flink应用程序产生的结果流可以发送到多种系统，并且可以通过RESTAPI访问Flink中的状态。

image

3 有状态

Flink的操作可以是有状态的。这意味着如何处理一个事件取决于它之前的所有事件的累积效果。状态可以用于简单的事情，例如每分钟在仪表板上显示的事件计数，或者用于更复杂的事情，例如欺诈检测模型的计算特性。

下面的图表显示了作业图中前三个操作符的并行性为2的作业，在并行性为1的接收器中结束。第三个运算符是有状态的，我们看到第二个和第三个运算符之间正在发生完全连接的网络洗牌。这样做是为了通过一些键对流进行分区，以便所有需要一起处理的事件都将被分割。

image

三 Flink的编程

基本步骤

image

下面是一个批处理的例子

package com.aii.bi.examples.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount {

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

        // 初始化运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 加载数据源
        DataSet<String> text = env.readTextFile("E:/opt/word.txt");

        // 数据转换
        DataSet<Tuple2<String, Integer>> counts = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            private static final long serialVersionUID = 1L;

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] tokens = value.split(" ");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        out.collect(new Tuple2<>(token, 1));
                    }
                }
            }
        }).groupBy(0).sum(1);

        //sink data
        counts.print();
    }

}

下面是一个读取kafka的流式场景的例子

package com.aii.bi.examples.streaming;

import java.util.Properties;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

public class KafkaStreamingPrint {

    public static void main(String[] args) throws Exception {
        
        //设置运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "10.21.35.124:21005,10.21.35.125:21005,10.21.35.126:21005");
        props.setProperty("auto.offset.reset", "earliest");
        props.setProperty("group.id", "test-001");
        
        FlinkKafkaConsumer010<String> consumer  = new FlinkKafkaConsumer010<>("flink_test03", new SimpleStringSchema(), props);
        
        DataStream<String> stream = env.addSource(consumer).map(new MapFunction<String, String>() {

            @Override
            public String map(String value) throws Exception {
                return value;
            }
        });
        
        stream.print();
        env.execute();
    }
}

1 运行环境

我们首先要获得已经存在的运行环境或者创建它。有3种方法得到运行环境：

（1）通过getExecutionEnvironment()获得；这将根据上下文得到运行环境，假如local模式，则它会创建一个local的运行环境；假      如是集群模式，则会创建一个分布式的运行环境；
（2）通过createLocalEnvironment() 创建一个本地的运行环境；
（3）通过createRemoteEnvironment (String host, int port, String, and .jar files)创建一个远程的运行环境。

2 数据源

Flink支持许多预定义的数据源，同时也支持自定义数据源。

2.1 基于文件

DataSet API支持读取文件数据：

你可以使用readTextFile(String path)来消费文件中的数据作为流数据的来源

//filePath The path of the file, as a URI (e.g., "file:///some/local/file" or "hdfs://host:port/file/path").
DataSet<String> text = env.readTextFile(String filePath);

2.2 基于socket

DataStream API支持从socket读取数据，有如下3个方法：

socketTextStream(hostName, port);
socketTextStream(hostName,port,delimiter)
socketTextStream(hostName,port,delimiter, maxRetry)

2.3 基于Kafka

DataStream API支持从Kafka读取数据，根据不同的kafka版本，对应的API不一样，以Kafka 0.10版本为例：

maven工程中pom文件的设置：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

代码样例：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "10.21.35.124:21005,10.21.35.125:21005,10.21.35.126:21005");
props.setProperty("auto.offset.reset", "earliest");
props.setProperty("group.id", "test-001");
        
FlinkKafkaConsumer010<String> consumer  = new FlinkKafkaConsumer010<>("flink_test03", new SimpleStringSchema(), props);

2.4 Transformation

Transformation允许将数据从一种形式转换为另一种形式，输入可以是1个源也可以是多个，输出则可以是0个、1个或者多个。下面我们一一介绍这些Transformations。

Flink学习之路（一）Flink初识

Flink学习之路（一）Flink初识

一官网介绍

1 Flink是什么

2 为什么要学Flink

二 Flink的流处理

1、批处理

2 流处理

3 有状态

三 Flink的编程

1 运行环境

2 数据源

2.1 基于文件

2.2 基于socket

2.3 基于Kafka

2.4 Transformation

Flink学习之路（一）Flink初识

一 官网介绍

1 Flink是什么

2 为什么要学Flink

二 Flink的流处理

1、批处理

2 流处理

3 有状态

三 Flink的编程

1 运行环境

2 数据源

2.1 基于文件

2.2 基于socket

2.3 基于Kafka

2.4 Transformation

一官网介绍