工業(yè)互聯(lián)網(wǎng)平臺實(shí)時數(shù)據(jù)開發(fā)《工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用》/

工業(yè)設(shè)備實(shí)時數(shù)據(jù)開發(fā)《工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用》/知識準(zhǔn)備任務(wù)實(shí)施目錄0102?數(shù)據(jù)庫定義語言(DDL)：用來定義數(shù)據(jù)庫模式。DDL主要包括數(shù)據(jù)庫模式定義、數(shù)據(jù)庫存儲結(jié)構(gòu)、存取方法定義、數(shù)據(jù)庫模式修改和刪除功能。?數(shù)據(jù)庫操縱語言(DML)：用來表示用戶對數(shù)據(jù)庫的操作請求

，功能主要包括查詢數(shù)據(jù)庫的查詢、刪除、修改、新增功能?數(shù)據(jù)查詢語言(DQL)：用來查詢數(shù)據(jù)庫中表的記錄

，關(guān)鍵字：select，where等?數(shù)據(jù)控制語言(DCL)：用來定義數(shù)據(jù)庫的訪問權(quán)限和安全級別。SQL(Structured

Query

Lanauage)結(jié)構(gòu)化查詢語言是關(guān)系數(shù)據(jù)庫中最普遍使用的語言。主要包括查詢、數(shù)據(jù)操縱、數(shù)據(jù)定義、數(shù)據(jù)控制功能，是一種通用的、功能強(qiáng)大的關(guān)系數(shù)據(jù)庫的標(biāo)準(zhǔn)語言概念介紹SQL語言分類SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用示例：SELECT*FROM

Table；//取出表中的所有列SELECTname，ageFROMTable；//取出表中name

和age兩列與此同時SELECT

語句中可以使用函數(shù)和別名，例如

WordCount

中：SELECT

word,COUNT(word)FROM

table

GROUP

BY

word;1)

SELECT：

SELECT用于從DataSet/DataStream中選擇數(shù)據(jù)，用于篩選出某些列。SQL常用算子SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用示例：SELECTname，ageFROMTablewherenameLIKE‘%小明%’；SELECT*FROM

TableWHEREage=20；WHERE

是從原數(shù)據(jù)中進(jìn)行過濾，那么在WHERE

條件中，F(xiàn)link

SQL

同樣支持=

、<、>、<>、>=、<=，以及AND

、OR

等表達(dá)式的組合，最終滿足過濾條件的數(shù)據(jù)會被選擇出來。并且WHERE

可以結(jié)合IN

、NOT

IN

聯(lián)合使用。舉個例子：SELECT

name,ageFROMTableWHERE

name

IN(SELECTnameFROMTable2)2)

WHERE：WHERE用于從數(shù)據(jù)集/流中過濾數(shù)據(jù)，與SELECT一起使用，用于根據(jù)某些條件對

關(guān)系做水平分割

，即選擇符合條件的記錄。SQL常用算子SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用示例：SELECT

DISTINCT

name

FROM

Table;對于流式查詢，計算查詢結(jié)果所需的State

可能會無限增長，用戶需要自己控制查詢的狀態(tài)范圍，以防止?fàn)顟B(tài)過大。3)

DISTINCT：

DISTINCT

用于從數(shù)據(jù)集/流中根據(jù)

SELECT

的結(jié)果進(jìn)行去重。SQL常用算子SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用示例：SELECT

name,SUM(score)as

TotalScore

FROM

Table

GROUP

BYname;4)

GROUP

BY：

GROUP

BY

是對數(shù)據(jù)進(jìn)行分組操作。例如：計算成績明細(xì)表中每個學(xué)生的總分。5)

UNION

和

UNION

ALL：

UNION

用于將兩個結(jié)果集合并起來

，要求兩個結(jié)果集字段完全一

致

，包括字段類型、字段順序。不同于UNION

ALL的是，UNION

會對結(jié)果數(shù)據(jù)去重。示例：SELECT*FROM

T1UNION(ALL)SELECT*FROM

T2；SQL常用算子SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用示例：JOIN(將訂單表數(shù)據(jù)和商品表進(jìn)行關(guān)聯(lián))SELECT*FROM

Orders

INNER

JOIN

Product

ON

Orders.productId=Product.idLEFT

JOIN

與JOIN

的區(qū)別是當(dāng)右表沒有與左邊相JOIN

的數(shù)據(jù)時候，右邊對應(yīng)的字段補(bǔ)NULL

輸出，RIGHT

JOIN

相當(dāng)于LEFT

JOIN

左右兩個表交互一下位置。FULL

JOIN

相當(dāng)于RIGHT

JOIN

和LEFTJOIN之后進(jìn)行UNION

ALL

操作。示例：SELECT*FROM

Orders

LEFT

JOIN

Product

ON

Orders.productId=Product.idSELECT*FROM

Orders

RIGHT

JOIN

Product

ON

Orders.productId=Product.idSELECT*FROM

Orders

FULL

OUTER

JOIN

Product

ON

Orders.productId=Product.id6)

JOIN

：JOIN

