How to use

The installation of the one-stop platform streampark-console has been introduced in detail in the previous chapter. In this chapter, let’s see how to quickly deploy and run a job with streampark-console. The official structure and specification) and projects developed with streampark are well supported. Let’s use streampark-quickstart to quickly start the journey of streampark-console

streampark-quickstart is a sample program for developing Flink by StreamPark. For details, please refer to:

Deploy DataStream tasks

The following example demonstrates how to deploy a DataStream application

Deploy the FlinkSql task

The following example demonstrates how to deploy a FlinkSql application

  • The flink sql used in the project demonstration is as follows
  1. CREATE TABLE user_log (
  2.     user_id VARCHAR,
  3.     item_id VARCHAR,
  4.     category_id VARCHAR,
  5.     behavior VARCHAR,
  6.     ts TIMESTAMP(3)
  7.  ) WITH (
  8. 'connector' = 'kafka', -- Using the kafka connector
  9. 'properties.group.id' = 'group01' ,
  10. 'topic' = 'user_behavior',  -- kafka topic
  11. 'properties.bootstrap.servers'='kafka-1:9092,kafka-2:9092,kafka-3:9092',
  12. 'scan.startup.mode' = 'earliest-offset', -- Read from start offset
  13. 'format' = 'json'  -- The data source format is json
  14.  );
  16. CREATE TABLE pvuv_sink (
  17.     dt VARCHAR,
  18.     pv BIGINT,
  19.     uv BIGINT,
  20.     PRIMARY KEY (dt,pv,uv) NOT ENFORCED
  21.  ) WITH (
  22. 'connector' = 'jdbc', -- using jdbc connector
  23. 'url' = 'jdbc:mysql://test-mysql:3306/test', -- jdbc url
  24. 'table-name' = 'pvuv_sink', -- Table Name
  25. 'username' = 'root', -- username
  26. 'password' = '123456', --password
  27. 'sink.buffer-flush.max-rows' = '1' -- Default 5000, changed to 1 for demonstration
  28.  );
  30. INSERT INTO pvuv_sink
  31. SELECT
  32.   DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') dt,
  33.   COUNT(*) AS pv,
  34.   COUNT(DISTINCT user_id) AS uv
  35. FROM user_log
  36. GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00');
  • The maven dependencies are used as follows
  2. <dependency>
  3.     <groupId>mysql</groupId>
  4.     <artifactId>mysql-connector-java</artifactId>
  5.     <version>5.1.48</version>
  6. </dependency>
  8. <dependency>
  9.     <groupId>org.apache.flink</groupId>
  10.     <artifactId>flink-sql-connector-kafka_2.11</artifactId>
  11.     <version>1.14.6</version>
  12. </dependency>
  14. <dependency>
  15.     <groupId>org.apache.flink</groupId>
  16.     <artifactId>flink-connector-jdbc_2.11</artifactId>
  17.     <version>1.14.6</version>
  18. </dependency>
  • The data sent by Kafka simulation is as follows
  2. {"user_id": "543462", "item_id":"1715", "category_id": "1464116", "behavior": "pv", "ts":"2021-02-01 01:00:00"}
  3. {"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "pv", "ts":"2021-02-01 01:00:00"}
  4. {"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "pv", "ts":"2021-02-01 01:00:00"}
  5. {"user_id": "662867", "item_id":"2244074","category_id":"1575622","behavior": "learning flink", "ts":"2021-02-01 01:00:00"}

Task start process

The task startup flow chart is as follows

Quick Start - 图1

streampark-console submit task process

Regarding the concept of the project, Development Mode, savepoint, NoteBook, custom jar management, task release, task recovery, parameter configuration, parameter comparison, multi-version management and more tutorials and documents will be continuously updated. ..