1. CREATE CATALOG iceberg WITH (
    2.   'type'='iceberg',
    3.   'catalog-type'='hive',
    4.   'uri'='thrift://172.16.48.191:9083',
    5.   'clients'='5',
    6.   'property-version'='1',
    7.   'hive-conf-dir'='/data/program/flink-114/conf/iceberg'
    8. );
    10. -- =================================
    11. -- 定义Source，View
    12. -- =================================
    13. -- 生成数据
    14. CREATE TABLE clicks_datagen
    15. (
    16.   `user` VARCHAR COMMENT '用户名称',
    17.   `cTime` AS LOCALTIMESTAMP COMMENT '点击时间',
    18.   `url`  VARCHAR COMMENT '点击链接',
    19.   `procTime` AS PROCTIME() COMMENT '处理时间',
    20.   WATERMARK FOR `cTime` AS `cTime` - INTERVAL '5' SECOND
    21. ) COMMENT '用户点击表'
    22. WITH (
    23.   'connector' = 'datagen',
    24.   'rows-per-second' = '100',
    25.   'fields.user.kind' = 'random',
    26.   'fields.user.length' = '5',
    27.   'fields.url.kind' = 'random',
    28.   'fields.url.length' = '5'
    29. );
    31. -- elasticsearch 6
    32. -- =================================
    33. -- elasticsearch sink
    34. CREATE TABLE clicks_es6
    35. (
    36.   `user`  VARCHAR COMMENT '用户名称',
    37.   `cTime` TIMESTAMP COMMENT '点击时间',
    38.   `url`   VARCHAR COMMENT '点击链接'
    39. ) WITH (
    40.   'connector' = 'elasticsearch-6',
    41.   'document-type' = '_doc',
    42.   'hosts' = 'http://192.168.5.16:9200;http://192.168.5.17:9200',
    43.   'index' = 'merlin_flink_clicks-{cTime|yyyy-MM-dd}'
    44. );
    46. -- datagen --> kafka --> es6
    47. INSERT INTO iceberg.shuidi.clicks
    48. SELECT `user`,
    49. `cTime`,
    50. `url`
    51. FROM clicks_datagen;