用于把來自兩個表的數(shù)據(jù)聯(lián)合起來形成結(jié)果表。SQL常用算子SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用?SQL語句可以單行或多行書寫，以分號結(jié)尾。?SQL語句不區(qū)分大小寫，建議關(guān)鍵字使用大寫。?可使用空格和縮進(jìn)增加語句的可讀性。?注釋。單行注釋：--注釋內(nèi)容或#注釋內(nèi)容(mysql特有)；多行注釋：/*注釋*/。SQL通用語法SQL基本概念工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用外部數(shù)據(jù)源?外部輸入的業(yè)務(wù)數(shù)據(jù)表、輸出到外部數(shù)據(jù)庫的計算結(jié)果?是用戶自定義的數(shù)據(jù)庫類型，平臺支持對外

部數(shù)據(jù)源的增加、刪除、修改、查看操作?從數(shù)據(jù)平臺獲取數(shù)據(jù)連接?內(nèi)部數(shù)據(jù)源列表顯示數(shù)據(jù)計算服務(wù)支持的數(shù)

據(jù)庫信息，包括數(shù)據(jù)源名稱、數(shù)據(jù)庫類型、版本和IP地址數(shù)據(jù)源管理：包括內(nèi)部數(shù)據(jù)源、外部數(shù)據(jù)源內(nèi)部數(shù)據(jù)源數(shù)據(jù)源管理工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用任務(wù)流中使用的外部數(shù)據(jù)表數(shù)據(jù)，需要先在數(shù)據(jù)源管理中添加外部數(shù)據(jù)源添加外部數(shù)據(jù)源①選擇外部數(shù)據(jù)源頁簽

，單擊添加按鈕數(shù)據(jù)源管理工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用數(shù)據(jù)源管理②選擇需要添加的數(shù)據(jù)庫類型可以選擇MySQL、PostgreSQL、SQLSever、Oracle關(guān)系型數(shù)據(jù)庫。

也可以選擇Kadka消息中間件。工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用數(shù)據(jù)源管理③填寫參數(shù)，完成添加數(shù)據(jù)源。以添加MySQL數(shù)據(jù)庫為例，根據(jù)已有數(shù)據(jù)庫的信息輸入數(shù)據(jù)源名稱、數(shù)據(jù)庫版本、主機(jī)、端口等信息后進(jìn)行測試連通性，測試成功后點(diǎn)擊“添加”，完成數(shù)據(jù)源的添加。工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用數(shù)據(jù)源名稱自定義數(shù)據(jù)源名稱。支持連接的數(shù)據(jù)庫版

本MySQL：5.7.x,8.0.xPostreSQL：12.xOracle：9.x.x,10.x.x,11.x.x,12.x.xKafka：1.1.x,2.0.x,2.1.x,2.2.x,2.3.x,2.4.x,2.5.x使用主機(jī)地址主機(jī)名或IP+

端口號。數(shù)據(jù)庫名/SID輸入源數(shù)據(jù)庫名。使用JDBC格式j(luò)dbc:mysql://host:port/database?useUnicode=yes&characterEncoding=utf8&useSSL=false&serverTimezone=UTC用戶名輸入用戶名。密碼輸入密碼。說明：數(shù)據(jù)庫版本支持下拉列表選擇。

輸入數(shù)據(jù)庫名或SID時，確保與

源數(shù)據(jù)庫名一致；使用JDBC時，請按照對應(yīng)的JDBC格式連接數(shù)

據(jù)庫，否則可能連接失敗。創(chuàng)建數(shù)據(jù)源參數(shù)說明數(shù)據(jù)源管理工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用參數(shù)說明Kafka版本支持的Kafka版本有：1.1.x、2.0.x、2.1.x、2.2.x、2.3.x、2.4.x、2.5.x集群地址輸入集群地址IP及端口號，如需添加多個集群地址，使用“,”區(qū)分。已經(jīng)添加的外部數(shù)據(jù)源，可以通過列表中的設(shè)置，進(jìn)入數(shù)據(jù)源詳情頁面，修改數(shù)據(jù)源的基本信息。也可以通過列表右側(cè)的圖標(biāo)移除不需要的數(shù)據(jù)源。編輯和刪除數(shù)據(jù)源添加消息中間件數(shù)據(jù)源管理自定義數(shù)據(jù)源名稱數(shù)據(jù)源名稱工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用實(shí)時數(shù)據(jù)開發(fā)過程?實(shí)時數(shù)據(jù)開發(fā)支持ApacheFlink流處理執(zhí)行模式?將設(shè)備接入平臺并上報數(shù)據(jù)后，對采集的數(shù)據(jù)進(jìn)行計算，或直接透傳輸出到指定數(shù)據(jù)庫中，是實(shí)現(xiàn)設(shè)備運(yùn)行情況分析、指標(biāo)預(yù)警、故障檢測、數(shù)據(jù)大屏展示等功能的重要前提?同時，實(shí)時數(shù)據(jù)開發(fā)提供秒級返回業(yè)務(wù)指標(biāo)和快速進(jìn)行多維分析等實(shí)時數(shù)據(jù)化能力實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用實(shí)施數(shù)據(jù)開發(fā)名稱說明計算任務(wù)計算任務(wù)是數(shù)據(jù)計算服務(wù)的基本計算單元，數(shù)據(jù)獲取、數(shù)據(jù)處理和計算結(jié)果輸出都是通過任務(wù)完成的。任務(wù)的執(zhí)行過程是個DAG

有向無環(huán)圖，圖中的點(diǎn)是執(zhí)行階

