Flink 1.9 实战：使用 SQL 读取 Kafka 并写入 MySQL

EnvironmentSettings settings = EnvironmentSettings.newInstance()     
        .useBlinkPlanner()     
        .inStreamingMode()     
        .build();     
// 创建一个使用 Blink Planner 的 TableEnvironment, 并工作在流模式     
TableEnvironment tEnv = TableEnvironment.create(settings);     
// 读取 SQL 文件     
List<String> sql = Files.readAllLines(path);     
// 通过正则表达式匹配前缀，来区分不同的 SQL 语句     
List<SqlCommandCall> calls = SqlCommandParser.parse(sql);     
// 根据不同的 SQL 语句，调用 TableEnvironment 执行     
for (SqlCommandCall call : calls) {     
    switch (call.command) {     
        case SET:     
            String key = call.operands[0];     
            String value = call.operands[1];     
            // 设置参数     
            tEnv.getConfig().getConfiguration().setString(key, value);     
            break;     
        case CREATE_TABLE:     
            String ddl = call.operands[0];     
            tEnv.sqlUpdate(ddl);     
            break;     
        case INSERT_INTO:     
            String dml = call.operands[0];     
            tEnv.sqlUpdate(dml);     
            break;     
        default:     
            throw new RuntimeException("Unsupported command: " + call.command);     
    }     
}     
// 提交作业     
tEnv.execute("SQL Job");     

{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}     
{"user_id": "662867", "item_id":"2244074", "category_id": "1575622", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}     

CREATE TABLE user_log (     
    user_id VARCHAR,     
    item_id VARCHAR,     
    category_id VARCHAR,     
    behavior VARCHAR,     
    ts TIMESTAMP     
) WITH (     
    'connector.type' = 'kafka', -- 使用 kafka connector     
    'connector.version' = 'universal',  -- kafka 版本，universal 支持 0.11 以上的版本     
    'connector.topic' = 'user_behavior',  -- kafka topic     
    'connector.startup-mode' = 'earliest-offset', -- 从起始 offset 开始读取     
    'connector.properties.0.key' = 'zookeeper.connect',  -- 连接信息     
    'connector.properties.0.value' = 'localhost:2181',      
    'connector.properties.1.key' = 'bootstrap.servers',     
    'connector.properties.1.value' = 'localhost:9092',      
    'update-mode' = 'append',     
    'format.type' = 'json',  -- 数据源格式为 json     
    'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则     
)     

CREATE TABLE pvuv_sink (     
    dt VARCHAR,     
    pv BIGINT,     
    uv BIGINT     
) WITH (     
    'connector.type' = 'jdbc', -- 使用 jdbc connector     
    'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- jdbc url     
    'connector.table' = 'pvuv_sink', -- 表名     
    'connector.username' = 'root', -- 用户名     
    'connector.password' = '123456', -- 密码     
    'connector.write.flush.max-rows' = '1' -- 默认5000条，为了演示改为1条     
)     

INSERT INTO pvuv_sink     
SELECT     
  DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,     
  COUNT(*) AS pv,     
  COUNT(DISTINCT user_id) AS uv     
FROM user_log     
GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00')     

FLINK_DIR=/Users/wuchong/dev/install/flink-1.9.0     

$ docker pull mysql     
$ docker run --name mysqldb -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 -d mysql