Flink SQL Client鍒濇帰

Posted 2020-12-21

1. 鏍囩锛?a href='http://www.mamicode.com/so/1/global' title='global'>global   杩炴帴   鍘熷垱鏂囩珷   color   mos   dde   ems   sel   group   

   娆㈣繋璁块棶鎴戠殑GitHub

   https://github.com/zq2599/blog_demos

   鍐呭锛氭墍鏈夊師鍒涙枃绔犲垎绫绘眹鎬诲強閰嶅婧愮爜锛屾秹鍙奐ava銆丏ocker銆並ubernetes銆丏evOPS绛夛紱

   鍏充簬Flink SQL Client

   Flink Table & SQL鐨凙PI瀹炵幇浜嗛€氳繃SQL璇█澶勭悊瀹炴椂鎶€鏈畻涓氬姟锛屼絾杩樻槸瑕佺紪鍐欓儴鍒咼ava浠ｇ爜(鎴朣cala)锛屽苟涓旇繕瑕佺紪璇戞瀯寤烘墠鑳芥彁浜ゅ埌Flink杩愯鐜锛岃繖瀵逛簬涓嶇啛鎮塉ava鎴朣cala鐨勫紑鍙戣€呭氨鐣ユ湁浜涗笉鍙嬪ソ浜嗭紱
   SQL Client鐨勭洰鏍囧氨鏄В鍐充笂杩伴棶棰橈紙瀹樻柟鍘熻瘽with a build tool before being submitted to a cluster.锛?/p>

   鐜淇℃伅

   鎺ヤ笅鏉ラ噰鐢ㄥ疄鎴樼殑鏂瑰紡瀵笷link SQL Client鍋氬垵姝ュ皾璇曪紝鐜淇℃伅濡備笅锛?/p>

   1. 鐢佃剳锛歁acBook Pro2018 13瀵革紝macOS Catalina 10.15.3
   2. Flink锛?.10.0
   3. JDK锛?.8.0_211

   鏈湴鍚姩flink

   1. 涓嬭浇flink鍖咃紝鍦板潃锛?font color="blue">http://ftp.kddilabs.jp/infosystems/apache/flink/flink-1.10.0/flink-1.10.0-bin-scala_2.11.tgz
   2. 瑙ｅ帇锛?font color="blue">tar -zxvf flink-1.10.0-bin-scala_2.11.tgz
   3. 杩涚洰褰昮link-1.10.0/bin/锛屾墽琛屽懡浠?font color="blue">./start-cluster.sh鍚姩鏈湴flink锛?/li>
   4. 璁块棶璇ユ満鍣ㄧ殑8081绔彛锛屽彲瑙佹湰鍦癴link鍚姩鎴愬姛锛?br>

   鍚姩SQL Client CLI

   1. 鍦ㄧ洰褰昮link-1.10.0/bin/鎵ц./sql-client.sh鍗冲彲鍚姩SQL Client CLI锛屽涓嬪浘鎵€绀猴紝绾㈡涓殑BETA鎻愰啋鐫€鍦ㄧ敓浜х幆澧冨鏋滆鐢ㄦ宸ュ叿锛?br>

   2. 绗竴涓鎺屾彙鐨勬槸HELP鍛戒护锛?br>

   3. 浠巋ello world寮€濮嬫妸锛屾墽琛屽懡浠?font color="blue">select 鈥楬ello world!鈥?锛屾帶鍒跺彴杈撳嚭濡備笅鍥炬墍绀猴紝杈撳叆Q鍙€€鍑猴細
      

   涓ょ灞曠ず妯″紡

   1. 绗竴绉嶆槸table mode锛屾晥鏋滃儚鏄鏅€氭暟鎹〃鐨勬煡璇紝璁剧疆璇ユā寮忕殑鍛戒护锛?/li>

   SET execution.result-mode=table;
   

   2. 绗簩绉嶆槸changelog mode锛屾晥鏋滃儚鏄墦鍗版瘡涓€娆℃暟鎹彉鏇寸殑鏃ュ織锛岃缃妯″紡鐨勫懡浠わ細

   SET execution.result-mode=changelog;
   

   3. 璁剧疆table mode鍚庯紝鎵ц浠ヤ笅鍛戒护浣滀竴娆＄畝鍗曠殑鍒嗙粍鏌ヨ锛?/li>

   SELECT name, 
     COUNT(*) AS cnt 
     FROM (VALUES (鈥楤ob鈥?, (鈥楢lice鈥?, (鈥楪reg鈥?, (鈥楤ob鈥?) 
     AS NameTable(name) 
     GROUP BY name;
   

   4. 涓轰簡渚夸簬瀵规瘮锛屼笅鍥惧悓鏃惰创涓婁袱绉嶆ā寮忕殑鏌ヨ缁撴灉锛屾敞鎰忕豢妗嗕腑鏄剧ず浜嗚琛岃褰曟槸澧炲姞杩樻槸鍒犻櫎锛?br>

   5. 涓嶈鏄摢绉嶆ā寮忥紝鏌ヨ缁撴瀯閮戒繚瀛樺湪SQL Client CLI杩涚▼鐨勫爢鍐呭瓨涓紱

   6. 鍦╟henglog妯″紡涓嬶紝涓轰簡淇濊瘉鎺у埗鍙板彲浠ユ甯歌緭鍏ヨ緭鍑猴紝鏌ヨ缁撴灉鍙睍绀烘渶杩?000鏉★紱

   7. table妯″紡涓嬶紝鍙互缈婚〉鏌ヨ鏇村缁撴灉锛岀粨鏋滄暟閲忓彈閰嶇疆椤?font color="blue">max-table-result-rows浠ュ強鍙敤鍫嗗唴瀛橀檺鍒讹紱

鎿嶄綔

1. 棣栧厛璇风‘淇滷link宸茬粡鍚姩锛?/li>
2. 鍒涘缓鍚嶄负book-store.csv鐨勬枃浠讹紝鍐呭濡備笅锛?/li>

name001,1,aaa
name002,2,aaa
name003,3,bbb
name004,4,bbb
name005,5,bbb
name006,6,ccc
name007,7,ccc
name008,8,ccc
name009,9,ccc
name010,10,ccc

3. 鍦?font color="blue">flink-1.10.0/conf鐩綍涓嬪垱寤哄悕涓?font color="blue">book-store.yaml鐨勬枃浠讹紝鍐呭濡備笅锛?/li>

tables:
  - name: BookStore
    type: source-table
    update-mode: append
    connector:
      type: filesystem
      path: "/Users/zhaoqin/temp/202004/26/book-store.csv"
    format:
      type: csv
      fields:
        - name: BookName
          type: VARCHAR
        - name: BookAmount
          type: INT
        - name: BookCatalog
          type: VARCHAR
      line-delimiter: "
"
      comment-prefix: ","
    schema:
      - name: BookName
        type: VARCHAR
      - name: BookAmount
        type: INT
      - name: BookCatalog
        type: VARCHAR
  - name: MyBookView
    type: view
    query: "SELECT BookCatalog, SUM(BookAmount) AS Amount FROM BookStore GROUP BY BookCatalog"


execution:
  planner: blink                    # optional: either 鈥榖link鈥?(default) or 鈥榦ld鈥?  type: streaming                   # required: execution mode either 鈥榖atch鈥?or 鈥榮treaming鈥?  result-mode: table                # required: either 鈥榯able鈥?or 鈥榗hangelog鈥?  max-table-result-rows: 1000000    # optional: maximum number of maintained rows in
                                    #   鈥榯able鈥?mode (1000000 by default, smaller 1 means unlimited)
  time-characteristic: event-time   # optional: 鈥榩rocessing-time鈥?or 鈥榚vent-time鈥?(default)
  parallelism: 1                    # optional: Flink鈥榮 parallelism (1 by default)
  periodic-watermarks-interval: 200 # optional: interval for periodic watermarks (200 ms by default)
  max-parallelism: 16               # optional: Flink鈥榮 maximum parallelism (128 by default)
  min-idle-state-retention: 0       # optional: table program鈥榮 minimum idle state time
  max-idle-state-retention: 0       # optional: table program鈥榮 maximum idle state time

                                    #   (default database of the current catalog by default)
  restart-strategy:                 # optional: restart strategy
    type: fallback                  #   "fallback" to global restart strategy by default

# Configuration options for adjusting and tuning table programs.

# A full list of options and their default values can be found
# on the dedicated "Configuration" page.
configuration:
  table.optimizer.join-reorder-enabled: true
  table.exec.spill-compression.enabled: true
  table.exec.spill-compression.block-size: 128kb

# Properties that describe the cluster to which table programs are submitted to.

deployment:
  response-timeout: 5000

4. 瀵逛簬book-store.yaml鏂囦欢锛屾湁浠ヤ笅鍑犲闇€瑕佹敞鎰忥細

a. tables.type绛変簬source-table锛岃〃鏄庤繖鏄暟鎹簮鐨勯厤缃俊鎭紱

b. tables.connector鎻忚堪浜嗚缁嗙殑鏁版嵁婧愪俊鎭紝path鏄?font color="blue">book-store.csv鏂囦欢鐨勫畬鏁磋矾寰勶紱

c. tables.format鎻忚堪浜嗘枃浠跺唴瀹癸紱

d. tables.schema鎻忚堪浜嗘暟鎹簮琛ㄧ殑琛ㄧ粨鏋勶紱

e. type涓簐iew琛ㄧずMyBookView鏄釜瑙嗗浘(鍙傝€冩暟鎹簱鐨勮鍥炬蹇?锛?/p>

5. 鍦?font color="blue">flink-1.10.0鐩綍鎵ц浠ヤ笅鍛戒护锛屽嵆鍙惎鍔⊿QL Client锛屽苟鎸囧畾book-store.yaml涓虹幆澧冮厤缃細

bin/sql-client.sh embedded -d conf/book-store.yaml

6. 鏌ュ叏琛細

SELECT * FROM BookStore;

7. 鎸夌収BookCatalog鍒嗙粍缁熻璁板綍鏁帮細

SELECT BookCatalog, COUNT(*) AS BookCount FROM BookStore GROUP BY BookCatalog;

8. 鏌ヨ瑙嗗浘锛?/li>

select * from MyBookView;

鑷虫锛孎link SQL Client鐨勫垵娆′綋楠屽氨瀹屾垚浜嗭紝鍜变滑姝ゅ伐鍏风畻鏄湁浜嗗熀鏈簡瑙ｏ紝鎺ヤ笅鏉ョ殑鏂囩珷浼氳繘涓€姝ヤ娇鐢‵link SQL Client鍋氫簺澶嶆潅鐨勬搷浣滐紱

https://github.com/zq2599/blog_demos