主頁 >  其他 > Flink - 更新中20220302

Flink - 更新中20220302

2022-03-02 20:10:26 其他

文章目錄

  • 前言
  • 一、概述
    • 1 Flink是什么
    • 2 架構分層
    • 3 基本組件
    • 4 其他流式計算框架
  • 二、入門與使用
    • 1 Flink基本安裝
      • 1.1 Linux
      • 1.2 Java
      • 1.3 Scala（待補充）
      • 1.4 集群模式
    • 2 常用API
      • 2.1 DataStream 流處理
        • DataSource
        • Transformation
        • Sink
        • 示例一：自定義資料源（SourceFunction）
        • 示例二：自定義磁區
        • 示例三：Socket通信示例
        • 示例四：RabbitMQ作為資料源
        • 示例五：自定義Sink
      • 2.2 DataSet 批處理
      • 2.3 Table API / SQL（待補充）
      • 2.4 關于序列化
  • 三、進階使用
  • 四、原理決議
  • 總結

前言

目前本人是Java開發工程師，所以里面大部分的學習筆記都是以Java代碼為主，Scala后面我再學所以后續再進行補充，

文獻
《Flink入門與實戰》 - 徐葳
/

一、概述

1 Flink是什么

? Apache Flink，內部是用Java及Scala撰寫的分布式流資料計算引擎，可以支持以批處理或流處理的方式處理資料，在2014年這個專案被Apache范訓器所接受后，Flink迅速成為ASF（ApacheSoftware Foundation）的頂級專案之一，在2019年1月，阿里巴巴集團收購了Flink創始公司（DataArtisans），打造了阿里云商業化的實時計算Flink產品，

它有如下幾個特點

  1. 低延遲
  2. 高吞吐
  3. 支持有界資料/無界資料的處理，資料流式計算
  4. 支持集群，支持HA，可靠性強

什么是有界資料/無界資料？

  • 有界資料：資料是有限的，一條SELECT查詢下的資料不會是源源不斷的
  • 無界資料：資料源源不斷，不知道為什么時候結束，例如監控下的告警

2 架構分層

名稱描述
Deploy 部署方式本地/集群/云服務部署，
Core 分布式流處理模型計算核心實作，為API層提供基礎服務，
API 呼叫介面提供面向無界資料的流處理API及有界資料的批處理API，其中流處理對應DataStream API，批處理對應DataSet API，
Library 應用層提供應用計算框架，面向流處理支持CEP（復雜事件處理）、基于SQL-like的操作（基于Table的關系操作），面向批處理支持FlinkML（機器學習庫）、Gelly（圖處理）、Table 操作，

3 基本組件

一個Flink任務 = DataSource + Transformation + DataSink

DataSource ：資料源

Transformation ：資料處理

DataSink：計算結果輸出

而Flink在網路傳輸中通過快取塊承載資料，可以通過設定快取塊的超時時間，變相的決定了資料在網路中的處理方式，

4 其他流式計算框架

文章目錄
Flink介紹、特點及和與其他大資料框架對比_zhangxm_qz的CSDN博客

二、入門與使用

1 Flink基本安裝

1.1 Linux

下載鏈接
Index of /dist/flink/flink-1.14.3 (apache.org)

首先去apache官網下載部署的軟體包，下載完成之后進行解壓

## 解壓
tar -zxvf flink-1.14.3-bin-scala_2.12.tgz 
## 進入bin目錄 啟動
./start-cluster.sh
## Flink提供的WebUI的埠是8081 此時可以去看看是否啟動完成
netstat -anp |grep 8081

接著通過頁面訪問8081埠來個初體驗

關于Linux下的Flink Shell終端的使用

文章目錄
flink~使用shell終端_cai_and_luo的博客-CSDN博客

1.2 Java

文章目錄
Flink入門之Flink程式開發步驟（java語言）_胖虎兒的博客-CSDN博客

匯入依賴

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.12</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>1.14.3</version>
</dependency>

