1. kafka

kafka schema

Written by Posted on Less than 0 min read

why schema?

引用Jay kreps的话，关于topic的schema。原文见《Why Avro for Kafka Data?》。

one thing you’ll need to do is pick a data format. The most important thing to do is be consistent across your usage。
  • json虽然表达能力强，但冗余，尤其对于kafka这类数据存储。
  • 引入序列化协议及schema注册中心，可以优化数据存储、支持schema evolve，保证生产消费对同个字段的相同解析，简化etl处理。
  • 但也有折衷，引入注册中心无于多了组件，如果registry宕机，kafka 不可用。所以很多公司不会使用json，而是使用自已约定的消息格式，除非像一些传统公司技术能力欠缺或者kafka消息体量很小还不考虑吞吐和性能问题，当然也有可能公司不差钱。

confluent官方提供了schema registry，用于schema的注册等级，生产消费都会从它那里登记或获取消息序列反序列化schema。

官方支持的schema类型有avro,json,protobuf，推荐avro，原因见上面的链接，摘要如下

We chose Avro as a schema representation language after evaluating all the common options—JSON, XML, Thrift, protocol buffers, etc. We recommend it because it is the best thought-out of these for this purpose. It has a pure JSON representation for readability but also a binary representation for efficient storage. It has an exact compatibility model that enables the kind of compatibility checks described above. It’s data model maps well to Hadoop data formats and Hive as well as to other data systems. It also has bindings to all the common programming languages which makes it convenient to use programmatically.

当然也有成本。

confluent对schema registry的license是部分community license，部分enterprise license，如validation。

具体licence见：https://docs.confluent.io/platform/current/installation/license.html

how or when to use?

Don’t do it.

如果只是做数据探索，可以参考schema registry的源码自定义，行为数据不用过于追求schema。cdc数据可以考虑debezium的schema如何实现。

参考：

https://www.confluent.io/product/confluent-platform/data-compatibility/

https://docs.confluent.io/platform/current/schema-registry/index.html#schemaregistry-intro

https://medium.com/slalom-technology/introduction-to-schema-registry-in-kafka-915ccf06b902

https://cloud.tencent.com/developer/article/1336568

Related articles

  1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（三）验证

前面两篇文章讲解了如何配置认证。本文接着说明如何做client验证ACL是否生效，我们之前开启了无acl信息不允许访问的配置。涉及的client有以下几个场景：shell脚本、python脚本、java应用、flink流。 kafka shell script验证 核心逻辑是，连zk的，使用zk的jaas，连kafka broker的，使用kafka 的jaas配置。 新建topic, kafka-run-class.sh已经配置了zk acl 添加admin shell认证配置文件 vim kafka/config/adminclient-configs.conf 查看所有用户 对应zk目录在config/users下 查看topic 列表，若未指定jaas配置文件，则会hang不响应 列出topic test的acl列表 尝试不指定用户 消费test topic 报错 消费与生产配置，添加用户jaas vim config/consumer.properties vim config/producer.properties 验证消费与生产 python验证 admin用户，验证通过。其他用户需添加用户，并添加acl分别验证 如果认证方式不对，报错信息是kafka.errors.NoBrokersAvailable: NoBrokersAvailable。因为client先获取了kafka version，获取失败就报了。如果用户信息不对，结果是进程hang挂起。 java 如果用户信息不对，会hang挂起 flink scala code 输出 : 若用户信息出错，flink不断重试，报错
Written by
  1. kafka

Another Kafka Monitor – Kowl

