Wormhole的核心概念

　　　　　　　　　　　　Wormhole的核心概念

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正杰

版权声明：原创作品，谢绝转载！否则将追究法律责任。

一.Wormhole概述

1>.什么是Wormhole

　　Wormhole 面向大数据流式处理项目的开发管理运维人员，致力于提供统一抽象的概念体系，直观可视化的操作界面，简单流畅的配置管理流程，基于 SQL 即可完成的业务逻辑开发方式，并且屏蔽了流式处理的底层技术细节，极大的降低了数据项目管理运维门槛，使得大数据流式处理项目的开发管理运维变得更加轻量敏捷可控可靠。

　　博主推荐阅读:
　　　　https://edp963.github.io/wormhole
　　　　https://edp963.github.io/wormhole/concept.html

2>.设计理念

　　1>.统一DAG高阶分形抽象
　　　　(1)构建由 Source DataSys，Kafka Topic，Spark Stream（Flink Stream），Sink DataSys 组成的物理 DAG
　　　　(2)每个物理 DAG 里可以并行处理多个由 Source Namespace，Flow，Sink Namespace 组成的逻辑 DAG
　　　　(3)每个Flow本身是典型的 Spark RDD DAG
　　2>.统一通用流消息UMS协议抽象
　　　　(1)UMS是Wormhole 定义的流消息协议规范
　　　　(2)UMS试图抽象统一所有结构化消息
　　　　(3)UMS自身携带结构化数据Schema信息
　　　　(4)Wh4支持用户自定义半结构化JSON格式
　　3>.统一数据逻辑表命名空间Namespace抽象
　　　　Namespace是Wormhole定义的唯一定位数据系统上数据表的规范，由七部分组成。例如："mysql.test.social.user.*.*.* kafka.test.social.user.*.*.*"

　　　　Namespace的语法格式为:"[Data System].[Instance].[Database].[Table].[Table Version].[Database Partition].[Table Partition]",各名词解释说明如下所示:
　　　　　　Data System:
　　　　　　　　代表数据存储系统类型，如 Oracle，Mysql，Hbase，Elasticsearch，Kafka 等
　　　　　　Instance:
　　　　　　　　代表数据存储系统物理地址别名
　　　　　　Database:
　　　　　　　　代表RDBS中的数据库，Hbase中的命名空间，Elasticsearch中的索引，Kafka中的主题
　　　　　　Table:
　　　　　　　　代表RDBS中的数据表，Hbase中的数据表，Elasticsearch中的文档，Kafka中的某主题下的数据
　　　　　　Table Version:
　　　　　　　　代表Table的表结构版本，一般情况下Table Version的值应随表结构的变化递增(目前Wormhole 中处理时用"*"匹配所有版本的数据)
　　　　　　Database Partition:
　　　　　　　　代表Databas 的分区名称(目前Wormhole中处理时用"*"匹配所有分区的数据)
　　　　　　Table Partition:
　　　　　　　　代表Table的分表名称(目前Wormhole中处理时用"*"匹配所有分表的数据)

3>.主要特性

　　支持可视化，配置化，SQL 化开发实施流式项目

　　支持指令式动态流式处理的管理，运维，诊断和监控

　　支持统一结构化 UMS 消息和自定义半结构化 JSON 消息

　　支持处理增删改三态事件消息流

　　支持单个物理流同时并行处理多个逻辑业务流

　　支持流上 Lookup Anywhere，Pushdown Anywhere

　　支持基于业务策略的事件时间戳流式处理

　　支持 UDF 的注册管理和动态加载

　　支持多目标数据系统的并发幂等入库

　　支持多级基于增量消息的数据质量管理

　　支持基于增量消息的流式处理和批量处理

　　支持 Lambda 架构和 Kappa 架构

　　支持与三方系统无缝集成，可作为三方系统的流控引擎

　　支持私有云部署，安全权限管控和多租户资源管理

二.UMS（统一消息规范）

　　UMS是Wormhole定义的消息规范(JSON 格式)，UMS试图抽象统一所有结构化消息，通过自身携带的结构化数据Schema信息，来实现一个物理DAG同时处理多个逻辑DAG的能力，也避免了和外部数据系统同步元数据的操作。

　　UMS由protocol,schema和payload三部分组成,各名词解释如下:
　　　　protocol(代表消息协议)，常见的协议如下所示:
　　　　　　data_increment_data:
　　　　　　　　代表增量数据
　　　　　　data_initial_data:
　　　　　　　　代表全量数据
　　　　　　data_increment_heartbeat:
　　　　　　　　代表增量心跳数据
　　　　　　data_increment_termination:
　　　　　　　　代表增量数据结束