入門Demo

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
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;

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

        /**
         * 大致的流程就分為
         * 1.環境準備
         * 設定運行模式
         * 2.加載資料源
         * 3.資料轉換
         * 4.資料輸出
         * 5.執行程式
         */

        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 2.加載資料源
        DataStreamSource<String> elementsSource = env.fromElements("java,scala,php,c++",
                "java,scala,php", "java,scala", "java");
        // 3.資料轉換
        DataStream<String> flatMap = elementsSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String element, Collector<String> out) throws Exception {
                String[] wordArr = element.split(",");
                for (String word : wordArr) {
                    out.collect(word);
                }
            }
        });
        // DataStream 下邊為DataStream子類
        SingleOutputStreamOperator<String> source = flatMap.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value.toUpperCase();
            }
        });
        // 4.資料輸出
        source.print();
        // 5.執行程式
        env.execute();
    }
}

關于在設定運行模式的代碼上，有三種選擇

/**
 * Runtime execution mode of DataStream programs. Among other things, this controls task scheduling,
 * network shuffle behavior, and time semantics. Some operations will also change their record
 * emission behaviour based on the configured execution mode.
 *
 * @see <a
 *     href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-134%3A+Batch+execution+for+the+DataStream+API">
 *     https://cwiki.apache.org/confluence/display/FLINK/FLIP-134%3A+Batch+execution+for+the+DataStream+API</a>
 */
@PublicEvolving
public enum RuntimeExecutionMode {

    /**
     * The Pipeline will be executed with Streaming Semantics. All tasks will be deployed before
     * execution starts, checkpoints will be enabled, and both processing and event time will be
     * fully supported.
     */
     /** 流處理模式 */
    STREAMING,

    /**
     * The Pipeline will be executed with Batch Semantics. Tasks will be scheduled gradually based
     * on the scheduling region they belong, shuffles between regions will be blocking, watermarks
     * are assumed to be "perfect" i.e. no late data, and processing time is assumed to not advance
     * during execution.
     */
     /** 批處理模式 */
    BATCH,

    /**
     * Flink will set the execution mode to {@link RuntimeExecutionMode#BATCH} if all sources are
     * bounded, or {@link RuntimeExecutionMode#STREAMING} if there is at least one source which is
     * unbounded.
     */
    /** 自動模式 */
    AUTOMATIC
}

1.3 Scala（待補充）

與Java一樣都在IDEA編譯器上做，此時引入依賴

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-scala -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-scala_2.12</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.12</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.12</artifactId>
    <version>1.14.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>1.14.3</version>
</dependency>

// …

待定 …

// …

1.4 集群模式

文章目錄
Flink集群部署詳細步驟 - 簡書 (jianshu.com)
Flink集群部署 - 云+社區 - 騰訊云 (tencent.com)

2 常用API

第一次學時，光看上面的Demo例子比較難以理解，所以通過書下面的API內容對照上面的Demo來進行理解，先來了解Flink四種層次的API詳情

層級描述資訊備注
底層 API偏底層，易用性比較差，提供時間/狀態的細粒度控制Stateful Stream Processing
核心 API對有界/無界資料提供處理方法DataStream（流處理） / DataSet（批處理）
Table API/宣告式DSL
SQL/高級語言

2.1 DataStream 流處理

主要分為三個流程

  1. DataSource 資料輸入：addSource(sourceFunction)為程式添加一個資料源，
  2. Transformation 資料處理：對一個或多個資料源進行操作，
  3. Sink 資料輸出：通過Transformation 處理后的資料輸出到指定的位置，

DataSource

看看他們的API

DataSource API描述
readTextFile(檔案路徑)逐行讀取文本檔案的資料
socketTextStream(地址資訊)從socket中讀取資料
fromCollection(集合資料)從集合內獲取資料
其他第三方輸入資料…或者自定義資料源通過Flink提供的內置連接器去鏈接其它資料源

如果是自定義資料源，有兩種實作方式

  1. 實作SourceFunction介面（并行度為1 = 無并行度）
  2. 實作ParallelSourceFunction介面 / 繼承RichParallelSourceFunction