段，各個執(zhí)行階段的依賴關(guān)系是圖的邊。數(shù)據(jù)計算會依照圖中的節(jié)點(diǎn)和邊依賴關(guān)系執(zhí)行各個階段。任務(wù)流大數(shù)據(jù)的計算、分析和處理，一般由多個任務(wù)單元組成(Hive

、Sparksql

、

Spark

、Shell等)，每個任務(wù)單元完成特定的數(shù)據(jù)處理邏輯。任務(wù)流預(yù)估出每個任

務(wù)處理所需時間，根據(jù)先后順序，計算出每個任務(wù)的執(zhí)行的起止時間，通過定時執(zhí)行任務(wù)的方式，讓整個系統(tǒng)保持穩(wěn)定的運(yùn)行。任務(wù)流實(shí)例每一個離線任務(wù)根據(jù)調(diào)度周期，在每一次運(yùn)行時生成一個任務(wù)流實(shí)例。任務(wù)實(shí)例任務(wù)流由多個任務(wù)組成，一個任務(wù)流實(shí)例執(zhí)行的過程中，每個組成該任務(wù)流的任務(wù)均生成一個任務(wù)實(shí)例，所有任務(wù)實(shí)例執(zhí)行完畢，即任務(wù)流執(zhí)行完成。任務(wù)實(shí)例狀態(tài)包括：啟動中、運(yùn)行中、運(yùn)行失敗、停止中、已停止。在建立實(shí)時數(shù)據(jù)開發(fā)過程中，通過拖拽任務(wù)節(jié)點(diǎn)，建立任務(wù)節(jié)點(diǎn)的關(guān)聯(lián)，通過可視化

DAG

形成的數(shù)據(jù)處理任務(wù)定義。

重要概念的說明表實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用實(shí)施數(shù)據(jù)開發(fā)名稱說明任務(wù)模板數(shù)據(jù)計算服務(wù)把使用頻率高的計算任務(wù)當(dāng)模板存儲，減少你創(chuàng)建任務(wù)和調(diào)試的時間。數(shù)據(jù)源計算任務(wù)獲取數(shù)據(jù)的源地址和計算結(jié)果輸出的目標(biāo)地址。數(shù)據(jù)平臺支持內(nèi)部數(shù)據(jù)源，非數(shù)據(jù)平臺手動添加的數(shù)據(jù)源稱為外部數(shù)據(jù)源。內(nèi)部數(shù)據(jù)源是數(shù)據(jù)平臺內(nèi)部獲取數(shù)據(jù)連接。外部數(shù)據(jù)源是用于交付項(xiàng)目中指定的外部數(shù)據(jù)源，外部數(shù)據(jù)源可能是外部輸入的業(yè)務(wù)數(shù)據(jù)表，也可能是計算結(jié)果輸出到外部數(shù)據(jù)庫中。輸入節(jié)點(diǎn)計算任務(wù)數(shù)據(jù)輸入的源頭。支持?jǐn)?shù)據(jù)平臺的Kafka

訂閱數(shù)據(jù)，按物模型和設(shè)備列表選擇。處理節(jié)點(diǎn)計算任務(wù)過程數(shù)據(jù)過濾和時間聚合計算相關(guān)的處理。支持按條件篩選和Flink

SQL查詢語句。輸出節(jié)點(diǎn)計算任務(wù)結(jié)果數(shù)據(jù)的輸出目的地，支持MySQL和PostgreSQL

等。

重要概念的說明續(xù)表實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用實(shí)時數(shù)據(jù)開發(fā)界面可以分三塊區(qū)域：左邊為任務(wù)流列表

，中間是節(jié)點(diǎn)庫

，右邊是節(jié)點(diǎn)編排界面及數(shù)據(jù)流程，單擊具體某個節(jié)點(diǎn)會展示當(dāng)前選中節(jié)點(diǎn)屬性編輯區(qū)域。數(shù)據(jù)流啟動要求實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用①有且僅有一個輸入節(jié)點(diǎn)；②每個節(jié)點(diǎn)有且僅有一個前節(jié)點(diǎn)③有至少一個輸出節(jié)點(diǎn)；④平臺支持對實(shí)時數(shù)據(jù)開發(fā)任務(wù)流的創(chuàng)建、保存、導(dǎo)入、導(dǎo)出、刪除、編輯、查詢、撤回和復(fù)制等操作。數(shù)據(jù)開發(fā)的數(shù)據(jù)流需要滿足以下要求才能正常啟動：數(shù)據(jù)流啟動要求實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用文件夾?文件夾的層級最多為5級?支持文件夾的新增、修改(編輯文件夾名稱)、刪除?支持文件夾下的任務(wù)流的新增、修改、刪除?文件夾目錄方便管理創(chuàng)建的多個任務(wù)流，例如：同時撤回和刪除多個任務(wù)流任務(wù)流的操作實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用③單擊確定完成創(chuàng)建②新建空白畫布，在彈出框中填入自定義的名稱和選擇上級文件夾①單擊按鈕創(chuàng)建任務(wù)流。實(shí)時數(shù)據(jù)開發(fā)創(chuàng)建任務(wù)流工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用①處是節(jié)點(diǎn)庫

②是進(jìn)行任務(wù)流保存發(fā)布等操

