Skip to content

Commit 07dad77

Browse files
author
xuchao
committed
Merge branch 'hotfix_1.8_3.10.x_24296' into 1.8_release
# Conflicts: # README.md
2 parents d1e9030 + d699874 commit 07dad77

File tree

75 files changed

+2264
-1327
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

75 files changed

+2264
-1327
lines changed

README.md

+30-309
Original file line numberDiff line numberDiff line change
@@ -1,315 +1,36 @@
1-
1+
FlinkStreamSQL
2+
============
3+
[![License](https://img.shields.io/badge/license-Apache%202-4EB1BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)
4+
25
## 技术交流
3-
* 招聘大数据平台开发工程师,想了解岗位详细信息可以添加本人微信号ysqwhiletrue,注明招聘,如有意者发送简历至[email protected]
4-
5-
* 可以搜索群号30537511或者可以扫下面的二维码进入钉钉群
6+
- 招聘**大数据平台开发工程师**,想了解岗位详细信息可以添加本人微信号ysqwhiletrue,注明招聘,如有意者发送简历至[[email protected]](mailto:[email protected])
7+
- 我们使用[钉钉](https://www.dingtalk.com/)沟通交流,可以搜索群号[**30537511**]或者扫描下面的二维码进入钉钉群
68
<div align=center>
7-
<img src=https://github.com/DTStack/flinkStreamSQL/blob/1.8_release/images/IMG_1573.JPG width=300 />
9+
<img src=docs/images/streamsql_dd.jpg width=300 />
810
</div>
911

10-
11-
## flinkStreamSQL
12-
> * 基于开源的flink,对其实时sql进行扩展
13-
> > * 自定义create table 语法(包括源表,输出表,维表)
14-
> > * 自定义create view 语法
15-
> > * 自定义create function 语法
16-
> > * 实现了流与维表的join
17-
> > * 支持原生FLinkSQL所有的语法
18-
> > * 扩展了输入和输出的性能指标到promethus
19-
20-
## 已支持
21-
* 源表:kafka 0.9、0.10、0.11、1.x版本
22-
* 维表:mysql, SQlServer,oracle, hbase, mongo, redis, cassandra, serversocket, kudu, postgresql, clickhouse, impala, db2, sqlserver
23-
* 结果表:mysql, SQlServer, oracle, hbase, elasticsearch5.x, mongo, redis, cassandra, console, kudu, postgresql, clickhouse, impala, db2, sqlserver
24-
25-
26-
## 1 快速起步
27-
### 1.1 运行模式
28-
29-
30-
* 单机模式:对应Flink集群的单机模式
31-
* standalone模式:对应Flink集群的分布式模式
32-
* yarn模式:对应Flink集群的yarn模式
33-
34-
### 1.2 执行环境
35-
36-
* Java: JDK8及以上
37-
* Flink集群: 1.4,1.5,1.8(单机模式不需要安装Flink集群)
38-
* 操作系统:理论上不限
39-
* kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数,配置案例:
40-
```
41-
## hadoop配置文件路径
42-
fs.hdfs.hadoopconf: /Users/maqi/tmp/hadoopconf/hadoop_250
43-
security.kerberos.login.use-ticket-cache: true
44-
security.kerberos.login.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/maqi.keytab
45-
security.kerberos.login.principal: [email protected]
46-
security.kerberos.login.contexts: Client,KafkaClient
47-
zookeeper.sasl.service-name: zookeeper
48-
zookeeper.sasl.login-context-name: Client
49-
50-
```
51-
52-
### 1.3 打包
53-
54-
进入项目根目录,使用maven打包:
55-
56-
```
57-
mvn clean package -Dmaven.test.skip
58-
59-
```
60-
61-
打包完成后的包结构:
62-
63-
> * dt-center-flinkStreamSQL
64-
> > * bin: 任务启动脚本
65-
> > * lib: launcher包存储路径,是任务提交的入口
66-
> > * plugins: 插件包存储路径
67-
> > * ........ : core及插件代码
68-
69-
### 1.4 启动
70-
71-
#### 1.4.1 启动命令
72-
73-
```
74-
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\"}
75-
```
76-
77-
#### 1.4.2 命令行参数选项
78-
79-
* **mode**
80-
* 描述:执行模式,也就是flink集群的工作模式
81-
* local: 本地模式
82-
* standalone: 提交到独立部署模式的flink集群
83-
* yarn: 提交到yarn模式的flink集群(即提交到已有flink集群)
84-
* yarnPer: yarn per_job模式提交(即创建新flink application)
85-
* 必选:否
86-
* 默认值:local
87-
88-
* **name**
89-
* 描述:flink 任务对应名称。
90-
* 必选:是
91-
* 默认值:无
92-
93-
* **sql**
94-
* 描述:执行flink sql 的主体语句。
95-
* 必选:是
96-
* 默认值:无
97-
98-
* **localSqlPluginPath**
99-
* 描述:本地插件根目录地址,也就是打包后产生的plugins目录。
100-
* 必选:是
101-
* 默认值:无
102-
103-
* **remoteSqlPluginPath**
104-
* 描述:flink执行集群上的插件根目录地址(将打包好的插件存放到各个flink节点上,如果是yarn集群需要存放到所有的nodemanager上)。
105-
* 必选:否
106-
* 默认值:无
107-
108-
* **addjar**
109-
* 描述:扩展jar路径,当前主要是UDF定义的jar;
110-
* 格式:json
111-
* 必选:否
112-
* 默认值:无
12+
## 介绍
13+
* 基于开源的flink,对其实时sql进行扩展
14+
* 自定义create table 语法(包括源表,输出表,维表)
15+
* 自定义create view 语法
16+
* 自定义create function 语法
17+
* 实现了流与维表的join
18+
* 支持原生FLinkSQL所有的语法
19+
* 扩展了输入和输出的性能指标到promethus
11320

114-
* **confProp**
115-
* 描述:一些参数设置
116-
* 格式: json
117-
* 必选:是 (如无参数填写空json即可)
118-
* 默认值:无
119-
* 可选参数:
120-
* sql.ttl.min: 最小过期时间,大于0的整数,如1d、1h(d\D:天,h\H:小时,m\M:分钟,s\s:秒)
121-
* sql.ttl.max: 最大过期时间,大于0的整数,如2d、2h(d\D:天,h\H:小时,m\M:分钟,s\s:秒),需同时设置最小时间,且比最小时间大5分钟
122-
* state.backend: 任务状态后端,可选为MEMORY,FILESYSTEM,ROCKSDB,默认为flinkconf中的配置。
123-
* state.checkpoints.dir: FILESYSTEM,ROCKSDB状态后端文件系统存储路径,例如:hdfs://ns1/dtInsight/flink180/checkpoints。
124-
* state.backend.incremental: ROCKSDB状态后端是否开启增量checkpoint,默认为true。
125-
* sql.env.parallelism: 默认并行度设置
126-
* sql.max.env.parallelism: 最大并行度设置
127-
* time.characteristic: 可选值[ProcessingTime|IngestionTime|EventTime]
128-
* sql.checkpoint.interval: 设置了该参数表明开启checkpoint(ms)
129-
* sql.checkpoint.mode: 可选值[EXACTLY_ONCE|AT_LEAST_ONCE]
130-
* sql.checkpoint.timeout: 生成checkpoint的超时时间(ms)
131-
* sql.max.concurrent.checkpoints: 最大并发生成checkpoint数
132-
* sql.checkpoint.cleanup.mode: 默认是不会将checkpoint存储到外部存储,[true(任务cancel之后会删除外部存储)|false(外部存储需要手动删除)]
133-
* flinkCheckpointDataURI: 设置checkpoint的外部存储路径,根据实际的需求设定文件路径,hdfs://, file://
134-
* jobmanager.memory.mb: per_job模式下指定jobmanager的内存大小(单位MB, 默认值:768)
135-
* taskmanager.memory.mb: per_job模式下指定taskmanager的内存大小(单位MB, 默认值:768)
136-
* taskmanager.num: per_job模式下指定taskmanager的实例数(默认1)
137-
* taskmanager.slots:per_job模式下指定每个taskmanager对应的slot数量(默认1)
138-
* savePointPath:任务恢复点的路径(默认无)
139-
* allowNonRestoredState:指示保存点是否允许非还原状态的标志(默认false)
140-
* logLevel: 日志级别动态配置(默认info)
141-
* [prometheus 相关参数](docs/prometheus.md) per_job可指定metric写入到外部监控组件,以prometheus pushgateway举例
21+
## 目录
22+
23+
[ 1.1 demo](docs/demo.md)
24+
[ 1.2 快速开始](docs/quickStart.md)
25+
[ 1.3 参数配置](docs/config.md)
26+
[ 1.4 支持的插件介绍和demo](docs/pluginsInfo.md)
27+
[ 1.5 指标参数](docs/newMetric.md)
28+
[ 1.6 自定义函数](docs/function.md)
29+
[ 1.7 自定义视图](docs/createView.md)
30+
31+
## 如何贡献FlinkStreamSQL
14232

143-
144-
* **flinkconf**
145-
* 描述:flink配置文件所在的目录(单机模式下不需要),如/hadoop/flink-1.4.0/conf
146-
* 必选:否
147-
* 默认值:无
148-
149-
* **yarnconf**
150-
* 描述:Hadoop配置文件(包括hdfs和yarn)所在的目录(单机模式下不需要),如/hadoop/etc/hadoop
151-
* 必选:否
152-
* 默认值:无
153-
154-
* **flinkJarPath**
155-
* 描述:per_job 模式提交需要指定本地的flink jar存放路径
156-
* 必选:否
157-
* 默认值:false
158-
159-
* **queue**
160-
* 描述:per_job 模式下指定的yarn queue
161-
* 必选:否
162-
* 默认值:false
163-
164-
* **pluginLoadMode**
165-
* 描述:per_job 模式下的插件包加载方式。classpath:从每台机器加载插件包,shipfile:将需要插件从提交的节点上传到hdfs,不需要每台安装插件
166-
* 必选:否
167-
* 默认值:classpath
168-
169-
* **yarnSessionConf**
170-
* 描述:yarn session 模式下指定的运行的一些参数,[可参考](https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/cli.html),目前只支持指定yid
171-
* 必选:否
172-
* 默认值:false
173-
174-
175-
## 2 结构
176-
### 2.1 源表插件
177-
* [kafka 源表插件](docs/kafkaSource.md)
178-
179-
### 2.2 结果表插件
180-
* [elasticsearch 结果表插件](docs/elasticsearchSink.md)
181-
* [hbase 结果表插件](docs/hbaseSink.md)
182-
* [mysql 结果表插件](docs/mysqlSink.md)
183-
* [oracle 结果表插件](docs/oracleSink.md)
184-
* [mongo 结果表插件](docs/mongoSink.md)
185-
* [redis 结果表插件](docs/redisSink.md)
186-
* [cassandra 结果表插件](docs/cassandraSink.md)
187-
* [kudu 结果表插件](docs/kuduSink.md)
188-
* [postgresql 结果表插件](docs/postgresqlSink.md)
189-
* [clickhouse 结果表插件](docs/clickhouseSink.md)
190-
* [impala 结果表插件](docs/impalaSink.md)
191-
* [db2 结果表插件](docs/db2Sink.md)
192-
* [sqlserver 结果表插件](docs/sqlserverSink.md)
193-
194-
### 2.3 维表插件
195-
* [hbase 维表插件](docs/hbaseSide.md)
196-
* [mysql 维表插件](docs/mysqlSide.md)
197-
* [oracle 维表插件](docs/oracleSide.md)
198-
* [mongo 维表插件](docs/mongoSide.md)
199-
* [redis 维表插件](docs/redisSide.md)
200-
* [cassandra 维表插件](docs/cassandraSide.md)
201-
* [kudu 维表插件](docs/kuduSide.md)
202-
* [postgresql 维表插件](docs/postgresqlSide.md)
203-
* [clickhouse 维表插件](docs/clickhouseSide.md)
204-
* [impala 维表插件](docs/impalaSide.md)
205-
* [db2 维表插件](docs/db2Side.md)
206-
* [sqlserver 维表插件](docs/sqlserverSide.md)
207-
208-
## 3 性能指标(新增)
209-
210-
### kafka插件
211-
* 业务延迟: flink_taskmanager_job_task_operator_dtEventDelay(单位s)
212-
数据本身的时间和进入flink的当前时间的差值.
213-
214-
* 各个输入源的脏数据:flink_taskmanager_job_task_operator_dtDirtyData
215-
从kafka获取的数据解析失败的视为脏数据
216-
217-
* 各Source的数据输入TPS: flink_taskmanager_job_task_operator_dtNumRecordsInRate
218-
kafka接受的记录数(未解析前)/s
219-
220-
* 各Source的数据输入RPS: flink_taskmanager_job_task_operator_dtNumRecordsInResolveRate
221-
kafka接受的记录数(解析后)/s
222-
223-
* 各Source的数据输入BPS: flink_taskmanager_job_task_operator_dtNumBytesInRate
224-
kafka接受的字节数/s
225-
226-
* Kafka作为输入源的各个分区的延迟数: flink_taskmanager_job_task_operator_topic_partition_dtTopicPartitionLag
227-
当前kafka10,kafka11有采集该指标
228-
229-
* 各个输出源RPS: flink_taskmanager_job_task_operator_dtNumRecordsOutRate
230-
写入的外部记录数/s
231-
232-
233-
## 4 样例
234-
235-
```
236-
237-
CREATE (scala|table|aggregate) FUNCTION CHARACTER_LENGTH WITH com.dtstack.Kun;
238-
239-
240-
CREATE TABLE MyTable(
241-
name varchar,
242-
channel varchar,
243-
pv int,
244-
xctime bigint,
245-
CHARACTER_LENGTH(channel) AS timeLeng //自定义的函数
246-
)WITH(
247-
type ='kafka09',
248-
bootstrapServers ='172.16.8.198:9092',
249-
zookeeperQuorum ='172.16.8.198:2181/kafka',
250-
offsetReset ='latest',
251-
topic ='nbTest1',
252-
parallelism ='1'
253-
);
254-
255-
CREATE TABLE MyResult(
256-
channel varchar,
257-
pv varchar
258-
)WITH(
259-
type ='mysql',
260-
url ='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
261-
userName ='dtstack',
262-
password ='abc123',
263-
tableName ='pv2',
264-
parallelism ='1'
265-
);
266-
267-
CREATE TABLE workerinfo(
268-
cast(logtime as TIMESTAMP) AS rtime,
269-
cast(logtime) AS rtime
270-
)WITH(
271-
type ='hbase',
272-
zookeeperQuorum ='rdos1:2181',
273-
tableName ='workerinfo',
274-
rowKey ='ce,de',
275-
parallelism ='1',
276-
zookeeperParent ='/hbase'
277-
);
278-
279-
CREATE TABLE sideTable(
280-
cf:name varchar as name,
281-
cf:info varchar as info,
282-
PRIMARY KEY(name),
283-
PERIOD FOR SYSTEM_TIME //维表标识
284-
)WITH(
285-
type ='hbase',
286-
zookeeperQuorum ='rdos1:2181',
287-
zookeeperParent ='/hbase',
288-
tableName ='workerinfo',
289-
cache ='LRU',
290-
cacheSize ='10000',
291-
cacheTTLMs ='60000',
292-
parallelism ='1'
293-
);
294-
295-
insert
296-
into
297-
MyResult
298-
select
299-
d.channel,
300-
d.info
301-
from
302-
( select
303-
a.*,b.info
304-
from
305-
MyTable a
306-
join
307-
sideTable b
308-
on a.channel=b.name
309-
where
310-
a.channel = 'xc2'
311-
and a.pv=10 ) as d
312-
```
313-
314-
315-
33+
## License
34+
FlinkStreamSQL is under the Apache 2.0 license. See the [LICENSE](http://www.apache.org/licenses/LICENSE-2.0) file for details.
35+
36+

0 commit comments

Comments
 (0)