什么是并行度？

? 一個Flink程式由多個任務（Source、Transformation和Sink）組成，一個任務由多個并行實體（執行緒）來執行，一個任務的并行實體（執行緒）數目被稱為該任務的并行度，

Transformation

接下來是Transformation資料處理，Flink針對DataStream提供了大量的已經實作的算子，

DataStream API描述
Map輸入一個元素，然后回傳一個元素，中間可以進行清洗轉換等操作
FlatMap輸入一個元素，可以回傳零個、一個或者多個元素
Filter過濾函式，對傳入的資料進行判斷，符合條件的資料會被留下
KeyBy根據指定的Key進行分組，Key相同的資料會進入同一個磁區，典型用法如下：1、DataStream.keyBy(“someKey”) 指定物件中的someKey段作為分組Key，2、DataStream.keyBy(0) 指定Tuple中的第一個元素作為分組Key，
Reduce對資料進行聚合操作，結合當前元素和上一次Reduce回傳的值進行聚合操作，然后回傳一個新的值
Aggregationssum()、min()、max()等
Union合并多個流，新的流會包含所有流中的資料，但是Union有一個限制，就是所有合并的流型別必須是一致的
Connect和Union類似，但是只能連接兩個流，兩個流的資料型別可以不同，會對兩個流中的資料應用不同的處理方法
coMap和coFlatMap在ConnectedStream中需要使用這種函式，類似于Map和flatMap
Split根據規則把一個資料流切分為多個流
Select和Split配合使用，選擇切分后的流

關于Flink針對DataStream提供的一些資料磁區規則

磁區規則描述
DataStream.shuffle()隨機磁區
DataStream.rebalance()對資料集進行再平衡、重磁區和消除資料傾斜
DataStream.rescale()重新調節
DataStream.partitionCustom(partitioner,0) 或者 DataStream.partitionCustom(partitioner,“smeKey”)自定義磁區

Sink

資料處理后的輸出

Sink API描述
writeAsText()將元素以字串形式逐行寫入，這些字串通過呼叫每個元素的toString()方法來獲取
print() / printToErr()列印每個元素的toString()方法的值到標準輸出或者標準錯誤輸出流中
自定義輸出addSink可以實作把資料輸出到第三方存盤介質中，系統提供了一批內置的Connector，它們會提供對應的Sink支持

自定義Sink的兩種方式

  1. 實作SinkFunction介面
  2. 繼承RichSinkFunction類

實際上，RichSinkFunction抽象類也是繼承了SinkFunction這個介面，所以實際上差別不大

示例一：自定義資料源（SourceFunction）

第一步，繼承SourceFunction介面，實作自定義資料源類

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;

/**
 * 自定義資料源
 * @author 李家民
 */
public class DemoTransactionSource implements SourceFunction<String> {
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (true) {
            // 發射元素
            ctx.collect(String.valueOf(new Random().nextInt(50)
            ));
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
    }
}

第二步，在Flink代碼中引入這個資料源

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
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 org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;

/**
 * @author 李家民
 */
@Component
public class FlinkInitialize {

    @PostConstruct
    public void starter() throws Exception {
        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 2.設定自定義資料源
        DataStreamSource<String> stringDataStreamSource = env.addSource(new DemoTransactionSource(), "測驗用的資料源");

        // 3.資料處理
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = stringDataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value;
            }
        });

        // 4.資料輸出
        stringSingleOutputStreamOperator.print();

        // 5.執行程式
        env.execute();
    }
}

此時執行代碼，就可以把引入的資料進行列印

SourceFunction定義了run和cancel兩個方法和SourceContext內部介面，

  • run(SourceContex)：實作資料獲取邏輯，并可以通過傳入的引數ctx進行向下游節點的資料轉發，
  • cancel()：用來取消資料源，一般在run方法中，會存在一個回圈來持續產生資料，cancel方法則可以使該回圈終止，
  • SourceContext：source函式用于發出元素和可能的watermark的介面，回傳source生成的元素的型別，