作的區(qū)域

③是畫布。可以將節(jié)點(diǎn)庫中的節(jié)點(diǎn)拖入到畫布進(jìn)行編輯，

在編輯完成后可以在②處選擇保

存、發(fā)布導(dǎo)入導(dǎo)出等操作。④在畫布中以拖拽的方式進(jìn)行實(shí)時數(shù)據(jù)開發(fā)。實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用導(dǎo)入導(dǎo)出任務(wù)流導(dǎo)入導(dǎo)出功能方便在不同的環(huán)境和組織之間復(fù)用相同功能的任務(wù)流。已發(fā)布的任務(wù)支持導(dǎo)出，不支持導(dǎo)入。查看任務(wù)流查詢已創(chuàng)建的實(shí)時數(shù)據(jù)開發(fā)列表，可以按發(fā)布狀態(tài)進(jìn)行檢索，也可通過任務(wù)名稱進(jìn)行模糊搜索。編輯任務(wù)流可以修改任務(wù)名稱及描述。編輯操作只能對未發(fā)布的任務(wù)操作，已發(fā)布的任務(wù)無法編輯復(fù)制任務(wù)流可以對任務(wù)流進(jìn)行復(fù)制操作，復(fù)制的任務(wù)名稱為原任務(wù)名稱后添加副本及隨機(jī)字符作為后綴。已發(fā)布的任務(wù)如需修改，建議先復(fù)制再修改。任務(wù)流的操作實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用刪除任務(wù)流對于未發(fā)布的任務(wù)，可進(jìn)行刪除操作，彈出彈窗需確定刪除，刪除后無法恢復(fù)。刪除操作只能對未發(fā)布的任務(wù)操作，已發(fā)布的任務(wù)無法刪除。任務(wù)流的操作實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用當(dāng)任務(wù)流在發(fā)布狀態(tài)時可以進(jìn)行創(chuàng)建任務(wù)流、導(dǎo)出和復(fù)制操作，但不能進(jìn)行編輯名稱和刪除操作。實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用輸入節(jié)點(diǎn)處理節(jié)點(diǎn)輸出節(jié)點(diǎn)實(shí)時數(shù)據(jù)物實(shí)例篩選MySQL輸出多組織實(shí)時數(shù)據(jù)數(shù)據(jù)篩選PostgreSQL輸出維表關(guān)聯(lián)Oracle輸出Flink

SQL查詢Kafka輸出說明：輸入節(jié)點(diǎn)最多只能有1個，葉子節(jié)點(diǎn)必須是輸出節(jié)點(diǎn)。節(jié)點(diǎn)類型實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用實(shí)時數(shù)據(jù)開發(fā)過程中的基本的節(jié)點(diǎn)操作如表所示。操作名稱具體操作創(chuàng)建節(jié)點(diǎn)選擇節(jié)點(diǎn)庫中的一個節(jié)點(diǎn)類型拖拽到畫布中。復(fù)制節(jié)點(diǎn)鼠標(biāo)懸浮于需要復(fù)制的節(jié)點(diǎn)，點(diǎn)擊右上角出現(xiàn)的第一個按鈕。刪除節(jié)點(diǎn)/連線鼠標(biāo)懸浮于需要刪除的節(jié)點(diǎn)，點(diǎn)擊右上角出現(xiàn)的第二個按鈕，或者選中需要刪除的節(jié)點(diǎn)或是連線，按一下鍵盤中的Delete鍵。連接節(jié)點(diǎn)點(diǎn)擊一個節(jié)點(diǎn)下方的連接點(diǎn)拖動連線至另一個節(jié)點(diǎn)。移動畫布在畫布空白處按下鼠標(biāo)并移動即可拖動畫布，使用鼠標(biāo)滾輪可縮放畫布。節(jié)點(diǎn)基本操作實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用②多組織實(shí)時數(shù)據(jù)多組織實(shí)時數(shù)據(jù)節(jié)點(diǎn)已經(jīng)過濾了組織和模型數(shù)據(jù)，因此，多組織實(shí)時數(shù)據(jù)節(jié)點(diǎn)后建議使用除物實(shí)例篩選節(jié)點(diǎn)以外的處理節(jié)點(diǎn)輸入節(jié)點(diǎn)該節(jié)點(diǎn)定義了需要輸入和處理什么數(shù)據(jù)。每個數(shù)據(jù)處理流程只能有一個輸入節(jié)點(diǎn)。①實(shí)時數(shù)據(jù)實(shí)時數(shù)據(jù)來源為數(shù)據(jù)平臺kafka的實(shí)時數(shù)據(jù)，實(shí)時數(shù)據(jù)節(jié)點(diǎn)只能連接物實(shí)例篩選節(jié)點(diǎn)常用節(jié)點(diǎn)說明實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)需設(shè)置節(jié)點(diǎn)名稱、拉取數(shù)據(jù)位置、數(shù)據(jù)源類型實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用處理節(jié)點(diǎn)①物實(shí)例篩選：?物實(shí)例篩選節(jié)點(diǎn)只能連接實(shí)時數(shù)據(jù)輸入節(jié)點(diǎn)?節(jié)點(diǎn)名稱：支持字符串?dāng)?shù)據(jù)，最多可輸入64個字符?物實(shí)例：接入與建模服務(wù)中已存在的且需要使用任務(wù)流完成數(shù)據(jù)計算的所有物理設(shè)備，包括設(shè)備、復(fù)合物、網(wǎng)關(guān)、物應(yīng)用接口等。當(dāng)物實(shí)例選擇多個選項(xiàng)時，輸出映射取字段的交集即schema取交集實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用處理節(jié)點(diǎn)①物實(shí)例篩選：?輸出映射(模型公共屬性)接入與建模服

