🚩使用Docker完成StreamPark的部署
✔️1.基于h2和docker-compose进行StreamPark部署
wget https://raw.githubusercontent.com/apache/incubator-streampark/dev/deploy/docker/docker-compose.yaml
wget https://raw.githubusercontent.com/apache/incubator-streampark/dev/deploy/docker/.env
docker-compose up -d
访问StreamPark http://g.htmltoo.com:10000
访问Flink http://g.htmltoo.com:8081
默认的用户和密码分别为 admin 和 streampark
该部署方式会自动给你启动一个flink-session集群供你去进行flink任务使用,同时也会挂载本地docker服务以及~/.kube来用于k8s模式的任务提交
配置flink home
配置session集群注意:在配置flink-sessin集群地址时,填写的ip地址不是localhost,而是宿主机网络ip,该ip地址可以通过ifconfig来进行获取
提交任务
✔️2.沿用已有的 Mysql 服务
该方式适用于企业生产,你可以基于docker快速部署strempark并将其和线上数据库进行关联 注意:部署支持的多样性是通过.env这个配置文件来进行维护的,要保证目录下有且仅有一个.env文件
wget https://raw.githubusercontent.com/apache/incubator-streampark/dev/deploy/docker/docker-compose.yaml
wget https://raw.githubusercontent.com/apache/incubator-streampark/dev/deploy/docker/mysql/.env
vim .env
需要先在mysql先创建streampark数据库,然后手动执行对应的schema和data里面对应数据源的sql 然后修改对应的连接信息
SPRING_PROFILES_ACTIVE=mysql
SPRING_DATASOURCE_URL=jdbc:mysql://localhost:3306/streampark?useSSL=false&useUnicode=true&characterEncoding=UTF-8&allowPublicKeyRetrieval=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8
SPRING_DATASOURCE_USERNAME=root
SPRING_DATASOURCE_PASSWORD=streampark
docker-compose up -d
✔️3.基于源码构建镜像进行StreamPark部署
git clone https://github.com/apache/incubator-streampark.git
cd incubator-streampark/deploy/docker
vim docker-compose.yaml
version: '3.8'
services:
## streampark-console容器
streampark-console:
## streampark的镜像
image: apache/streampark:latest
## streampark的镜像启动命令
command: ${
RUN_COMMAND}
ports:
- 10000:10000
## 环境配置文件
env_file: .env
environment:
## 声明环境变量
HADOOP_HOME: ${
HADOOP_HOME}
volumes:
- flink:/streampark/flink/${
FLINK}
- /var/run/docker.sock:/var/run/docker.sock
- /etc/hosts:/etc/hosts:ro
- ~/.kube:/root/.kube:ro
privileged: true
restart: unless-stopped
networks:
- streampark
## flink-jobmanager容器
flink-jobmanager:
image: ${
FLINK_IMAGE}
ports:
- "8081:8081"
command: jobmanager
volumes:
- flink:/opt/flink
env_file: .env
restart: unless-stopped
privileged: true
networks:
- streampark
## streampark-taskmanager容器
flink-taskmanager:
image: ${
FLINK_IMAGE}
depends_on:
- flink-jobmanager
command: taskmanager
deploy:
replicas: 1
env_file: .env
restart: unless-stopped
privileged: true
networks:
- streampark
networks:
streampark:
driver: bridge
volumes:
flink:
最后,执行启动命令:
cd deploy/docker
docker-compose up -d
上传配置至容器
在前面的env文件,声明了HADOOP_HOME,对应的目录为“/streampark/hadoop”,所以需要上传hadoop安装包下的/etc/hadoop至/streampark/hadoop目录,命令如下:
## 上传hadoop资源
docker cp etc整个目录 streampark-docker_streampark-console_1:/streampark/hadoop
## 进入容器
docker exec -it streampark-docker_streampark-console_1 bash
## 查看
ls
同时,其它配置文件,如maven的settings.xml文件也是以同样的方式上传。
🚩什么是StreamPark
实时即未来,在实时处理流域 Apache Spark 和 Apache Flink 是一个伟大的进步,尤其是Apache Flink被普遍认为是下一代大数据流计算引擎, 我们在使用 Flink & Spark 时发现从编程模型, 启动配置到运维管理都有很多可以抽象共用的地方, 我们将一些好的经验固化下来并结合业内的最佳实践, 通过不断努力终于诞生了今天的框架 —— StreamPark, 项目的初衷是 —— 让流处理更简单, 使用StreamPark开发,可以极大降低学习成本和开发门槛, 让开发者只用关心最核心的业务,StreamPark 规范了项目的配置,鼓励函数式编程,定义了最佳的编程方式,提供了一系列开箱即用的Connectors,标准化了配置、开发、测试、部署、监控、运维的整个过程, 提供了scala和java两套api, 其最终目的是打造一个一站式大数据平台,流批一体,湖仓一体的解决方案.
🚩组成部分
StreamPark有三部分组成,分别是streampark-core,streampark-pump 和 streampark-console
➡️streampark-core
streampark-core 定位是一个开发时框架,关注编码开发,规范了配置文件,按照约定优于配置的方式进行开发,提供了一个开发时 RunTime Content和一系列开箱即用的Connector,扩展了DataStream相关的方法,融合了DataStream和Flink sql api,简化繁琐的操作,聚焦业务本身,提高开发效率和开发体验
➡️streampark-pump
pump 是抽水机,水泵的意思,streampark-pump的定位是一个数据抽取的组件,类似于flinkx,基于streampark-core中提供的各种connector开发,目的是打造一个方便快捷,开箱即用的大数据实时数据抽取和迁移组件,并且集成到streampark-console中,解决实时数据源获取问题,目前在规划中
➡️streampark-console
streampark-console 是一个综合实时数据平台,低代码(Low Code)平台,可以较好的管理Flink任务,集成了项目编译、发布、参数配置、启动、savepoint,火焰图(flame graph),Flink SQL, 监控等诸多功能于一体,大大简化了Flink任务的日常操作和维护,融合了诸多最佳实践。旧时王谢堂前燕,飞入寻常百姓家,让大公司有能力研发使用的项目,现在人人可以使用, 其最终目标是打造成一个实时数仓,流批一体的一站式大数据解决方案,该平台使用但不仅限以下技术
Flink,Apache YARN,Spring Boot,Mybatis,Mybatis-Plus,Flame Graph,JVM-Profiler,Vue,VuePress,Ant Design of Vue,ANTD PRO VUE,xterm.js,Monaco Editor,...
🚩 快速开始
0.streampark-quickstart
https://github.com/apache/incubator-streampark-quickstart Apache StreamPark 开发Flink的上手示例程序,分为四个模块:
quickstart-apacheflink quickstart-datastream quickstart-flinksql quickstart-connector
➡️quickstart-apacheflink
quickstart-apacheflink 是Flink官方的SocketWindowWordCount程序, 其中项目结构,打包规则,开发方式等和Flink官网要求的规范完全一致, 之所以有这个模块,是因为在streampark-console中做了对标准的Flink程序(按照官方要求的开发规范开发的Flink程序)的部署支持,后续方便演示使用.
➡️quickstart-datastream
该模块主要演示了如果利用Apache StreamPark快速开发一个DataStream程序,其中有java和scala两种语言的开发示例,供开发者快速上手学习使用
➡️quickstart-flinksql
该模块主要演示了如果利用Apache StreamPark快速开发一个Flink & SQL程序,其中有java和scala两种语言的开发示例,供开发者快速上手学习使用
➡️quickstart-connector
该模块演示了如何使用Apache StreamPark里提供的各种 Datastream connector , 里面集合了各种 Datastream connector 的配置和使用示例, 供开发者快速上手学习使用
1.部署 DataStream 任务
http://assets.streamxhub.com/datastream.mp4
2.部署 FlinkSql 任务
项目演示使用到的 flink sql:
CREATE TABLE user_log (
user_id VARCHAR,
item_id VARCHAR,
category_id VARCHAR,
behavior VARCHAR,
ts TIMESTAMP(3)
) WITH (
'connector.type' = 'kafka', -- 使用 kafka connector
'connector.version' = 'universal', -- kafka 版本,universal 支持 0.11 以上的版本
'connector.topic' = 'user_behavior', -- kafka topic
'connector.properties.bootstrap.servers'='kafka-1:9092,kafka-2:9092,kafka-3:9092',
'connector.startup-mode' = 'earliest-offset', -- 从起始 offset 开始读取
'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://test-mysql:3306/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');
使用到 maven 依赖如下:
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.48</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-connector-kafka_2.11</artifactId>
<version>1.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.11</artifactId>
<version>1.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>1.12.0</version>
</dependency>
Kafka 模拟发送的数据如下
{"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts":"2021-02-01T01:00:00Z"}
{"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "pv", "ts":"2021-02-01T01:00:00Z"}
{"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "pv", "ts":"2021-02-01T01:00:00Z"}
{"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "learning flink", "ts":"2021-02-01T01:00:00Z"}
3.任务启动流程
任务启动流程图如下关于项目的概念,Development Mode,savepoint,NoteBook,自定义 jar 管理,任务发布,任务恢复,参数配置,参数对比,多版本管理等等更多使用教程和文档后续持续更新。
评论区