示例二：自定義磁區

資料源沿用上述案例的代碼，自定義磁區是通過實作Partitioner介面去做處理

首先看看自定義磁區的實作類

/**
 * 自定義磁區
 * @author 李家民
 */
public class DemoPartitioner implements Partitioner<String> {
    @Override
    public int partition(String key, int numPartitions) {
        System.out.println("目前磁區總數=" + numPartitions + "  當前值=" + key + "  通過最左邊的值看磁區號");

        if (new Integer(key) > 20) {
            return 1;
        } else {
            return 2;
        }
    }
}

然后在Flink的代碼中體現

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;

@Component
public class FlinkInitialize {
    @PostConstruct
    public void starter() throws Exception {
        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 2.設定自定義資料源
        DataStreamSource<String> stringDataStreamSource = env.addSource(new DemoTransactionSource(), "測驗用的資料源");
        // 3.資料處理
        DataStream<String> dataStream = stringDataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value;
            }
        }).partitionCustom(new DemoPartitioner(), new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return value;
            }
        });
        // 4.資料輸出
        dataStream.print();
        // 5.執行程式
        env.execute();
    }
}

輸出后的結果如下

示例三：Socket通信示例

第一步：搭建資料來源，這里使用Linux作為資料來源，在Linux上打命令把埠開啟

nc -l 16668

第二步：撰寫flink代碼

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;

@Component
public class FlinkInitialize {

    @PostConstruct
    public void starter() throws Exception {
        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // 2.設定自定義資料源
        String address = "47.106.207.254";
        int port = 16668;
        DataStream<String> dataStreamSource = env.socketTextStream(address, port).map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value;
            }
        });
        dataStreamSource.print();

        // 5.執行程式
        env.execute();

    }
}

效果如下

你學廢了嗎

示例四：RabbitMQ作為資料源

第一步：搭建RabbitMQ子系統

.....代碼省略，不會RabbitMQ的看下面這篇文章
文章目錄
RabbitMQ - SpringBoot集成版 - 開發+運維__-CSDN博客

第二步：撰寫flink代碼，首先引入RabbitMQ/Flink的依賴

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-rabbitmq -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq_2.12</artifactId>
    <version>1.14.3</version>
</dependency>

撰寫java代碼

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;

@Component
public class FlinkInitialize {

    @PostConstruct
    public void starter() throws Exception {
        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 2.設定資料源
        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("47.106.207.254")
                .setPort(5672)
                .setUserName("admin")
                .setPassword("admin")
                .setVirtualHost("/")
                .build();

        // 3.將RabbitMQ資料源加入
        DataStreamSource<String> dataStreamSource = env.addSource(
                new RMQSource<String>(
                        connectionConfig,
                        "Demo01_queue",
                        true,
                        new SimpleStringSchema()));

        // 4.資料轉換并輸出
        dataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value;
            }
        });
        dataStreamSource.print();

        // 5.執行程式
        env.execute();
    }
}

在Flink代碼中，有兩步對于RabbitMQ的加入很關鍵

示例五：自定義Sink

很簡單，把上面的代碼稍微改一下就好了

package com.ljm.flink;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;

/**
 * @author 李家民
 */
@Component
public class FlinkInitialize {

    @PostConstruct
    public void starter() throws Exception {
        // 1.準備環境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 設定運行模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 2.設定資料源
        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("47.106.207.254")
                .setPort(5672)
                .setUserName("admin")
                .setPassword("admin")
                .setVirtualHost("/")
                .build();

        // 3.將RabbitMQ資料源加入
        DataStreamSource<String> dataStreamSource = env.addSource(
                new RMQSource<String>(
                        connectionConfig,
                        "Demo01_queue",
                        true,
                        new SimpleStringSchema()));

        // 4.資料轉換并輸出
        dataStreamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return value;
            }
        });
        
        // 自定義輸出
        dataStreamSource.addSink(new SinkDemo());

        // 5.執行程式
        env.execute();

    }
}

