Confluent Platform: ksqlDB 实时流处理 (quick start)

梦里梦外; 2022-10-07 10:52 309阅读 0赞

### 文章目录 ###

*  1, Confluent Platform介绍
 *   *  功能说明
 *  2, 快速部署: quick start
 *   *  a, 安装配置并启动服务
     *  b, 页面化操作 (Control Center)：创建topic并生成测试数据
     *  c, 使用ksqlDB :查看数据/ 创建table或stream

# 1, Confluent Platform介绍 #

*  Confluent Platform 是一个全面的事件流平台，使您能够以连续的实时流的形式轻松访问、存储和管理数据。Confluent 由 Apache Kafka® 的原始创建者构建，通过企业级功能扩展了 Kafka 的优势，同时消除了 Kafka 管理或监控的负担。它的核心是Apache Kafka，这是最流行的开源分布式流媒体平台，集成的服务有 Kafka, Kafka REST, ZooKeeper, Control Center(web可视化), ksqlDB Server(sql引擎), Connect, Schema Registry(表结构注册中心)
 *  Confluent Platform 让您可以专注于如何从数据中获取业务价值，而不必担心底层机制，例如数据如何在不同的系统之间传输或集成。Confluent Platform 简化了将数据源连接到 Kafka、构建流应用程序以及保护、监控和管理您的 Kafka 基础设施的过程。 通过将历史和实时数据集成到一个单一的、中心的事实来源中，Confluent 可以轻松构建全新类别的现代事件驱动应用程序，获得通用数据管道，并解锁具有完全可扩展性的强大新用例、性能和可靠性

## 功能说明 ##

*  连接其他数据库系统：提供与 Kafka 的融合连接器，利用 Kafka Connect API 将 Kafka 连接到其他系统，例如数据库、键值存储、搜索索引和文件系统。
 *  ETL实时数据抽取 和视图数据缓存(预先计算查询结果以加快数据读取)： ksqlDB 组件是 Kafka 的流式 SQL 引擎，它为 Kafka 上的流处理提供了一个易于使用但功能强大的交互式 SQL 接口，而无需使用 Java 或 Python 等编程语言编写代码。ksqlDB 具有可扩展性、弹性、容错性和实时性。它支持广泛的流操作，包括数据过滤、转换、聚合、连接、窗口化和会话化。

# 2, 快速部署: quick start #

