前提：
1.配置flink的元数据到hive，不然每次重启flink-cli，之前建的表都丢失了
在这个默认的sql-client-defaults.yaml修改
catalogs:
- name: catalog_1
type: hive
hive-conf-dir: /opt/module/hive/conf
default-database: mydb
execution:
type: streaming
result-mode: tableau #查询出来的数据表格化，更好看
2.需要导入包：（我使用的flink1.11，自带 flink-json-1.11.3.jar,flink-sql-connector-kafka_2.11-1.11.3.jar）
flink-connector-hive_2.11-1.11.0.jar
flink-connector-jdbc_2.12-1.11.3.jar
flink-table-planner_2.11-1.11.3.jar
flink-table-planner-blink_2.11-1.11.3.jar
hive-exec-3.1.2.jar
mysql-connector-java-8.0.15.jar
有了这些包分发到flink集群，flink就可以和mysql，hive，kafka互通
一、Flink 与 Mysql
1.mysql库mydw建维度表
CREATE TABLE dim_province (
province_id BIGINT,
province_name VARCHAR,
region_name VARCHAR) ;
并插入数据：
insert into dim_province (province_id,province_name,region_name) values (1,‘广东‘,‘华南‘);
insert into dim_province (province_id,province_name,region_name) values (2,‘湖北‘,‘华中‘);
insert into dim_province (province_id,province_name,region_name) values (3,‘山东‘,‘华北‘);
2.Flink Source mysql
建维度表：维度数据存储在mysql
CREATE TABLE dim_province (
province_id BIGINT,
province_name VARCHAR,
region_name VARCHAR
) WITH (
‘connector.type‘ = ‘jdbc‘,
‘connector.url‘ = ‘jdbc:mysql://192.168.9.103:3306/mydw?useUnicode=true&characterEncoding=utf-8‘,
‘connector.table‘ = ‘dim_province‘,
‘connector.driver‘ = ‘com.mysql.jdbc.Driver‘,
‘connector.username‘ = ‘root‘,
‘connector.password‘ = ‘000000‘,
‘connector.lookup.cache.max-rows‘ = ‘5000‘,
‘connector.lookup.cache.ttl‘ = ‘10min‘
);
校验是否成功：
Flink-sql>select * from dim_province;
Mysql的数据可以在这查出来；
3.Flink sink mysql
CREATE TABLE region_sales_sink (
region_name VARCHAR(30),
buy_cnt BIGINT
) WITH (
‘connector.type‘ = ‘jdbc‘,
‘connector.url‘ = ‘jdbc:mysql://192.168.9.103:3306/mydw?useUnicode=true&characterEncoding=utf-8‘,
‘connector.table‘ = ‘top_region‘, -- MySQL中的待插入数据的表
‘connector.driver‘ = ‘com.mysql.jdbc.Driver‘,
‘connector.username‘ = ‘root‘,
‘connector.password‘ = ‘000000‘,
‘connector.write.flush.interval‘ = ‘1s‘
);
数据写到mysql过程可能会出现中文乱码，把mysql服务器配置文件设置为urf-8，
首先修改MySQL的配置文件/etc/mysql/my.cnf：
在[mysqld]下追加：
character-set-server=utf8
Mysql>status;
可以查看服务器编码状态都改为utf-8;
二、Flink 与 Kafka
建事实行为表：行为数据存储在kafka
CREATE TABLE user_behavior (
user_id BIGINT, -- 用户id
item_id BIGINT, -- 商品id
cat_id BIGINT, -- 品类id
action STRING, -- 用户行为
province INT, -- 用户所在的省份
ts BIGINT, -- 用户行为发生的时间戳
proctime as PROCTIME(), -- 通过计算列产生一个处理时间列
eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(ts, ‘yyyy-MM-dd HH:mm:ss‘)), -- 事件时间
WATERMARK FOR eventTime as eventTime - INTERVAL ‘5‘ SECOND
) WITH (
‘connector.type‘ = ‘kafka‘, -- 使用 kafka connector
‘connector.version‘ = ‘universal‘, -- kafka 版本，universal 支持 0.11 以上的版本
‘connector.topic‘ = ‘user_behavior‘, -- kafka主题
‘connector.startup-mode‘ = ‘earliest-offset‘, -- 偏移量，从起始 offset 开始读取
‘connector.properties.group.id‘ = ‘group1‘, -- 消费者组
‘connector.properties.zookeeper.connect‘ = ‘hadoop101:2181‘, -- zookeeper 地址
‘connector.properties.bootstrap.servers‘ = ‘hadoop101:9092‘, -- kafka broker 地址
‘format.type‘ = ‘json‘ -- 数据源格式为 json
);
kafka生产者生产数据，检验数据：
Flink-sql>select * from user_behavior ;
打印数据
三、Flink + Hive
配置好catalog之后，Flink-Sql建的表都会在Hive落库