繼承RichSinkFunction抽象類

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 自定義Flink輸出
 * @author 李家民
 */
public class SinkDemo extends RichSinkFunction<String> {

    /**
     * 將給定值寫入接收器，為每條記錄呼叫此函式
     * @param value 獲取到的值
     * @param context 可用于獲取有關輸入記錄的附加資料的背景關系
     * @throws Exception
     */
    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.println(value + "   " + context.timestamp());
    }

    @Override
    public void writeWatermark(Watermark watermark) throws Exception {
        super.writeWatermark(watermark);
    }

    /**
     * 此方法在資料處理結束時呼叫
     * @throws Exception
     */
    @Override
    public void finish() throws Exception {
        System.out.println("此方法在資料處理結束時呼叫");
    }
}

接收到資料以后，就可以進行后續的一系列操作了

2.2 DataSet 批處理

組件跟上面的DataStream差不多，都是分為這么三個，

  1. DataSource
  2. Transformation
  3. Sink

一般是用來讀取HDFS（分布式檔案存盤）中的檔案資料，不作解釋了，

2.3 Table API / SQL（待補充）

Flink針對標準的流處理和批處理提供的兩種關系型API：Table API 和 SQL，

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table</artifactId>
    <version>1.14.3</version>
    <type>pom</type>
    <scope>provided</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-common -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>1.14.3</version>
    <scope>provided</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java</artifactId>
    <version>1.14.3</version>
</dependency>

// …

待定 …

// …

2.4 關于序列化

Flink自帶針對一些標準型別的序列化器，如果涉及到這些自帶的序列化器也無法處理的資料，則需要自定義序列化器，

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 使用Avro序列化
env.getConfig().enableForceAvro();
// 使用Kryo序列化
env.getConfig().enableForceKryo();
// 自定義序列化器
env.getConfig().addDefaultKryoSerializer(xxxxx,xxxxx);

在自定義序列化器引數中，需要填寫序列化的類物件類，并且這個類切記需要繼承序列化介面Serializer，

三、進階使用

1

四、原理決議

1

總結

提示：這里對文章進行總結：
例如：以上就是今天要講的內容，本文僅僅簡單介紹了pandas的使用，而pandas提供了大量能使我們快速便捷地處理資料的函式和方法，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/435947.html

標籤：其他

上一篇：React原生地圖，animateCamera在ios上有offset