務(wù)中設(shè)備已經(jīng)定義的所有屬性均可映射至

數(shù)據(jù)計算服務(wù)中，根據(jù)任務(wù)流需求選擇需要參加數(shù)據(jù)計算的物模型公共屬性實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用處理節(jié)點(diǎn)①物實(shí)例篩選：?輸出映射(模型非公共屬性)：接入與建模服務(wù)中設(shè)備已經(jīng)定義的所有屬性均可映射至數(shù)據(jù)計算服務(wù)中，根據(jù)任務(wù)流需求選擇需要參加數(shù)據(jù)計算的物模型非公共屬性。實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用支持的數(shù)據(jù)類型支持的篩選條件BOOLEAN(布爾型true/false)=、!=、null、非null、在列表中、不在列表中INTEGER(64位整型)=、!=、<=、>=、<、>、在列表中、不在列表中、null、非null、在范圍中、不在范圍中NUMBER(64位浮點(diǎn)型)=、!=、<=、>=、<、>、在列表中、不在列表中、null、非null、在范圍中、不在范圍中STRING(64位字符型)=、!=、在列表中、不在列表中、null、非null、包含、不包含、以開始、不以開始、以結(jié)束、

不以結(jié)束、空、非空TIMESTAMP(64位時間戳)=、!=、<=、>=、<、>、null、非null、在范圍中、不在范圍中數(shù)據(jù)篩選選擇輸入節(jié)點(diǎn)中的某一輸出字段，并按照指定條件進(jìn)行篩選，只有滿足條件的數(shù)據(jù)才能繼續(xù)處理實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用?SQL編輯框格式化：在SQL編輯框中編寫完成的SQL語句，可通過格式化將原有語句的格式規(guī)范化，便于排查錯誤。?SQL語句校驗(yàn)：校驗(yàn)語法，可驗(yàn)證已輸入的SQL語句是否符合Flink

SQL語法，

如不符合頁面上方會出現(xiàn)警告。?窗口最大化：編輯欄窗口可放大縮小，便于查看編輯Flink

SQL查詢對流入的數(shù)據(jù)，按照編寫的語句進(jìn)行處理，計算結(jié)果流入下一個節(jié)點(diǎn)實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用?數(shù)據(jù)延遲時間：如果沒有時間窗口的設(shè)置，實(shí)時數(shù)據(jù)產(chǎn)生馬上就寫入mysql，mysql無法抗住寫入的壓力可能會出現(xiàn)宕機(jī)。為解決此類問題，可設(shè)置數(shù)據(jù)延遲時間，解決數(shù)據(jù)丟失的問題。?輸出字段：手動指定數(shù)據(jù)類型實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用?支持的時間窗口類型數(shù)據(jù)統(tǒng)計邏輯事件時間EventTime和觸發(fā)計算的處理時間ProcessingTime?時間窗口大小當(dāng)前時間窗口建議使用分鐘級，大時間窗口建議使用批處理注意：

Number

數(shù)據(jù)類型需要指定小數(shù)后幾位。確保

Flink

SQL

運(yùn)算準(zhǔn)確。避免兩個字

段相乘導(dǎo)致的小數(shù)位溢出

，

Flink

SQL

輸出就為

null

會影響數(shù)據(jù)準(zhǔn)確性實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用描述字段ID數(shù)據(jù)類型時間戳timestampTIMESTAMP設(shè)備IDdeviceIdSTRING物模型deviceTypeIdSTRING物標(biāo)識assetIdSTRING上云時間cloud_timeTIMESTAMP租戶IDtenantIdSTRING工作狀態(tài)workstatusINTEGER在線狀態(tài)onlineJson當(dāng)前位置locationJson說明：

Flink

SQL查詢節(jié)點(diǎn)，同步字段功能同步時，忽略Flink

SQL別名輸出數(shù)據(jù)類型默認(rèn)為

DECIMAL

類型，可根據(jù)需要修改為

STRING

和

INTEGER