Confluent Platform quickstart: [https://docs.confluent.io/platform/current/quickstart/ce-quickstart.html][https_docs.confluent.io_platform_current_quickstart_ce-quickstart.html]

<table> 
 <thead> 
  <tr> 
   <th align="left">类别</th> 
   <th align="left">命令</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">服务启动/停止/状态</td> 
   <td align="left">confluent local services start /stop /status /list</td> 
  </tr> 
  <tr> 
   <td align="left">停止并删除服务和数据</td> 
   <td align="left">confluent local destroy</td> 
  </tr> 
  <tr> 
   <td align="left">安装插件,重启connect</td> 
   <td align="left">confluent-hub install xx.zip ; kill <code>cat /tmp/confluent.413977/connect/connect.pid</code></td> 
  </tr> 
  <tr> 
   <td align="left">ksql connect 打开debug日志</td> 
   <td align="left">/opt/confluent/etc/kafka/connect-log4j.properties; tailf /tmp/confluent.413977/connect/logs/connect.log</td> 
  </tr> 
 </tbody> 
</table>

## a, 安装配置并启动服务 ##

<table> 
 <thead> 
  <tr> 
   <th align="left">启动的进程</th> 
   <th align="left">启动参数</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">1, org.apache.kafka.connect.cli.ConnectDistributed</td> 
   <td align="left">connect.properties</td> 
  </tr> 
  <tr> 
   <td align="left">2, io.confluent.controlcenter.ControlCenter</td> 
   <td align="left">control-center.properties</td> 
  </tr> 
  <tr> 
   <td align="left">3, kafka.Kafka</td> 
   <td align="left">kafka.properties</td> 
  </tr> 
  <tr> 
   <td align="left">4, io.confluent.kafkarest.KafkaRestMain</td> 
   <td align="left">kafka-rest.properties</td> 
  </tr> 
  <tr> 
   <td align="left">5, io.confluent.ksql.rest.server.KsqlServerMain</td> 
   <td align="left">ksql-server.properties</td> 
  </tr> 
  <tr> 
   <td align="left">6, io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain</td> 
   <td align="left">schema-registry.properties</td> 
  </tr> 
  <tr> 
   <td align="left">7, org.apache.zookeeper.server.quorum.QuorumPeerMain</td> 
   <td align="left">zookeeper.properties</td> 
  </tr> 
 </tbody> 
</table>

#1, 解压并配置环境变量
    [root@c7-docker confluent-6.1.1]# 
     ln -s /opt/confluent-6.1.1/ /opt/confluent
     export CONFLUENT_HOME=/opt/confluent
     echo 'export CONFLUENT_HOME=/opt/confluent' >>  /etc/profile
     export PATH=$PATH:$CONFLUENT_HOME/bin
     echo 'export PATH=$PATH:$CONFLUENT_HOME/bin' >> /etc/profile
    
    #2, 安装kafka连接器 kafka-connect-datagen
    #connector doc: https://docs.confluent.io/home/connect/overview.html
    #插件安装目录： /opt/confluent-6.1.1/share/confluent-hub-components 
    [root@c7-docker confluent-6.1.1]# grep 'plugin.path' /opt/confluent-6.1.1/etc/ -r
    /opt/confluent-6.1.1/etc/kafka/connect-standalone.properties:# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
    /opt/confluent-6.1.1/etc/kafka/connect-standalone.properties:plugin.path=/usr/share/java,/opt/confluent-6.1.1/share/confluent-hub-components
    /opt/confluent-6.1.1/etc/kafka/connect-distributed.properties:# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
    /opt/confluent-6.1.1/etc/kafka/connect-distributed.properties:plugin.path=/usr/share/java,/opt/confluent-6.1.1/share/confluent-hub-components
    /opt/confluent-6.1.1/etc/ksqldb/connect.properties:# plugin.path=
    /opt/confluent-6.1.1/etc/schema-registry/connect-avro-standalone.properties:# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
    /opt/confluent-6.1.1/etc/schema-registry/connect-avro-standalone.properties:plugin.path=share/java,/opt/confluent-6.1.1/share/confluent-hub-components
    /opt/confluent-6.1.1/etc/schema-registry/connect-avro-distributed.properties:# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
    /opt/confluent-6.1.1/etc/schema-registry/connect-avro-distributed.properties:plugin.path=share/java,/opt/confluent-6.1.1/share/confluent-hub-components
    /opt/confluent-6.1.1/etc/kafka-connect-replicator/replicator-connect-distributed.properties:#plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors
    /opt/confluent-6.1.1/etc/kafka-connect-replicator/replicator-connect-standalone.properties:#plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors
    
    #本地编译安装： git clone https://github.com/confluentinc/kafka-connect-datagen.git
    # git checkout v0.4.0
    # mvn clean package
    # confluent-hub install target/components/packages/confluentinc-kafka-connect-datagen-0.4.0.zip 
    ### Usage: confluent-hub install : install a component from either Confluent Hub or from a local file
    [root@c7-docker confluent-6.1.1]# confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:latest
    Running in a "--no-prompt" mode
    Implicit acceptance of the license below:
    Apache License 2.0
    https://www.apache.org/licenses/LICENSE-2.0
    Downloading component Kafka Connect Datagen 0.5.0, provided by Confluent, Inc. from Confluent Hub and installing into /opt/confluent-6.1.1//share/confluent-hub-components
    Adding installation directory to plugin path in the following files:
      /opt/confluent-6.1.1//etc/kafka/connect-distributed.properties
      /opt/confluent-6.1.1//etc/kafka/connect-standalone.properties
      /opt/confluent-6.1.1//etc/schema-registry/connect-avro-distributed.properties
      /opt/confluent-6.1.1//etc/schema-registry/connect-avro-standalone.properties
    Completed
    
    #3, 修改配置文件 ( 默认ksqlDB的连接地址为 localhost:8088, 防止远程连接 http://192.168.56.7:9021/ 查询的sql会报错)
    [root@c7-docker confluent-6.1.1]# cd etc/
    [root@c7-docker etc]# ls
    cli                       confluent-control-center-fe  confluent-metadata-service  kafka                     ksqldb
    confluent-common          confluent-hub-client         confluent-rebalancer        kafka-connect-replicator  rest-utils
    confluent-control-center  confluent-kafka-mqtt         confluent-security          kafka-rest                schema-registry
    [root@c7-docker etc]# grep ':8088' * -r
    confluent-control-center/control-center.properties:#confluent.controlcenter.ksql.ksqlDB.url=http://localhost:8088
    confluent-control-center/control-center.properties:confluent.controlcenter.ksql.ksqlDB.url=http://192.168.56.117:8088
    confluent-control-center/control-center-minimal.properties:#confluent.controlcenter.ksql.ksqlDB.url=http://localhost:8088
    confluent-control-center/control-center-minimal.properties:confluent.controlcenter.ksql.ksqlDB.url=http://192.168.56.117:8088
    confluent-control-center/control-center-dev.properties:#confluent.controlcenter.ksql.ksqlDB.url=http://localhost:8088
    confluent-control-center/control-center-dev.properties:confluent.controlcenter.ksql.ksqlDB.url=http://192.168.56.117:8088
    confluent-control-center/control-center-production.properties:#confluent.controlcenter.ksql.ksqlDB.url=http://ksql:8088
    confluent-control-center/control-center-production.properties:confluent.controlcenter.ksql.ksqlDB.url=http://192.168.56.117:8088
    ksqldb/ksql-server.properties:#listeners=http://0.0.0.0:8088
    ksqldb/ksql-server.properties:listeners=http://192.168.56.117:8088
    ksqldb/ksql-server.properties:# listeners=http://[::]:8088
    ksqldb/ksql-server.properties:# listeners=https://0.0.0.0:8088
    ksqldb/ksql-production-server.properties:#listeners=http://0.0.0.0:8088
    ksqldb/ksql-production-server.properties:listeners=http://192.168.56.117:8088
    ksqldb/ksql-production-server.properties:# listeners=http://[::]:8088
    ksqldb/ksql-production-server.properties:# listeners=https://0.0.0.0:8088
    
    #4，启动服务并查看日志
    [root@c7-docker etc]# confluent local services start
    The local commands are intended for a single-node development environment only,
    NOT for production usage. https://docs.confluent.io/current/cli/index.html
    Using CONFLUENT_CURRENT: /tmp/confluent.007829
    Starting ZooKeeper
    ZooKeeper is [UP]
    Starting Kafka
    Kafka is [UP]
    Starting Schema Registry
    Schema Registry is [UP]
    Starting Kafka REST
    Kafka REST is [UP]
    Starting Connect
    Connect is [UP]
    Starting ksqlDB Server
    ksqlDB Server is [UP]
    Starting Control Center
    Control Center is [UP]
    [root@c7-docker etc]# ls /tmp/confluent.007829
    connect  control-center  kafka  kafka-rest  ksql-server  schema-registry  zookeeper
    #数据文件，日志文件： 
    [root@c7-docker lib]# ls /tmp/confluent.007829/
    connect  control-center  kafka  kafka-rest  ksql-server  schema-registry  zookeeper
    [root@c7-docker lib]# ls /tmp/confluent.007829/connect/
    connect.properties  connect.stdout  data  logs

## b, 页面化操作 (Control Center)：创建topic并生成测试数据 ##

*  访问 http://xxx:9021 进行页面化操作
 *  创建topic: pageviews , users
 *  安装kafka 连接器 (kafka-connect-datagen) , 并生成测试数据  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

页面化创建 datagen source connector 相关参数：

#a, 给 pageviews topic生成测试数据 in AVRO format ： 
        { 
    	  "name": "datagen-pageviews",
    	  "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
    	  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    	  "kafka.topic": "pageviews",
    	  "max.interval": "100",
    	  "quickstart": "pageviews"
    	}
    #b, 给 users topic生成测试数据 in AVRO format ： 
    	{ 
    	  "name": "datagen-users",
    	  "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
    	  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    	  "kafka.topic": "users",
    	  "max.interval": "1000",
    	  "quickstart": "users"
        }

## c, 使用ksqlDB :查看数据/ 创建table或stream ##

*  ksqlDB doc : [https://docs.ksqldb.io/en/latest/concepts/streams/][https_docs.ksqldb.io_en_latest_concepts_streams]
 *  页面化操作ksqlDB 等同于命令行执行： ksql http://localhost:8088  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]

<table> 
 <thead> 
  <tr> 
   <th align="left">ksqldb基本语法</th> 
   <th align="left">sql格式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">创建 stream</td> 
   <td align="left">create stream users( rowkey int key, username varchar) with( kafka_topic=<code>'users'</code>,value_format=<code>'JSON'</code>) ;</td> 
  </tr> 
  <tr> 
   <td align="left">创建 table</td> 
   <td align="left">create table tab_gender_count with (KEY_FORMAT=<code>'JSON'</code>) AS select gender, COUNT(*) AS numusers from users GROUP BY gender EMIT CHANGES limit 3 ;</td> 
  </tr> 
  <tr> 
   <td align="left">读取 stream/table 数据</td> 
   <td align="left">select * from users/tab_gender_count emit changes limit 1 ;</td> 
  </tr> 
 </tbody> 
</table>

######################### 使用ksqlDB创建table/stream : 
    ### ksqlDB 样例1 ( 自动创建 kafka topic)
    # ksql> CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE)
    # WITH (kafka_topic='locations', value_format='json', partitions=1);
    #参数说明：
    # kafka_topic : Name of the Kafka topic underlying the stream. In this case it will be automatically created because it doesn't exist yet, but streams may also be created over topics that already exist.
    # value_format : Encoding of the messages stored in the Kafka topic. For JSON encoding, each row will be stored as a JSON object whose keys/values are column names/values.
    # partitions : Number of partitions to create for the locations topic. Note that this parameter is not needed for topics that already exist.
    
    ### ksqlDB 样例2 (使用已存在的 kafka topic)
    # ksql> create stream users( rowkey int key, username varchar) with( KAFKA_TOPIC='users',VALUE_FORMAT='JSON');
    # Message
    #----------------
    # Stream created
    #----------------
    # ksql> insert into users(username) values('a');
    # ksql> insert into users(username) values('b');
    # ksql> select 'hello,'+ username as greeting from users emit changes;
    #+----------------------------------------------------------------------------------------------------------------------------------------------------------+
    #|GREETING |
    #+----------------------------------------------------------------------------------------------------------------------------------------------------------+
    #|hello,b |
    #|hello,a |
    
    # create a stream for the pageviews topic ： 
    ksql> CREATE STREAM pageviews WITH (KAFKA_TOPIC='pageviews', VALUE_FORMAT='AVRO');
    
    # create a table for the users topic： 
    ksql> CREATE TABLE  users (id VARCHAR PRIMARY KEY) WITH (KAFKA_TOPIC='users', VALUE_FORMAT='AVRO');
    
    ksql> set 'auto.offset.reset'='earliest';
    Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
    ksql>  show topics;
     Kafka Topic                 | Partitions | Partition Replicas
    ---------------------------------------------------------------
     pageviews                   | 1          | 1
     users                       | 1          | 1
    ---------------------------------------------------------------
    
    ######################### 1，非持久化的查询：non-persistent query
    ksql> SELECT * FROM pageviews EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |VIEWTIME                                          |USERID                                            |PAGEID                                            |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |129321                                            |User_1                                            |Page_40                                           |
    Limit Reached
    Query terminated
    
    
    ksql> SELECT * from  users EMIT CHANGES LIMIT 1;
    +-----------------------------+-----------------------------+-----------------------------+-----------------------------+-----------------------------+
    |ID                           |REGISTERTIME                 |USERID                       |REGIONID                     |GENDER                       |
    +-----------------------------+-----------------------------+-----------------------------+-----------------------------+-----------------------------+
    |User_5                       |1489800608800                |User_5                       |Region_5                     |MALE                         |
    Limit Reached
    Query terminated
    
    ######################### 2， 持久化的查询：persistent query (as a stream) : 过滤 pageviews stream 中的女性用户，把查询结果保存到 pageviews_female topic 里面
    ksql> SELECT users.id AS userid, pageid, regionid
      FROM pageviews LEFT JOIN users ON pageviews.userid = users.id
      WHERE gender = 'FEMALE'
      EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |USERID                                            |PAGEID                                            |REGIONID                                          |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |User_3                                            |Page_89                                           |Region_7                                          |
    Limit Reached
    Query terminated
    
    
    ksql> CREATE STREAM pageviews_female AS 
      SELECT users.id AS userid, pageid, regionid
      FROM pageviews LEFT JOIN users ON pageviews.userid = users.id
      WHERE gender = 'FEMALE'
      EMIT CHANGES;
    
    ksql> SELECT * from  pageviews_female EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |USERID                                            |PAGEID                                            |REGIONID                                          |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |User_8                                            |Page_97                                           |Region_4                                          |
    Limit Reached
    Query terminated
    
    ######################### 3，持久化的查询：persistent query : 过滤 regionid 以8或9结尾的，把查询结果保存到 pageviews_enriched_r8_r9 topic 里面
    ksql>  SELECT * FROM pageviews_female
      WHERE regionid LIKE '%_8' OR regionid LIKE '%_9'
      EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |USERID                                            |PAGEID                                            |REGIONID                                          |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |User_9                                            |Page_95                                           |Region_8                                          |
    Limit Reached
    Query terminated
    
    
    ksql> CREATE STREAM pageviews_female_like_89
      WITH (KAFKA_TOPIC='pageviews_enriched_r8_r9', VALUE_FORMAT='AVRO')
      AS SELECT * FROM pageviews_female
      WHERE regionid LIKE '%_8' OR regionid LIKE '%_9'
      EMIT CHANGES;
    
    ksql> SELECT * from  pageviews_female_like_89 EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |USERID                                            |PAGEID                                            |REGIONID                                          |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |User_8                                            |Page_67                                           |Region_8                                          |
    Limit Reached
    Query terminated
    
    ######################### 4， 持久化的查询：persistent query : 统计 pageviews (stream) 里面 每个 REGIONID 和 GENDER ( 30s 为一个窗口 ) , 并且 count >1 , 把结果保存为 table ( topic 为 pageviews_regions ) 
    ksql> SELECT gender, regionid, COUNT(*) AS numusers
      FROM pageviews LEFT JOIN users ON pageviews.userid = users.id
      WINDOW TUMBLING (SIZE 30 SECOND)
      GROUP BY gender, regionid
      HAVING COUNT(*) > 1 EMIT CHANGES LIMIT 1;
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |GENDER                                            |REGIONID                                          |NUMUSERS                                          |
    +--------------------------------------------------+--------------------------------------------------+--------------------------------------------------+
    |FEMALE                                            |Region_6                                          |2                                                 |
    Limit Reached
    Query terminated
    
    
    ksql> CREATE TABLE pageviews_regions WITH (KEY_FORMAT='JSON')
      AS SELECT gender, regionid, COUNT(*) AS numusers
      FROM pageviews LEFT JOIN users ON pageviews.userid = users.id
      WINDOW TUMBLING (SIZE 30 SECOND)
      GROUP BY gender, regionid
      HAVING COUNT(*) > 1
      EMIT CHANGES;
    
    ksql> SELECT * from  pageviews_regions EMIT CHANGES LIMIT 1;
    +-------------------------------------+-------------------------------------+-------------------------------------+-------------------------------------+
    |KSQL_COL_0                           |WINDOWSTART                          |WINDOWEND                            |NUMUSERS                             |
    +-------------------------------------+-------------------------------------+-------------------------------------+-------------------------------------+
    |OTHER|+|Region_3                     |1623913830000                        |1623913860000                        |3                                    |
    Limit Reached
    Query terminated

[https_docs.confluent.io_platform_current_quickstart_ce-quickstart.html]: https://docs.confluent.io/platform/current/quickstart/ce-quickstart.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20221005/5c294cc9b7b74cb6bd52c736ca27a906.png
[https_docs.ksqldb.io_en_latest_concepts_streams]: https://docs.ksqldb.io/en/latest/concepts/streams/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20221005/f39324b33e794cc6b5f43c4cf9059f9e.png