下一篇：利用msfvenom生成木馬檔案

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • 網閘典型架構簡述

    網閘架構一般分為兩種：三主機的三系統架構網閘和雙主機的2+1架構網閘。 三主機架構分別為內端機、外端機和仲裁機。三機無論從軟體和硬體上均各自獨立。首先從硬體上來看，三機都用各自獨立的主板、記憶體及存盤設備。從軟體上來看，三機有各自獨立的作業系統。這樣能達到完全的三機獨立。對于“2+1”系統，“2”分為 ......

    uj5u.com 2020-09-10 02:00:44 more
  • 如何從xshell上傳檔案到centos linux虛擬機里

    如何從xshell上傳檔案到centos linux虛擬機里及：虛擬機CentOs下執行 yum -y install lrzsz命令，出現錯誤：鏡像無法找到軟體包 前言 一、安裝lrzsz步驟 二、上傳檔案 三、遇到的問題及解決方案 總結 前言 提示：其實很簡單，往虛擬機上安裝一個上傳檔案的工具 ......

    uj5u.com 2020-09-10 02:00:47 more
  • 一、SQLMAP入門

    一、SQLMAP入門 1、判斷是否存在注入 sqlmap.py -u 網址/id=1 id=1不可缺少。當注入點后面的引數大于兩個時。需要加雙引號， sqlmap.py -u "網址/id=1&uid=1" 2、判斷文本中的請求是否存在注入 從文本中加載http請求，SQLMAP可以從一個文本檔案中 ......

    uj5u.com 2020-09-10 02:00:50 more
  • Metasploit 簡單使用教程

    metasploit 簡單使用教程 浩先生， 2020-08-28 16:18:25 分類專欄： kail 網路安全 linux 文章標簽： linux資訊安全 編輯 著作權 metasploit 使用教程 前言 一、Metasploit是什么？ 二、準備作業 三、具體步驟 前言 Msfconsole ......

    uj5u.com 2020-09-10 02:00:53 more
  • 游戲逆向之驅動層與用戶層通訊

    驅動層代碼： #pragma once #include <ntifs.h> #define add_code CTL_CODE(FILE_DEVICE_UNKNOWN,0x800,METHOD_BUFFERED,FILE_ANY_ACCESS) /* 更多游戲逆向視頻www.yxfzedu.com ......

    uj5u.com 2020-09-10 02:00:56 more
  • 北斗電力時鐘（北斗授時服務器）讓網路資料更精準

    北斗電力時鐘（北斗授時服務器）讓網路資料更精準 北斗電力時鐘（北斗授時服務器）讓網路資料更精準 京準電子科技官微——ahjzsz 近幾年，資訊技術的得了快速發展，互聯網在逐漸普及，其在人們生活和生產中都得到了廣泛應用，并且取得了不錯的應用效果。計算機網路資訊在電力系統中的應用，一方面使電力系統的運行 ......

    uj5u.com 2020-09-10 02:01:03 more
  • 【CTF】CTFHub 技能樹 彩蛋 writeup

    ?碎碎念 CTFHub：https://www.ctfhub.com/ 筆者入門CTF時時剛開始刷的是bugku的舊平臺，后來才有了CTFHub。 感覺不論是網頁UI設計，還是題目質量，賽事跟蹤，工具軟體都做得很不錯。 而且因為獨到的金幣制度的確讓人有一種想去刷題賺金幣的感覺。 個人還是非常喜歡這個 ......

    uj5u.com 2020-09-10 02:04:05 more
  • 02windows基礎操作

    我學到了一下幾點 Windows系統目錄結構與滲透的作用 常見Windows的服務詳解 Windows埠詳解 常用的Windows注冊表詳解 hacker DOS命令詳解（net user / type /md /rd/ dir /cd /net use copy、批處理 等） 利用dos命令制作 ......

    uj5u.com 2020-09-10 02:04:18 more
  • 03.Linux基礎操作

    我學到了以下幾點 01Linux系統介紹02系統安裝，密碼啊破解03Linux常用命令04LAMP 01LINUX windows： win03 8 12 16 19 配置不繁瑣 Linux：redhat,centos(紅帽社區版)，Ubuntu server,suse unix:金融機構，證券，銀 ......

    uj5u.com 2020-09-10 02:04:30 more
  • 05HTML

    01HTML介紹 02頭部標簽講解03基礎標簽講解04表單標簽講解 HTML前段語言 js1.了解代碼2.根據代碼 懂得挖掘漏洞 （POST注入/XSS漏洞上傳）3.黑帽seo 白帽seo 客戶網站被黑帽植入劫持代碼如何處理4.熟悉html表單 <html><head><title>TDK標題，描述 ......

    uj5u.com 2020-09-10 02:04:36 more
