Flink入门学习

Flink入门学习 Flink学习

官网: https://flink.apache.org/

1. Flink简介

Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.

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

1.2 为什么选择Flink

流数据 源源不断的数据

目标: 低延迟 高吞吐 准确性 容错性

1.3 Flink特点

事件驱动

有界流 使用DataSet

无界流 使用 DataStreamAPI

分层API

支持事件时间和处理时间

精确一次的状态一致性保证

低延迟 每秒百万个事件 毫秒级延迟

高可用

与众多常用存储系统的链接

1.4 Flink VS Spark Streaming

流处理 vs 微批处理

数据模型运行时架构

2.Flink 部署 2.1 Standalone单机

可以使用webui界面部署

也可以使用shell命令

#启动命令
/bin/start-cluster.sh
#停止
/bin/stop-cluster.sh

#提交任务
/bin/flink run -c [指定启动类] -p [并行度] [要提交的jar包地址] [指定jvm参数]
#查看当前所有作业
/bin/flink list
#取消作业
/bin/flink cancel [jobId]

2.2. Yarn

需要hadoop集群

没有安装条件 略

2.3 k8s

略

3. Flink 运行架构 3.1 运行时组件 3.1.1 作业管理器JobManager

3.1.2 任务管理器TaskManager

3.1.3 资源管理器ResourceManager

3.1.4 分发器Dispatcher

3.2 任务提交流程

3.3. 任务调度原理

3.4 Slot

并行度: 一个特定算子的子任务的个数称为其并行度

一般情况下,一个stream的并行度,可以认为就是其所有算子中最大的并行度

Slots 是指Flink计算中执行一个线程所需要资源(CPU,内存)的最小单元

所以Slot的数量一般设置为TaskManager(JVM)的核心数

Slot 有分组的概念

如果是不同的组,必须使用不同的Slot

3.5 程序与数据流DataFlow

Flinke程序分为三大块: Source transform sink

数据传输的形式:

One-to-one 必须是同共享组,并行度也相同的情况下才会One-to-oneRedistributing 重新分区 *** 作, 当并行度不一样时会进行重新分区轮询 *** 作

4. 流处理API

流处理过程

Environment => source => transform => sink

4.1 Environment

执行环境

//流处理执行环境
StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
// 批处理执行环境
 ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

//创建本地执行环境
ExecutionEnvironment.createLocalEnvironment([并行度]);

//创建远程执行环境
ExecutionEnvironment.createRemoteEnvironment(host,port,jar包地址 );

4.2 Source

Flink可以从不同数据源读取数据

4.2.1 从集合和元素中读取数据

API executionEnvironment.fromCollection(list);

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

        // 创建流处理执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度1
        executionEnvironment.setParallelism(1);

        // 创造集合数据
        List list = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            list.add(new SensorReading("Sensor" + i, LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli(), ThreadLocalRandom.current().nextDouble(35, 40)));
        }
        // 从集合中收集数据
        DataStreamSource sensorReadingDataStreamSource = executionEnvironment.fromCollection(list);
        // 打印集合数据
        sensorReadingDataStreamSource.print("sensor");

        // 从元素中收集数据
        DataStreamSource integerDataStreamSource = executionEnvironment.fromElements(1, 2, 3, 4, 56, 7);
        // 打印从元素中收集到数据
        integerDataStreamSource.print("element");

        // 执行Flink程序
        executionEnvironment.execute();

    }

4.2.2 从文件中读取数据

API executionEnvironment.readTextFile(inputPath);

public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        String inputPath = "E:\张尧\idea项目\FlinkTutorial\src\main\resources\word.txt";
 		DataStreamSource dataStreamSource = executionEnvironment.readTextFile(inputPath);

        SingleOutputStreamOperator> sum = dataStreamSource.flatMap(new WorkCount.MyFlagMapFunction()).keyBy(0).sum(1);
        sum.print();

        executionEnvironment.execute();
    }

4.2.3 从Kafka中读取数据 4.2.3.1 kafka配置

下载kafka 1.0.0版本以上

需要配置kafka的监听地址(本机除外)

修改config/server.properties

advertised.listeners=PLAINTEXT://192.168.164.205:9092

#启动kafka bin目录下
#启动zookeeper
./zookeeper-server-start.sh ../config/zookeeper.properties 
#启动kafka
./kafka-server-start.sh config/server.properties

package com.zy.flink.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;


public class KafkaSourceTest {

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

        // 创建kafka连接配置信息
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.164.205:9092");
//        properties.setProperty("group.id", "")

        // 创建流处理执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);

        // 从kafka从获取数据
        DataStreamSource dataStreamSource = executionEnvironment.addSource(new FlinkKafkaConsumer("sourcetest",
                new SimpleStringSchema(), properties));

        dataStreamSource.print();

        executionEnvironment.execute();
    }
}

4.2.4 自定义数据源

package com.zy.Flink.source;

import com.sun.org.apache.xpath.internal.operations.Bool;
import com.zy.Flink.entity.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;


public class UDFSourceTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        DataStreamSource sensorReadingDataStreamSource = executionEnvironment.addSource(new MySensorSource());
        sensorReadingDataStreamSource.print();
        executionEnvironment.execute();
    }


    public static class MySensorSource implements SourceFunction{

        //定义属性控制数据的生成
        private Boolean running = true;



        @Override
        public void run(SourceContext sourceContext) throws Exception {
            //定义传感器集合
            HashMap map = new HashMap<>();
            for (int i = 0; i < 10; i++) {
                map.put("sensor"+i, 60 + ThreadLocalRandom.current().nextGaussian() * 20);
            }
            while (running){
                for (String s : map.keySet()) {
                    sourceContext.collect(new SensorReading(s, LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli(),map.get(s)+ThreadLocalRandom.current().nextGaussian()));
                }
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

4.3 Transform

转换算子

4.3.1 基本转换算子

map flatMap filter 这三个是基本转换算子

package com.zy.Flink.transform;

import com.zy.Flink.entity.SensorReading;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;


public class TransormTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);

        // 创造集合数据
        List list = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            list.add(new SensorReading("Sensor" + i, LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli(), ThreadLocalRandom.current().nextDouble(35, 40)));
        }

        // 使用集合收集数据
        DataStreamSource sensorReadingDataStreamSource = executionEnvironment.fromCollection(list);


        // map转换 返回sensorReading的sensorId
        SingleOutputStreamOperator