技術標籤：Flink入門

宣告：本系列部落格是根據SGG的視訊整理而成，非常適合大家入門學習。

《2021年最新版大資料面試題全面開啟更新》

分離的 SQL 查詢

為定義端到端的 SQL 管道，SQL 的INSERT INTO語句可以向 Flink 叢集提交長時間執行的分離查詢。查詢產生的結果輸出到除 SQL 客戶端外的擴充套件系統中。這樣可以應對更高的併發和更多的資料。CLI 自身在提交後不對分離查詢做任何控制。

INSERT INTO MyTableSink SELECT * FROM MyTableSource

sinkMyTableSink必須在環境配置檔案中宣告。檢視更多關於 Flink 支援的外部系統及其配置資訊，參見

tables:
  - name: MyTableSink
    type: sink-table
    update-mode: append
    connector:
      property-version: 1
      type: kafka
      version: "0.11"
      topic: OutputTopic
      properties:
        bootstrap.servers: localhost:9092
        group.id: testGroup
    format:
      property-version: 1
      type: json
      derive-schema: true
    schema:
      - name: rideId
        data-type: BIGINT
      - name: lon
        data-type: FLOAT
      - name: lat
        data-type: FLOAT
      - name: rideTime
        data-type: TIMESTAMP(3)

SQL 客戶端要確保語句成功提交到叢集上。一旦提交查詢，CLI 將展示關於 Flink 作業的相關資訊。

[INFO] Table update statement has been successfully submitted to the cluster:
Cluster ID: StandaloneClusterId
Job ID: 6f922fe5cba87406ff23ae4a7bb79044
Web interface: http://localhost:8081

注意提交後，SQL 客戶端不追蹤正在執行的 Flink 作業狀態。提交後可以關閉 CLI 程序，並且不會影響分離的查詢。Flink 的

SQL 檢視

檢視是一張虛擬表，允許通過 SQL 查詢來定義。檢視的定義會被立即解析與驗證。然而，提交常規INSERT INTO或SELECT語句後不會立即執行，在訪問檢視時才會真正執行。

檢視可以用環境配置檔案或者 CLI 會話來定義。

下例展示如何在一個檔案裡定義多張檢視。檢視註冊的順序和定義它們的環境配置檔案一致。支援諸如檢視 A 依賴檢視 B ，檢視 B 依賴檢視 C的引用鏈。

tables:
  - name: MyTableSource
    # ...
  - name: MyRestrictedView
    type: view
    query: "SELECT MyField2 FROM MyTableSource"
  - name: MyComplexView
    type: view
    query: >
      SELECT MyField2 + 42, CAST(MyField1 AS VARCHAR)
      FROM MyTableSource
      WHERE MyField2 > 200

相較於 table soruce 和 sink，會話環境配置檔案中定義的檢視具有最高優先順序。

檢視還可以在 CLI 會話中用CREATE VIEW語句來建立：

CREATE VIEW MyNewView AS SELECT MyField2 FROM MyTableSource;

檢視能在 CLI 會話中建立，也能用DROP VIEW語句刪除：

DROP VIEW MyNewView;

注意CLI 中檢視的定義僅限於上述語法。將來版本會支援定義檢視結構以及在表名中加入轉義的空格。

臨時表（Temporal Table）

臨時表是在變化的歷史記錄表上的（引數化）檢視，該檢視在某個特定時間點返回表的內容。這對於在特定的時間戳將一張表的內容聯結另一張表是非常有用的。更多資訊見聯結臨時表頁面。

下例展示如何定義一張臨時表SourceTemporalTable：

tables:

  # 定義包含對臨時表的更新的 table source （或檢視）

  - name: HistorySource
    type: source-table
    update-mode: append
    connector: # ...
    format: # ...
    schema:
      - name: integerField
        data-type: INT
      - name: stringField
        data-type: STRING
      - name: rowtimeField
        data-type: TIMESTAMP(3)
        rowtime:
          timestamps:
            type: from-field
            from: rowtimeField
          watermarks:
            type: from-source

  # 在具有時間屬性和主鍵的變化歷史記錄表上定義臨時表
  - name: SourceTemporalTable
    type: temporal-table
    history-table: HistorySource
    primary-key: integerField
    time-attribute: rowtimeField  # could also be a proctime field

如例子中所示，table source，檢視和臨時表的定義可以相互混合。它們按照在環境配置檔案中定義的順序進行註冊。例如，臨時表可以引用一個檢視，該檢視依賴於另一個檢視或 table source。