最新发布
  • 2023年最新微信小程式抓包教程

    01 開門見山 隔一個月發一篇文章，不過分。 首先回顧一下《微信系結手機號資料庫被脫庫事件》，我也是第一時間得知了這個訊息，然后跟蹤了整件事情的經過。下面是這起事件的相關截圖以及近日流出的一萬條資料樣本： 個人認為這件事也沒什么，還不如關注一下之前45億快遞資料查詢渠道疑似在近日復活的訊息。 訊息是 ......

    uj5u.com 2023-04-20 08:48:24 more
  • web3 產品介紹：metamask 錢包 使用最多的瀏覽器插件錢包

    Metamask錢包是一種基于區塊鏈技術的數字貨幣錢包，它允許用戶在安全、便捷的環境下管理自己的加密資產。Metamask錢包是以太坊生態系統中最流行的錢包之一，它具有易于使用、安全性高和功能強大等優點。 本文將詳細介紹Metamask錢包的功能和使用方法。 一、 Metamask錢包的功能 數字資 ......

    uj5u.com 2023-04-20 08:47:46 more
  • vulnhub_Earth

    前言 靶機地址->>>vulnhub_Earth 攻擊機ip：192.168.20.121 靶機ip：192.168.20.122 參考文章 https://www.cnblogs.com/Jing-X/archive/2022/04/03/16097695.html https://www.cnb ......

    uj5u.com 2023-04-20 07:46:20 more
  • 從4k到42k，軟體測驗工程師的漲薪史，給我看哭了

    清明節一過，盲猜大家已經無心上班，在數著日子準備過五一，但一想到銀行卡里的余額……瞬間心情就不美麗了。最近，2023年高校畢業生就業調查顯示，本科畢業月平均起薪為5825元。調查一出，便有很多同學表示自己又被平均了。看著這一資料，不免讓人想到前不久中國青年報的一項調查：近六成大學生認為畢業10年內會 ......

    uj5u.com 2023-04-20 07:44:00 more
  • 最新版本 Stable Diffusion 開源 AI 繪畫工具之中文自動提詞篇

    🎈 標簽生成器 由于輸入正向提示詞 prompt 和反向提示詞 negative prompt 都是使用英文，所以對學習母語的我們非常不友好 使用網址：https://tinygeeker.github.io/p/ai-prompt-generator 這個網址是為了讓大家在使用 AI 繪畫的時候 ......

    uj5u.com 2023-04-20 07:43:36 more
  • 漫談前端自動化測驗演進之路及測驗工具分析

    隨著前端技術的不斷發展和應用程式的日益復雜，前端自動化測驗也在不斷演進。隨著 Web 應用程式變得越來越復雜，自動化測驗的需求也越來越高。如今，自動化測驗已經成為 Web 應用程式開發程序中不可或缺的一部分，它們可以幫助開發人員更快地發現和修復錯誤，提高應用程式的性能和可靠性。 ......

    uj5u.com 2023-04-20 07:43:16 more
  • CANN開發實踐：4個DVPP記憶體問題的典型案例解讀

    摘要：由于DVPP媒體資料處理功能對存放輸入、輸出資料的記憶體有更高的要求（例如，記憶體首地址128位元組對齊），因此需呼叫專用的記憶體申請介面，那么本期就分享幾個關于DVPP記憶體問題的典型案例，并給出原因分析及解決方法。 本文分享自華為云社區《FAQ_DVPP記憶體問題案例》，作者：昇騰CANN。 DVPP ......

    uj5u.com 2023-04-20 07:43:03 more
  • msf學習

    msf學習 以kali自帶的msf為例 一、msf核心模塊與功能 msf模塊都放在/usr/share/metasploit-framework/modules目錄下 1、auxiliary 輔助模塊，輔助滲透（埠掃描、登錄密碼爆破、漏洞驗證等） 2、encoders 編碼器模塊，主要包含各種編碼 ......

    uj5u.com 2023-04-20 07:42:59 more
  • Halcon軟體安裝與界面簡介

    1. 下載Halcon17版本到到本地 2. 雙擊安裝包后 3. 步驟如下 1.2 Halcon軟體安裝 界面分為四大塊 1. Halcon的五個助手 1) 影像采集助手：與相機連接，設定相機引數，采集影像 2) 標定助手：九點標定或是其它的標定，生成標定檔案及內參外參，可以將像素單位轉換為長度單位 ......

    uj5u.com 2023-04-20 07:42:17 more
  • 在MacOS下使用Unity3D開發游戲

    第一次發博客，先發一下我的游戲開發環境吧。 去年2月份買了一臺MacBookPro2021 M1pro(以下簡稱mbp)，這一年來一直在用mbp開發游戲。我大致分享一下我的開發工具以及使用體驗。 1、Unity 官網鏈接： https://unity.cn/releases 我一般使用的Apple ......

    uj5u.com 2023-04-20 07:40:19 more

Copyright © 2010-2020 有解無憂