　　　　schema(代表消息来源及表结构信息)
　　　　　　namespace 指定消息的来源，fields 指定表结构信息，其中 ums_id_,ums_ts_,ums_op_ 三个字段是必需的三个系统字段，Wormhole 使用这三个字段实现幂等写数据的逻辑。
　　　　　　ums_id_: 
　　　　　　　　long 类型，用来唯一标识消息，须根据消息生成的顺序递增
　　　　　　ums_ts_: 
　　　　　　　　datetime 类型，每一条消息产生的时间
　　　　　　ums_op_: 
　　　　　　　　string 类型，指定每条消息生成的方式，值为"i"或"u"或"d"，分别代表新增，更新，删除操作

　　　　payload(代表数据本身)
　　　　　　tuple：
　　　　　　　　一个tuple对应一条消息。

  {
  "protocol": {
    "type": "data_increment_data"          
  },
  "schema": {
    "namespace": "kafka.kafka01.datatopic.user.*.*.*",
    "fields": [
      {
        "name": "ums_id_",
        "type": "long",
        "nullable": false
      },
      {
        "name": "ums_ts_",
        "type": "datetime",
        "nullable": false
      },
      {
        "name": "ums_op_",
        "type": "string",
        "nullable": false
      },
      {
        "name": "key",
        "type": "int",
        "nullable": false
      },
      {
        "name": "value1",
        "type": "string",
        "nullable": true
      },
      {
        "name": "value2",
        "type": "long",
        "nullable": false
      }
    ]
  },
  "payload": [
    {
      "tuple": [ "1", "2016-04-11 12:23:34.345123", "i", "23", "aa", "45888" ]
    },
    {
      "tuple": [ "2", "2016-04-11 15:23:34.345123", "u", "33", null, "43222" ]
    },
    {
      "tuple": [ "3", "2016-04-11 16:23:34.345123", "d", "53", "cc", "73897" ]
    }
  ]
}

三.UMS_Extension (UMS 扩展格式)

　　除UMS格式外，Wormhole 支持 UMS_Extension（UMS扩展格式），用户可自定义数据格式，且支持嵌套结构。使用时须将 Kafka 消息的 key 设置为 data_increment_data.sourceNamespace，然后在 Wormhole 页面上粘贴数据样式简单配置即可。如 sourceNamespace 为 kafka.kafka01.datatopic.user.*.*.*，则 Kafka 消息的 key 须为 data_increment_data.kafka.kafka01.datatopic.user.*.*.*。

　　若一个 sourceNamespace 的消息需要随机分配到多个 partition，消息的 key 可设置为data_increment_data.kafka.kafka01.datatopic.user.*.*.*...abc 或 data_increment_data.kafka.kafka01.datatopic.user.*.*.*...bcd，即在 sourceNamespace 后面添加“…”，之后可添加随机数或任意字符。

　　若 UMS_Extension 类型数据有增删改操作且需要幂等写入，也须配置 ums_id_,ums_ts_,ums_op_ 三个字段。

四.Stream

　　Stream是在Spark Streaming上封装的一层计算框架，消费的数据源是Kafka。

　　Stream作为Wormhole的流式计算引擎，匹配消息的key，sourceNamespace和其对应处理逻辑，可将数据以幂等的方式写入多种数据系统中。处理过程中Stream会反馈错误信息、心跳信息、处理数据量及延时等信息。

　　一个Stream可以处理多个Namespace及其处理逻辑，共享计算资源。

五.Flow

　　Flow关注的是数据从哪来(sourceNamespace)，到哪去(sinkNamespace)，及中间的处理逻辑。

　　Flow支持SQL配置，自定义UDF，自定义Class，且可以关联其他RDBS/Hbase/Phoenix/Redis/Es等系统中的数据。

　　Flow配置好后可以注册到Stream，Stream接收Flow指令后，根据指令中的sourceNamespace，sinkNamespace及业务逻辑处理数据。

六.Job

　　Job相当于Spark Job，其数据源是HdfsLog Stream备份在Hdfs上的数据。Stream/Flow/Job组合可实现Lambda架构和Kappa架构。

　　Kafka中数据有一定的生命周期，可通过Stream将Kafka中数据备份到Hdfs上。后续需要从某个时间节点重新计算或者补充某个时间段的数据，可通过Job读取Hdfs上的备份数据，配置与Flow相同的处理逻辑，将数据写入目标表。

　　温馨提示：
　　　　目前UMS_Extension类型数据只支持通过Stream将Kafak中数据备份到Hdfs上，Job还不支持读取UMS_Extension类型数据。

七.博主推荐阅读

　　编译Wormhole实战篇:
　　　　https://www.cnblogs.com/yinzhengjie2020/p/12969557.html

　　Wormhole部署实战案例:
　　　　https://www.cnblogs.com/yinzhengjie2020/p/12977470.html