使用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://ip:10000
• 访问Flink http://ip:8081
• 默认的用户和密码分别为 admin 和 streampark

该部署方式会自动给你启动一个flink-session集群供你去进行flink任务使用，同时也会挂载本地docker服务以及~/.kube来用于k8s模式的任务提交

1. 配置flink home
2. 配置session集群注意：在配置flink-sessin集群地址时，填写的ip地址不是localhost，而是宿主机网络ip，该ip地址可以通过ifconfig来进行获取
3. 提交任务

??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 管理，任务发布，任务恢复，参数配置，参数对比，多版本管理等等更多使用教程和文档后续持续更新。