git clone https://github.com/streamxhub/streamx.gitcd Streamxmvn clean install -DskipTests -Denv=dev

.streamx-console-service-1.0.0├── bin│    ├── flame-graph│    ├──   └── *.py                             //火焰图相关功能脚本(内部使用,用户无需关注)│    ├── startup.sh                             //启动脚本  │    ├── setclasspath.sh                        //java环境变量相关的脚本(内部使用,用户无需关注)│    ├── shutdown.sh                            //停止脚本│    ├── yaml.sh                                //内部使用解析yaml参数的脚本(内部使用,用户无需关注)├── conf                           │    ├── application.yaml                       //项目的配置文件(注意不要改动名称)│    ├── application-prod.yml                   //项目的配置文件(开发者部署需要改动的文件,注意不要改动名称)│    ├── flink-application.template             //flink配置模板(内部使用,用户无需关注)│    ├── logback-spring.xml                     //logback│    └── streamx.sql                                 //工程初始化脚本├── lib│    └── *.jar                                  //项目的jar包├── plugins   │    ├── streamx-jvm-profiler-1.0.0.jar         //jvm-profiler,火焰图相关功能(内部使用,用户无需关注)│    └── streamx-flink-sqlcli-1.0.0.jar         //Flink SQl提交相关功能(内部使用,用户无需关注)├── logs                                        //程序log目录└── temp                                        //内部使用到的零时路径,不要删除

  datasource:    dynamic:      # 是否开启 SQL日志输出，生产环境建议关闭，有性能损耗      p6spy: true      hikari:        connection-timeout: 30000        max-lifetime: 1800000        max-pool-size: 15        min-idle: 5        connection-test-query: select 1        pool-name: HikariCP-DS-POOL      # 配置默认数据源      primary: primary      datasource:        # 数据源-1，名称为 primary        primary:          username: $user          password: $password          driver-class-name: com.mysql.cj.jdbc.Driver          url: jdbc:mysql://$host:$port/streamx?useUnicode=true&characterEncoding=UTF-8&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=GMT%2B8

cd streamx-console-service-1.0.0/binbash start.sh

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_sinkSELECT  DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,  COUNT(*) AS pv,  COUNT(DISTINCT user_id) AS uvFROM user_logGROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00');

<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.12</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>

{"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"}