flink sql脚本

深藏阁楼爱情的钟 2023-05-31 07:00 51阅读 0赞

*  flink version:1.9.0

flink sql client不支持CREATE TABLE的DDL语句，Table api是支持的。  
可以自己写一个提交sql的jar包练习下，学习基于flink1.9如何使用flink sql。

*  主要代码

public class SqlSubmit { 
    
        /** * run flink sql * @param sqls * @throws Exception */
        public void run(List<String> sqls)throws Exception { 
    
            final EnvironmentSettings settings = EnvironmentSettings
                    .newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
            final TableEnvironment tableEnv = TableEnvironment.create(settings);
            //sql文本中只有create、insert语句
            for (String sql : sqls) { 
                tableEnv.sqlUpdate(sql);
            }
    
            tableEnv.execute("submit sql job");
    
        }
    
        public static void main(String[] args) throws Exception { 
    
            if (args == null || args.length < 1) { 
                throw new IllegalArgumentException("缺少<sql-file>");
            }
            //获取sql-file的全路径名
            String fileName = args[0];
    
            //解析内容，获取sql语句
            List<String> sqlLines = Files.readAllLines(Paths.get(fileName));
            List<String> sqls = parse(sqlLines);
            //提交sql job
            new SqlSubmit().run(sqls);
        }
    
    
        private static List<String> parse(List<String> sqlLines) { 
            List<String> sqlList = new ArrayList<>();
            StringBuilder stmt = new StringBuilder();
            for (String line : sqlLines) { 
    
                if (line.trim().isEmpty() || line.trim().startsWith("#")) { 
                    // 过滤掉空行以及注释行
                    continue;
                }
                stmt.append("\n").append(line.trim());
                if (line.trim().endsWith(";")) { 
                    sqlList.add(stmt.substring(0, stmt.length() - 1).trim());
                    //清空，复用
                    stmt.setLength(0);
                }
            }
            return sqlList;
        }
    
    }

*  test.sql脚本

# source
    CREATE TABLE start_log_source( 
       mid_id VARCHAR,  
       user_id INT,  
       version_code VARCHAR,  
       version_name VARCHAR,  
       lang VARCHAR,  
       source VARCHAR,  
       os VARCHAR,  
       area VARCHAR,  
       model VARCHAR,  
       brand VARCHAR,  
       sdk_version VARCHAR,  
       height_width VARCHAR,  
       app_time VARCHAR,  
       network VARCHAR,  
       lng FLOAT,  
       lat FLOAT  
    ) WITH ( 
       'connector.type' = 'kafka',  
       'connector.version' = '0.11',
       'connector.topic' = 'start_log',  
       'connector.startup-mode' = 'earliest-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',  
       'format.derive-schema' = 'true'  
    );
    
    # sink
    CREATE TABLE start_log_sink ( 
        mid_id VARCHAR, 
        user_id INT 
    ) WITH ( 
        'connector.type' = 'jdbc', 
        'connector.url' = 'jdbc:mysql://localhost:3306/flink_test',
        'connector.table' = 'start_log_to_mysql', 
        'connector.username' = 'root', 
        'connector.password' = '123456', 
        'connector.write.flush.max-rows' = '1' 
    );
    
    insert into start_log_sink 
    select mid_id, user_id
    from start_log_source;

*  写一个run.sh脚本

#!/usr/bin/env bash
    
    FLINK_DIR=/local/sda/flink/flink-1.9.1
    
    $FLINK_DIR/bin/flink run -d -p 4 -c com.zhdan.flink.cli.SqlSubmit target/flink-demo-1.0-SNAPSHOT.jar $1

*  提交

run.sh test.sql

*  代码在github上，后续会加上如何根据时间消费kafka数据  
    [https://github.com/iamsir/flink-demo][https_github.com_iamsir_flink-demo]

[https_github.com_iamsir_flink-demo]: https://github.com/iamsir/flink-demo