kafka监控工具众多，使用过 yahoo的Kafka Manager, 滴滴的kafka manager, kafka eagle, 也造过kafka offset monitor, 甚至有结合kibana与prometheus做的kafka消费监控面板。最近又接触了这个go写的界面清爽的kowl，详见github。 官方支持docker，参考Dockerfile整理一下本地开发环境搭建过程: 1. 后端构建 安装依赖 go mod download 构建 cd backend && go build -o ./bin/kowl ./cmd/api，生成名为kowl二进制可执行文件 2. 前端 安装依赖 npm install 构建 注释package.json中的build script中的ssl option，npm run build 3. 运行 目录结构 其中，kowl.yaml为doc中的config拷贝，调整相关配置 启动 ./kowl -config.filepath ./kowl.yaml {"level":"info","ts":"2022-02-16T16:08:24.297+0800","msg":"started Kowl","version":"","git_sha":"dev","built":""}{"level":"info","ts":"2022-02-16T16:08:24.299+0800","msg":"connecting to Kafka seed brokers, trying to fetch […]
Written by
  1. kafka

Kafka SASL认证授权（五）ACL源码解析

一、ACL检查流程解析 一起看一下kafka server的启动与监听流程： Kafka -> KafkaServer -> SocketServer、KafkaRequestHandler 其中KafkaServer做相关的初始化，包括SocketServer 与 handler pool。 SocketServer的start up流程 看一下acceptor acceptor实际执行流程：监听网络事件，提交给processor处理 看一下processor，截取KafkaServer代码：这边的numIoThreads就是配置的具体处理线程数量。KafkaApis封装具体的各api的处理逻辑，handler pool则是工作线程池。 看看kafkaApis 接着看pool中每个线程的while循环，从request队列拿一个请求，调用apis处理，结果发送给对应channel。 request的入队，见SocketServer，这里也可以看到acl使用的principal和sasl认证是同一个。 之前我们看到，在selector做channel初始化时，已经做了sasl认证。那acl在哪里处理？KafkaApis。以offset commit为例 看一下，这边的authorize，各api共用的方法。构建resource pattern，即资源描述，以及对应的操作。然后使用authorizer实现类进行处理。 看一下鉴权实现类AclAuthorizer，初始化，主要是zk相关的监听，因为token,user,acl都存在zk。 遍历每个action 每个action的检查，区分超管，否则根据acl进行匹配。使用acl 缓存，结合zk的linstener更新缓存。不管是否开启acl，都会去取acl缓存列表。 二、总结 kafka以事件、线程池的方式，对接口请求做处理。 每个api请求，都会做相关的acl校验，不管是否开启acl，都会去取acl缓存列表。差别在于是否使用acl cache map对action进行匹配，匹配本身是用遍历的方式，action不多的话，对吞吐的影响可以忽略不计。 网上有一篇性能测试：见这里
Written by
  1. kafka
  2. 工具

本地搭建源码阅读开发构建环境示例: kafka

大家对开源项目有兴趣、想成为committer，或者工作需要，会从github上获取最新的开源项目源码。本文做一个示例，怎样搭建本地的源码阅读、开发、构建环境。 首先，在github上找到项目的链接，clone到本地自己的目录中，这步略过。 下面一起看一个示例：使用idea导入kafka 2.4.1版本的源码。 一、idea打开项目后，切换到指定的分支 获取最新分支列表（如果卡很久，shell加一下自己的网络代理，翻墙一下）： git fetch –all –tags 获取2.4前缀分支列表: -n 加描述 -l <pattern> 加正则 从git tag创建分支，切换为指定的版本：git checkout tags/<tag> -b <branch> 二、下载依赖 gradle build项目 报错1：本地默认gradle版本过高，没有MavenDeployment这个插件。 为了解决gradle版本问题，项目使用gradle wrapper版本配置。调整idea使用项目的gradle，preference->Build,->Gradle->Use Gradle from，选择gradle/wrapper/gradle-wrapper.properties，重新reload。或者直接执行项目根目录的gradlew，会自动下载指定版本编译。 三、本地开发 熟悉代码内容，方式推荐两种： 1、结合文档看 2、结合jira看，熟悉kafka的代码模块、开发、提交pr的流程 kafka2.8之前需要zk，所以先把本地zk起起来，入口类：core模块的kafka.Kafka 运行前读一下kafka-server-start.sh，知道它的log配置是通过命令参数的方式： 所以，运行时带上参数，如下图： 此外，依然报错： 排查gradle脚本，compile阶段没有依赖slf4j实现库，而是在发布时从test compile拷贝这个依赖，所以可以加一下依赖到core项目的依赖配置中： Tips: 本地一般会有各版本的kafka，这边是开发调试用的，改一下zk地址，添加一下chroot zookeeper.connect=localhost:2181/kafka_dev 四、构建 使用命令行（README.md中有说明），或者直接在idea gradle窗口双击执行相关task。 查看输出的release包 五、延伸 kafka code指引：https://kafka.apache.org/coding-guide.html kafka bug jira：jira kafka contribute：https://kafka.apache.org/contributing […]
Written by
  1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（二）ACL

由上一篇博文，讲解了SASL认证的安装过程。本文主要围绕ACL设置进行说明。 参考： kafka authentorization：官方 一、开启ZK ACL（可选，内网环境，用户无机器访问权限时） 给kafka meta都加上zk的acl，默认是world，即任何人可访问。 zk acl详见：https://www.cnblogs.com/dalianpai/p/12748144.html 执行已有kafka meta的迁移 vim kafka/bin/zookeeper-security-migration.sh 执行迁移 到zk上查看生效 二、设置kafka acl 上篇文章已经在server.properties中添加了acl的配置。这里执行命令设置acl。 kafka的acl对象，即资源类型分为 cluster, topic, consumer group, transation id, token。实际应用中，主要以topic及consumer group为主进行运维管控。即控制”哪个用户的哪个消费组可以访问哪个topic“。 常用acl操作 列出 test 这个 topic 的 ACL 列表 对用户 test 授权，以访问test 这个 topic。 先添加scram用户 test 因为添加了zk的acl，所以需指定ZK的acl jaas账号信息。若不指定ZK的jaas账号信息，会报如下错误： 加acl认证信息的方式：先调整bin/kafka-run-class.sh，用于zk acl验证。这个文件被大多数的kafka bin目录下的script所引用，如添加topic的脚本等，所以改它最方便，添加一行。 再执行一次添加即可，输出如下提示表示添加用户test完成。 添加consumer 授权方式1，先加topic权限： 输出： 再添加消费组权限： 添加consumer 授权方法2，参考文档。一步到位： […]
Written by
  1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（一）认证

kafka安全涉及3部份：传输加密，用户认证与授权，ZK开启ACL（Zookeeper存储了kafka的元数据以及用户信息，默认不开启acl所有用户可改，内网环境机器不对外开放可考虑使用默认不开启ZK ACL）。 词汇说明： 认证，即用户登陆。 授权，即管理用户可见的资源。 ACL，Access Control List 访问控制列表 SASL认证与Kerberos认证：SASL资料很多, java的见这里，Kerberos的资料，点这里。 Kafka权限控制指引 支持的认证方式 Kafka支持的认证类别有kerberos（和hadoop一样，大多数公司应该没用kerberos）、ldap（在传统企业中比较普遍）与rbac（这两个是要企业license，基于Confluence的平台组件MDS，自建集群如果没用kafka connect之类的组件，没用conflunece的cli，也用不了）、sasl/plain（用户信息用文件进行管理，修改需重启Kafka，生产大概率不会接受并使用）和sasl/scram（用户信息用api或命令行进行管理，存储在zk上，不需重启）。考虑用哪一种取决于公司自身的情况，这里主要是用sasl/scram。 接入现有认证系统 如果对zk不放心或想对接已有的认证系统，接入kafka权限管控，可参考这个kip-86自定义认证方式，文中有场景列表及sample code。 开启认证主要的关注点 使用者的需求，生产者和消费者都有哪些，场景如何，需求怎样，支持以后需要先验证，对使用方提供技术上的指引文档等。 开启方式，是重启集群还是迁移集群。开启认证不是平滑的，需要短暂中断业务流程。 使用的认证方式：基于现有的环境及实际需要。 运维成本：考虑运维、监控的方式及成本（监控需要admin账号） kafka版本是否需要升级：公司现有的版本是0.10.1，1.1，2.4等，需要升级吗？看使用哪种认证方式,rbac需要2.4+，sasl/scram只要0.9+，本文描述的基于2.4.1版本。 本文主要描述用户认证： 参考： sasl/scram 官方文档 sasl/plain 方式 一、zookeeper sasl开启（可选，内网环境，用户无机器访问权限时） vim zookeeper/conf/zoo.cfg vim zookeeper/conf/zookeeper_jaas.conf Server是ZK SERVER之间， Client是zkclient与zk server之间 vim zookeeper/bin/zkEnv.sh zk启动引入jaas配置 vim zookeeper/conf/adminclient_jaas.conf vim zookeeper/bin/zkCli.sh zkCli命令默认添加jaas 重启zk bin/zkServer.sh restart 验证 bin/zkCli.sh 是否成功，调整密码验证 二、配置kafka […]
Written by
  1. kafka

Kafka序列化反序列化解析

kafka有自己的rpc协议，即nio bytebuf中的数据格式，详见之前的kafka相关介绍的文章。这里我们来看一下大家常用，有时又疑惑的序列化反序列化，对应rpc协议中的records，kafka叫Serdes，实际上也是字面上的意思serialize and deserialize。在程序中序列化就是toBinary，反序列化是fromBinary，存储和传输都是用二进制。 通过以上描述，我们知道序列化和反序列化的作用。我们来看kafka的实现。Serdes工厂类，和提供基本数据类型序列化。 一、先看基础的抽象 interface：Serializer 包括 configure，例如配置ChartSet字符集，serialize，序列化。 interface：Deserializer 同理。 再来看包装接口：Serde 指定了serializer和deserializer。 二、 实现类，以Serializer为例 提供了几大应用的实现，分别是common，connect，stream，也即对应了kafka一般场景、kafka抽数场景如debezium，kafka stream原生流计算。我们看common，主要是实现了基本的数据类型的序列化，举两个例子，Long和String。 从上我们知道实现Serialize类，主要做了toBinary动作。 三、包装类，工厂方法 这里看WrapperSerde 为静态内部类，有一批子类： 看其中一个常用的StringSerde 引入String序列化类和反序列化类。
Written by
  1. kafka

Kafka SASL认证授权（四）认证源码解析

一、认证流程 在了解kafka网络模型的基础上，了解它的认证流程： ApiVersionsRequest->SaslHandshakeRequest->a series of SASL client and server tokens corresponding to the mechanism are sent->认证成功，继续处理后续的请求，否则关闭连接。 状态转换见SaslServerAuthenticator https://kafka.apache.org/protocol#sasl_handshake 具体逻辑chanel验证流程以sasl/plain为例： 分析这个包的内容： server factory创建sasl plain server单例。server负责按协议获取username 与 password，再最终调用plain server callback handler进行验证（handler的指定是在SaslChannelBuilder中根据具体配的协议类型进行指定）。 下面看一下关键类SaslChannelBuilder，创建sasl channel的实现： 初始化时，配置handler： 再看具体的channel，结合selector poll的过程KafkaChannel： java doc有说明具体的网络模型： 我们来看prepare方法： prepare触发的地方，调用频率在selector，事件通知触发： 也就是每次网络读事件且channle尚未验证成功都会触发认证，和官方文档描述一致。一旦验证通过就不再认证。所以开启认证对吞吐影响可以忽略不计。 校验不过，会直接抛异常捕获后，断开连接了。 二、总结 以上我们了解到kafka sasl的认证流程，现在，可以自己添加认证方式，例如加上自己公司平台的权限认证，注意尽量使用轻量级的方式即cache。 scram方式，除了以上流程，还会多一步，见ScramSaslServer。第一步获取credential，无效则报错，第二步计算校验。详见scram rfc定义：这里 kafka server这边的final message 截图： 自定义认证可以参考这篇：博文 三、token验证 kafka实现了基于token验证的sasl。详见这里：文档 流程是：admin client管理client（创建和刷新），存放在zk上，验证时会验token是否存在，以及用户信息，密码是否一致。token的生命周期内有效，过期无效。 弊端：在公司内一般不会频繁用到这种方式， […]
Written by
Next
No Comments
Leave a comment Cancel
Comments to: kafka schema

Your email address will not be published. Required fields are marked *