等類型在實(shí)時數(shù)據(jù)開發(fā)中，常需要獲取以下表中系統(tǒng)變量進(jìn)行FlinkSQL查詢實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用參數(shù)說明節(jié)點(diǎn)名稱自定義節(jié)點(diǎn)的名稱，例如：每天用電量。目標(biāo)數(shù)據(jù)源選擇在數(shù)據(jù)源管理中添加的“外部數(shù)據(jù)源”作為需要存入輸出數(shù)據(jù)的數(shù)據(jù)庫。目標(biāo)數(shù)據(jù)表選擇外部數(shù)據(jù)源中需要存放輸出數(shù)據(jù)的數(shù)據(jù)庫表。插入方式-插入：直接寫入數(shù)據(jù)，若寫入失敗，則丟棄數(shù)據(jù)。-更新插入：先查詢是否有該字段，沒有就插入，有就按指定的方式對字段的值進(jìn)行更新。插入鍵默認(rèn)為主鍵PRIMARY。輸出映射將輸出字段和數(shù)據(jù)庫字段進(jìn)行匹配。選擇“更新插入”還需要設(shè)置更新數(shù)據(jù)的方式，當(dāng)前支持的更新策略如下：替換：將原數(shù)據(jù)替換成經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)后輸出。累加：將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行累加后輸出。最大值：將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行對比，取較大值作為輸出數(shù)據(jù)。最小值：將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行對比，取較小值作為輸出數(shù)據(jù)。各輸出字段類型支持的更新策略如下：Boolean支持無、替換。Integer、Number支持全部策略。String支持無、替換。Timestamp支持無、替換。按時間聚合輸出默認(rèn)勾選，勾選后，將按照下列參數(shù)設(shè)置的時間規(guī)則輸出數(shù)據(jù)。若不勾選，則直接輸出計算數(shù)據(jù)。時區(qū)不同經(jīng)緯度地區(qū)的地方時間。選擇時區(qū)后，輸出到數(shù)據(jù)庫中的時間為選擇的時區(qū)時間。時間聚合字段選擇用戶自己的庫表中，與時間對應(yīng)的字段，例如：time。聚合時長每半小時、時、日、周、月和年輸出一次數(shù)據(jù)。開始時間業(yè)務(wù)核算的時間。例如聚合時長為1分鐘，開始時間為1月1日0時0分，則將在今年1月1日0時1分輸出第一條數(shù)據(jù)。輸出節(jié)點(diǎn)

：該節(jié)點(diǎn)定義了數(shù)據(jù)最后輸出的字段、輸出到什么數(shù)據(jù)庫中、是否以時間聚合進(jìn)行輸出等實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用數(shù)據(jù)計算服務(wù)TIMESTAMPBOOLEANINTEGERINTSTRINGNUMBERMySQLDATATIMEDATETIMETIMESTAMPTINYINTTINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALTINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALCHARVARCHARTINYTEXTTEXTMEDIUMTEXTLONGTEXTDATETIMEDATETIMETINYINTSMALLINTMEDIUMINTINTBIGINTFLOATDOUBLEDECIMALMySQL輸出節(jié)點(diǎn)?目標(biāo)數(shù)據(jù)源：指定數(shù)據(jù)庫寫入經(jīng)過數(shù)據(jù)處理后得到的結(jié)果，支持內(nèi)部數(shù)據(jù)源和外部數(shù)據(jù)源?插入方式：插入表示直接追加數(shù)據(jù)庫記錄；更新插入表示基于數(shù)據(jù)庫的唯一鍵進(jìn)行更新?輸出映射：選擇需要寫入目標(biāo)數(shù)據(jù)源的字段，根據(jù)數(shù)據(jù)類型的一致性做了篩選的過濾實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用更新策略說明累加將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行累加后輸出替換將原數(shù)據(jù)替換成經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)后輸出最大值將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行對比，取較大值作為輸出數(shù)據(jù)最小值將經(jīng)過數(shù)據(jù)開發(fā)后得到的結(jié)果數(shù)據(jù)與原數(shù)據(jù)進(jìn)行對比，取較小值作為輸出數(shù)據(jù)?時間聚合輸出：MySQL寫入時會按照所選的時間聚合字段和聚合時長按照數(shù)據(jù)更新的策略進(jìn)行更新數(shù)據(jù)?更新策略：數(shù)值類型的更新策略有累加、替換、最大值、最小值；字符串類型的數(shù)據(jù)有替換策略實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用DAG圖計算校驗(yàn)內(nèi)容：?流計算每個節(jié)點(diǎn)只有唯一父節(jié)點(diǎn)?批計算Flink

SQL節(jié)點(diǎn)可以有多個父節(jié)點(diǎn)?有向不循環(huán)DAG全稱DirectedAcyclic

Graph，有向無環(huán)圖DAG圖編輯工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用?流計算通過深度優(yōu)先搜索(DFS)判斷是否有節(jié)點(diǎn)在圖中遍歷兩次，以判斷有向圖中是否存在閉環(huán)通過深度優(yōu)先搜索(DFS)判斷是否有節(jié)點(diǎn)在圖中從未被遍歷，以判斷有向圖中是否存孤立節(jié)點(diǎn)?批計算通過Topological

ordering是否存在判斷有向圖中是否存在閉環(huán)通過深度優(yōu)先搜索(DFS)判斷是否有節(jié)點(diǎn)在圖中從未被遍歷，以判斷無向圖中是否存孤立節(jié)點(diǎn)DAG圖編輯工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用任務(wù)流MySQL節(jié)點(diǎn)校驗(yàn)規(guī)則表節(jié)點(diǎn)節(jié)點(diǎn)枚舉前端校驗(yàn)發(fā)布校驗(yàn)流任務(wù)MySQL輸出節(jié)點(diǎn)MYSQL_OUTPUT_NODE-檢查是否選擇了相同的映射字段-檢查是否將所有非空字段映射-檢查是否選擇了插入鍵-檢查是否選擇了數(shù)據(jù)源-檢查是否選擇了表-檢查是否有勾選了輸出但是沒選映射字段的-檢查是否輸出了所有非自增的插入鍵字段(插入更新的情

況下)-檢查是否在開啟時間聚合的情況下選擇了時間聚合字段、

