Skip to content
基于开源的flink,对其实时sql进行扩展;主要实现了流与维表的join,支持原生flink SQL所有的语法
Java Other
  1. Java 99.9%
  2. Other 0.1%
Branch: 1.8.0_release
Clone or download
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
bin flink1.6+perjob Oct 29, 2018
cassandra cassandra plugin Dec 5, 2019
clickhouse clickhouse sink and field parse Nov 5, 2019
console [pom文件加入branch][flinkStreamSQL支持not null 语法][17872] Sep 18, 2019
core Merge branch 'hotfix_1.8_3.9.x_22171' into 1.8_release_3.9.x Jan 20, 2020
db2 添加db2维表和结果表 Nov 20, 2019
docs fix hbase sink Dec 5, 2019
elasticsearch5 [21724][essink]使用flink es sink 不设置id的时候分割异常 Jan 2, 2020
hbase extract method Dec 27, 2019
impala remove rdb predicate pushdown Dec 7, 2019
kafka Revert "Merge branch 'v1.8.0_dirtydata_manager' into 'v1.8.0_dev' " Dec 5, 2019
kafka08 remove flink shade class Nov 11, 2019
kafka09 remove no need Dec 12, 2019
kafka10 remove no need Dec 12, 2019
kafka11 remove no need Dec 12, 2019
kudu fix async side parse error Dec 27, 2019
launcher modify parmas position Dec 6, 2019
mongo Merge branch 'v1.8.0_dev_bugfix_leftjoinnull' into 'v1.8.0_dev' Nov 18, 2019
mysql remove flink shade class Nov 11, 2019
oracle remove flink shade class Nov 11, 2019
postgresql pgsql Nov 7, 2019
rdb [21756][core] flink-sql double 类型转换问题 Jan 6, 2020
redis5 extract method Dec 27, 2019
serversocket remove flink shade class Nov 11, 2019
sqlserver 去掉sqljdbc jar Nov 26, 2019
.gitignore Merge branch '1.5_v3.6.0' into v1.5.0_dev Jul 25, 2019
LICENSE.txt upload license Sep 12, 2018
README.md modify readme Jan 20, 2020
pom.xml pom Dec 3, 2019

README.md

flinkStreamSQL

  • 基于开源的flink,对其实时sql进行扩展
  • 自定义create table 语法(包括源表,输出表,维表)
  • 自定义create view 语法
  • 自定义create function 语法
  • 实现了流与维表的join
  • 支持原生FLinkSQL所有的语法
  • 扩展了输入和输出的性能指标到promethus

已支持

  • 源表:kafka 0.9、0.10、0.11、1.x版本
  • 维表:mysql, SQlServer,oracle, hbase, mongo, redis, cassandra, serversocket, kudu, postgresql, clickhouse, impala, db2, sqlserver
  • 结果表:mysql, SQlServer, oracle, hbase, elasticsearch5.x, mongo, redis, cassandra, console, kudu, postgresql, clickhouse, impala, db2, sqlserver

1 快速起步

1.1 运行模式

  • 单机模式:对应Flink集群的单机模式
  • standalone模式:对应Flink集群的分布式模式
  • yarn模式:对应Flink集群的yarn模式

1.2 执行环境

  • Java: JDK8及以上
  • Flink集群: 1.4,1.5,1.8(单机模式不需要安装Flink集群)
  • 操作系统:理论上不限

1.3 打包

进入项目根目录,使用maven打包:

mvn clean package -Dmaven.test.skip

打包结束后,项目根目录下会产生plugins目录,plugins目录下存放编译好的数据同步插件包,在lib目下存放job提交的包

1.4 启动

1.4.1 启动命令

sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\} -yarnSessionConf \{\"yid\":\"application_1564971615273_38182\"}

1.4.2 命令行参数选项

  • mode

    • 描述:执行模式,也就是flink集群的工作模式
      • local: 本地模式
      • standalone: 提交到独立部署模式的flink集群
      • yarn: 提交到yarn模式的flink集群(即提交到已有flink集群)
      • yarnPer: yarn per_job模式提交(即创建新flink application)
    • 必选:否
    • 默认值:local
  • name

    • 描述:flink 任务对应名称。
    • 必选:是
    • 默认值:无
  • sql

    • 描述:执行flink sql 的主体语句。
    • 必选:是
    • 默认值:无
  • localSqlPluginPath

    • 描述:本地插件根目录地址,也就是打包后产生的plugins目录。
    • 必选:是
    • 默认值:无
  • remoteSqlPluginPath

    • 描述:flink执行集群上的插件根目录地址(将打包好的插件存放到各个flink节点上,如果是yarn集群需要存放到所有的nodemanager上)。
    • 必选:否
    • 默认值:无
  • addjar

    • 描述:扩展jar路径,当前主要是UDF定义的jar;
    • 格式:json
    • 必选:否
    • 默认值:无
  • confProp

    • 描述:一些参数设置
    • 格式: json
    • 必选:是 (如无参数填写空json即可)
    • 默认值:无
    • 可选参数:
      • sql.ttl.min: 最小过期时间,大于0的整数,如1d、1h(d\D:天,h\H:小时,m\M:分钟,s\s:秒)
      • sql.ttl.max: 最大过期时间,大于0的整数,如2d、2h(d\D:天,h\H:小时,m\M:分钟,s\s:秒),需同时设置最小时间,且比最小时间大5分钟
      • state.backend: 任务状态后端,可选为MEMORY,FILESYSTEM,ROCKSDB,默认为flinkconf中的配置。
      • state.checkpoints.dir: FILESYSTEM,ROCKSDB状态后端文件系统存储路径,例如:hdfs://ns1/dtInsight/flink180/checkpoints。
      • state.backend.incremental: ROCKSDB状态后端是否开启增量checkpoint,默认为true。
      • sql.env.parallelism: 默认并行度设置
      • sql.max.env.parallelism: 最大并行度设置
      • time.characteristic: 可选值[ProcessingTime|IngestionTime|EventTime]
      • sql.checkpoint.interval: 设置了该参数表明开启checkpoint(ms)
      • sql.checkpoint.mode: 可选值[EXACTLY_ONCE|AT_LEAST_ONCE]
      • sql.checkpoint.timeout: 生成checkpoint的超时时间(ms)
      • sql.max.concurrent.checkpoints: 最大并发生成checkpoint数
      • sql.checkpoint.cleanup.mode: 默认是不会将checkpoint存储到外部存储,[true(任务cancel之后会删除外部存储)|false(外部存储需要手动删除)]
      • flinkCheckpointDataURI: 设置checkpoint的外部存储路径,根据实际的需求设定文件路径,hdfs://, file://
      • jobmanager.memory.mb: per_job模式下指定jobmanager的内存大小(单位MB, 默认值:768)
      • taskmanager.memory.mb: per_job模式下指定taskmanager的内存大小(单位MB, 默认值:768)
      • taskmanager.num: per_job模式下指定taskmanager的实例数(默认1)
      • taskmanager.slots:per_job模式下指定每个taskmanager对应的slot数量(默认1)
      • savePointPath:任务恢复点的路径(默认无)
      • allowNonRestoredState:指示保存点是否允许非还原状态的标志(默认false)
      • prometheus 相关参数 per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
  • flinkconf

    • 描述:flink配置文件所在的目录(单机模式下不需要),如/hadoop/flink-1.4.0/conf
    • 必选:否
    • 默认值:无
  • yarnconf

    • 描述:Hadoop配置文件(包括hdfs和yarn)所在的目录(单机模式下不需要),如/hadoop/etc/hadoop
    • 必选:否
    • 默认值:无
  • flinkJarPath

    • 描述:per_job 模式提交需要指定本地的flink jar存放路径
    • 必选:否
    • 默认值:false
  • queue

    • 描述:per_job 模式下指定的yarn queue
    • 必选:否
    • 默认值:false
  • pluginLoadMode

    • 描述:per_job 模式下的插件包加载方式。classpath:从每台机器加载插件包,shipfile:将需要插件从提交的节点上传到hdfs,不需要每台安装插件
    • 必选:否
    • 默认值:classpath
  • yarnSessionConf

    • 描述:yarn session 模式下指定的运行的一些参数,可参考,目前只支持指定yid
    • 必选:否
    • 默认值:false

2 结构

2.1 源表插件

2.2 结果表插件

2.3 维表插件

3 性能指标(新增)

kafka插件

  • 业务延迟: flink_taskmanager_job_task_operator_dtEventDelay(单位s)
    数据本身的时间和进入flink的当前时间的差值.

  • 各个输入源的脏数据:flink_taskmanager_job_task_operator_dtDirtyData
    从kafka获取的数据解析失败的视为脏数据

  • 各Source的数据输入TPS: flink_taskmanager_job_task_operator_dtNumRecordsInRate
    kafka接受的记录数(未解析前)/s

  • 各Source的数据输入RPS: flink_taskmanager_job_task_operator_dtNumRecordsInResolveRate
    kafka接受的记录数(解析后)/s

  • 各Source的数据输入BPS: flink_taskmanager_job_task_operator_dtNumBytesInRate
    kafka接受的字节数/s

  • Kafka作为输入源的各个分区的延迟数: flink_taskmanager_job_task_operator_topic_partition_dtTopicPartitionLag
    当前kafka10,kafka11有采集该指标

  • 各个输出源RPS: flink_taskmanager_job_task_operator_dtNumRecordsOutRate
    写入的外部记录数/s

4 样例


CREATE (scala|table|aggregate) FUNCTION CHARACTER_LENGTH WITH com.dtstack.Kun;


CREATE TABLE MyTable(
    name varchar,
    channel varchar,
    pv int,
    xctime bigint,
    CHARACTER_LENGTH(channel) AS timeLeng //自定义的函数
 )WITH(
    type ='kafka09',
    bootstrapServers ='172.16.8.198:9092',
    zookeeperQuorum ='172.16.8.198:2181/kafka',
    offsetReset ='latest',
    topic ='nbTest1',
    parallelism ='1'
 );

CREATE TABLE MyResult(
    channel varchar,
    pv varchar
 )WITH(
    type ='mysql',
    url ='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
    userName ='dtstack',
    password ='abc123',
    tableName ='pv2',
    parallelism ='1'
 );

CREATE TABLE workerinfo(
    cast(logtime as TIMESTAMP) AS rtime,
    cast(logtime) AS rtime
 )WITH(
    type ='hbase',
    zookeeperQuorum ='rdos1:2181',
    tableName ='workerinfo',
    rowKey ='ce,de',
    parallelism ='1',
    zookeeperParent ='/hbase'
 );

CREATE TABLE sideTable(
    cf:name varchar as name,
    cf:info varchar as info,
    PRIMARY KEY(name),
    PERIOD FOR SYSTEM_TIME //维表标识
 )WITH(
    type ='hbase',
    zookeeperQuorum ='rdos1:2181',
    zookeeperParent ='/hbase',
    tableName ='workerinfo',
    cache ='LRU',
    cacheSize ='10000',
    cacheTTLMs ='60000',
    parallelism ='1'
 );

insert
into
    MyResult
    select
        d.channel,
        d.info
    from
        (      select
            a.*,b.info
        from
            MyTable a
        join
            sideTable b
                on a.channel=b.name
        where
            a.channel = 'xc2'
            and a.pv=10      ) as d

招聘

1.大数据平台开发工程师,想了解岗位详细信息可以添加本人微信号ysqwhiletrue,注明招聘,如有意者发送简历至sishu@dtstack.com

You can’t perform that action at this time.