檢查時間聚合字段是否在輸出字段中被映射了insert：非空字段是否全部映射(包

含了主鍵)upsert:-除自增insertkey外是否全部

映射-insertkey不能為空切不能全

為自增-非空主鍵是否全部映射DAG圖編輯工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用約束條件校驗(yàn)完整的DAG圖需滿足必須有且只有一個輸入節(jié)點(diǎn)，輸入節(jié)點(diǎn)是Kafka或多組織實(shí)時數(shù)據(jù)節(jié)點(diǎn)，所有葉子節(jié)點(diǎn)必須是輸出節(jié)點(diǎn)校驗(yàn)規(guī)則①DAG圖校驗(yàn)；②節(jié)點(diǎn)配置校驗(yàn)最大深度圖從輸入節(jié)點(diǎn)到輸出節(jié)點(diǎn)的最大深度為10

(MAX_DEPTH),每個節(jié)點(diǎn)的最大子節(jié)點(diǎn)個數(shù)為MAX_WIDTH,全部節(jié)點(diǎn)個數(shù)不大于50(MAX_TOTAL)，用于限制計算復(fù)雜度。DAG圖編輯工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用以時間驅(qū)動，根據(jù)時間對數(shù)據(jù)流進(jìn)行分組例如：每30秒鐘

滾動時間窗口(Tumbling

Time

Window，無重疊) 滑動時間窗口(Sliding

Time

Window，有重疊)

會話窗口(Session

Window，以會話間隔驅(qū)動)以事件數(shù)量驅(qū)動例如：每一百個事件

滾動計數(shù)窗口(TumblingCount

Window，無重疊)

滑動計數(shù)窗口(SlidingCountWindow，有重疊)窗口提供了這樣一種依據(jù)，決定了數(shù)據(jù)何時開始處理,處理哪些數(shù)據(jù)Count

WindowTime

Window相關(guān)概念Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用

采用會話持續(xù)時長作為窗口處理依據(jù)

會話窗口根據(jù)會話間隔切分不同的窗口

窗口的長度固定，以一個步長(Slide)不斷向前滑動

步長設(shè)置的大小決定了Flink以多大的頻率來創(chuàng)建新的窗口

滾動窗口的窗口長度是固定的，窗口之間不重疊。滑動窗口(Sliding

Window)滾動窗口(TumGbling

indow)會話窗口(Session

Window)Flink有3個內(nèi)置Time

Window類型Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用

優(yōu)勢：Flink的時間系統(tǒng)中最簡單的概念，不需要流和機(jī)器之間的協(xié)調(diào)。它提供了最佳的性能和最低的延遲。

劣勢：前面發(fā)生數(shù)據(jù)積壓，時間窗口內(nèi)數(shù)據(jù)量暴增，占用內(nèi)存高

優(yōu)勢：一般由數(shù)據(jù)生產(chǎn)方自身攜帶，從消息的產(chǎn)生就誕生了，不會改變。

劣勢：設(shè)備離線后，沒有數(shù)據(jù)上傳，窗口不結(jié)算；設(shè)備上線后，窗口恢復(fù)計算，離線前統(tǒng)計數(shù)據(jù)延遲輸出。

優(yōu)勢：可以防止Flink內(nèi)部處理數(shù)據(jù)是發(fā)生亂序的情況。

劣勢：數(shù)據(jù)進(jìn)入Flink前出現(xiàn)數(shù)據(jù)積壓或者斷線緩沖數(shù)據(jù)導(dǎo)致數(shù)據(jù)遲到，數(shù)據(jù)統(tǒng)計到下個周期中。設(shè)備離線同理。事件時間(Event

Time)處理時間(Processing

Time)進(jìn)入(攝取)時間

(Ingestion

Time)三種時間窗口的區(qū)別Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用窗口函數(shù)描述舉例TUMBLE(time_attr,interval)翻滾時間窗口，將行分配給具有固定持續(xù)時間(interval)的非重疊、連續(xù)窗口。一個5分鐘的滾動窗口即每五分鐘分組一次，這五分鐘內(nèi)的數(shù)據(jù)為一組HOP(time_attr,interval,

interval)跳躍時間窗口，具有固定的持續(xù)時間(第二個interval參數(shù))，并按指定的跳躍間隔(第一個interval參數(shù))跳躍。一個5分鐘的跳躍窗口，其中跳躍間隔為3分鐘，即每五分鐘分組一次，這五分鐘內(nèi)的數(shù)據(jù)為一組，等待三分鐘再進(jìn)行下一次分組SESSION(time_attr,interval)會話時間窗口沒有固定的持續(xù)時間，但它們的界限由interval不活動的時間定義，即，如果在定義的間隙期間沒有事件出現(xiàn)，則會話窗口關(guān)閉。設(shè)置interval為5分鐘，若窗口在上一次發(fā)生事件至今超過5分鐘，則該窗口關(guān)閉FlinkSQL-窗口函數(shù)Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用滾動窗口滑動窗口會話窗口SELECT

count(event),TUMBLE_START(time_attr,

INTERVAL‘30’SECOND)AS

ts_start,FROM

tumble_streamGROUP

BY

TUMBLE(time_attr,INTERVAL‘30’SECOND)SELECT

count(event),HOP_START(time_attr,INTERVAL‘30’SECOND,INTERVAL‘1’MINUTE)

ASts_start,FROM

slide_streamGROUP

BY

HOP(time_attr,INTERVAL‘30’SECOND,INTERVAL‘1’MINUTE)SELECT

count(event),SESSION_START(time_attr,INTERVAL‘30’SECOND)AS

ts_start,FROM

session_streamGROUP

BY

SESSION(time_attr,

INTERVAL‘30’SECOND)窗口函數(shù)示例Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用窗口使用的輔助函數(shù)輔助函數(shù)描述TUMBLE_START(time_attr,interval)HOP_START(time_attr,interval,interval)SESSION_START(time_attr,interval)返回相應(yīng)翻滾、跳躍或會話窗口的包含下限的時間戳。TUMBLE_END(time_attr,interval)HOP_END(time_attr,interval,interval)SESSION_END(time_attr,interval)返回對應(yīng)的翻滾、跳躍或會話窗口的獨(dú)占上限的時間戳。注意：在后續(xù)的基于時間的操作中，例如區(qū)間連接和組窗口或跨窗口聚合中，不能將獨(dú)占上限時間戳用作行時間屬性。TUMBLE_ROWTIME(time_attr,interval)HOP_ROWTIME(time_attr,interval,interval)SESSION_ROWTIME(time_attr,interval)返回相應(yīng)翻滾、跳躍或會話窗口的包含上限的時間戳。

結(jié)果屬性是一個行時間屬性，可用于后續(xù)基于時間的操作，例如間隔連接和組窗口或窗口聚合。TUMBLE_PROCTIME(time_attr,interval)HOP_PROCTIME(time_attr,interval,interval)SESSION_PROCTIME(time_attr,interval)返回一個proctime屬性，該屬性可用于后續(xù)基于時間的操作，例如間隔連接和組窗口或窗口聚合。TUMBLE_START(time_attr,interval)HOP_START(time_attr,interval,interval)SESSION_START(time_attr,interval)返回相應(yīng)翻滾、跳躍或會話窗口的包含下限的時間戳。Flink窗口工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用《工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用》/知識準(zhǔn)備任務(wù)實(shí)施目錄0102數(shù)據(jù)源分為外部數(shù)據(jù)源和內(nèi)部數(shù)據(jù)源，外部數(shù)據(jù)源支持MySQL

，

PostgreSQL、Oracle。

可以根據(jù)實(shí)踐應(yīng)用場景進(jìn)行添加。本次任務(wù)采用外部數(shù)據(jù)源，建立外部數(shù)據(jù)源的步驟為：1)登錄平臺，進(jìn)入控制臺。2)數(shù)據(jù)計算-

>數(shù)據(jù)源管理。3)選擇外部數(shù)據(jù)源，單擊“添加”。連接外部數(shù)據(jù)源工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用在外部數(shù)據(jù)源可以選擇MySQL、PstagreSQL、SQLserver、Oracle關(guān)系型數(shù)據(jù)庫，還可以添

加消息中間件Kadka。如圖所示，本次任務(wù)選擇關(guān)系型MySQL數(shù)據(jù)庫。連接外部數(shù)據(jù)源工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用在選擇數(shù)據(jù)庫類型后，根據(jù)已有數(shù)據(jù)庫的信息輸入數(shù)據(jù)源名稱、數(shù)據(jù)庫版本、主機(jī)、端口等信息后進(jìn)行測試連通性，測試成功后點(diǎn)擊“添加”，完成數(shù)據(jù)源的添加。數(shù)據(jù)源管理工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用(1)添加文件夾1)創(chuàng)建一級文件夾，操作步驟：登錄控制臺>數(shù)據(jù)計算>數(shù)據(jù)開發(fā)>實(shí)時數(shù)據(jù)開發(fā)>點(diǎn)擊符號>添加一級文件夾，如圖所示：創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用2)自定義文件夾名稱，點(diǎn)擊“確定”，完成文件夾創(chuàng)建。創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用(2)創(chuàng)建任務(wù)流1)如圖所示，創(chuàng)建任務(wù)流。創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用2)創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)，自定義任務(wù)名稱“機(jī)器人產(chǎn)量分析”，上級文件夾選擇“課程項(xiàng)目”，點(diǎn)擊“確定”，完成實(shí)時數(shù)據(jù)開發(fā)創(chuàng)建。創(chuàng)建實(shí)時數(shù)據(jù)開發(fā)工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用1)選擇輸入節(jié)點(diǎn)“實(shí)時數(shù)據(jù)”，將節(jié)點(diǎn)拖拽至畫布中。2)編輯輸入節(jié)點(diǎn)，自定義節(jié)點(diǎn)名稱，拉取位置選擇“從上一次的位置拉取數(shù)據(jù)”，數(shù)據(jù)類型選擇“實(shí)時數(shù)據(jù)”。輸入節(jié)點(diǎn)設(shè)置工業(yè)互聯(lián)網(wǎng)平臺綜合應(yīng)用1)選擇處理節(jié)點(diǎn)“物實(shí)例篩選”，將節(jié)點(diǎn)拖拽至畫布中，然后點(diǎn)擊“實(shí)時數(shù)據(jù)”節(jié)點(diǎn)下方的連接點(diǎn)拖動連線至“