Kafka全面深度指南：从基础架构到高级特性
引言Apache Kafka 是一款开源的分布式事件流平台，被广泛应用于高性能数据管道、流分析、数据集成和关键任务应用。它以其高吞吐量、可扩展性和持久性著称，成为现代数据架构中不可或缺的一部分。本文将从 Kafka 的基础概念、核心架构、生产者、消费者、Broker机制，一直到高级特性如事务、幂等性、文件存储和性能优化，提供一个全面而深入的指南。 Kafka概述定义 Kafka传统定义：Kafka是一个分布式的基于发布/订阅模式的消息队列（MessageQueue），主要应用于大数据实时处理领域。 发布/订阅：消息的发布者不会将消息直接发送给特定的订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息。 Kafka最新定义：Kafka是一个开源的分布式事件流平台（Event StreamingPlatform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。 消息队列目前企业中比较常见的消息队列产品主要有 Kafka、ActiveMQ 、RabbitMQ 、RocketMQ 等。 在大数据场景主要采用 Kafka 作为消息队列...
记录 Flink 结合 Docker 部署配置
首先安装flink,这里以1.13.1举例(docker安装的话,jobmanager和taskmanager需要分开部署)flink-docker-github-repository (这个参数flink文档中没找到)关于query.server.port: 6125不用去管,这个不填也会默认自动生成: yml中的内容不说了, 下面贴几张关于 Flink 生产配置最佳实践的图，钉钉公开课趣头条实时平台负责人分享：https://www.bilibili.com/video/BV1iE411r7S6 Flink在yarn上运行，每个TaskManager的slot个数怎么设置？经验公式：slot个数tm个数=并行度并行度=kafka的分区个数（10分区）slot的个数要小于yarn设置的单个container最大可以申请的cpu核数（5个 8-36个）。那么就是 5个slot2个tm=并行度=kafka分区数多个小集群好还是一个大集群好？1.小集群的话任务分布在不同的集群，任务之间的影响小。不会因为一个任务占用...
记录开发中遇到关于 Flink 的一些错误（长期）
2022-01-20 09:25:58,308 WARN org.apache.flink.client.deployment.application.DetachedApplicationRunner [] - Could not execute application:org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Job was submitted in detached mode. Results of job execution, such as accumulators, runtime, etc. are not available. Please make sure your program doesn't call an eager execution function [collect, print, printToErr, count] 这是因为当 flink 作业是以 detached 模式提交的时候，...
Kettle 记录集合里面的记录数说明
记录集合里面的记录数 该数量是指组件与组件之间通信的缓存队列的 size 大小，Kettle 内部用 List 实现该缓存队列，每一条语句都会被封装成一个 RowSet 对象，每个组件之间都会有个 List<RowSet> 队列，源 step 每次会往该队列写一条数据，目标 step 每次会从队列读取一条数据。 Kettle 里面转换是并行的，数据是一条一条流经每个组件，队列大小采用默认的 10000 条即可。 但是当使用到「阻塞数据直至步骤完成」、「阻塞数据」相关组件时，则需要根据业务数据量扩大该 size 的大小。 否则，数据会一直在某两个组件之间的缓存队列中存放，当数量达到一定大小时，就会卡住不动。 缓存队列的大小对作业没有影响，查看源码发现，缓存队列由 inputRowSet 和 outputRowset 组成，底层实现是 ArrayList，初始化创建的时候，默认容量为 10，并不会直接创建设定长度的数组。并且使用 ArrayList 的特性：当要添加的数据量超过数组的容量时候，ArrayList 会动态扩容，size 扩大 1.5 倍。
Kettle 安装及支持 ClickHouse 开发配置
Kettle 下载地址 选择需要的版本，进入这个路径，下载这个 CE 解压 Kettle 安装包下载 ClickHouse 所需驱动包 将 clickhouse-plugins 文件夹复制到 Kettle 的 data-integration\plugins 文件夹里 复制驱动包里的其余 jar 包，粘贴到 Kettle 的 data-integration\libswt\xxx 目录下根据自己的系统而定，Linux 环境使用 uname -r 先看下系统。 启动 Kettle 编辑工具双击 data-integration 目录下 Spoon.bat 启动。
Kettle ES-Bulk-Insert 组件二开记录（基于 Kettle 9.2）
TODO: 待改造，目前这种可以配置变量的地方却无法配置变量（不识别，只能写死），之后尝试改下。 背景目前 Kettle 最新的版本是 9.2，但是 ES 仅支持到 6.4.2，而且也无法配置用户名和密码还有 schema，所以需要对 es-bulk-insert 组件改造下。 下载源码Kettle GitHub 源码地址 注意：有些包可能国内网络下载不下来，而且国内镜像也没有，就需要单独去想办法了，有人需要的话我会把包发出来。 主要改两个类 ElasticSearchBulk ElasticSearchBulkDialog 前者是业务数据执行需要的，后者是那个组件框需要的，比如你点 Test，就会调用这个类的 test 方法。 首先需要改 pom 文件，改成需要的版本 其次是这些在高版本中弃用的类和方法 在 Elastic 官网中也给出了对应的替代方法：官方迁移指南 ElasticSearchBulk 类的修改如下1234567891011121314151617181920212223242526272829303132333435363738394...
Flink ES Connector 7 - ElasticsearchSink
截至1.13.1,官方文档所提供的方式已经废弃12345678910111213141516171819202122232425262728293031HttpHost httpHost = new HttpHost(esHost, esPort, esScheme);List<HttpHost> httpPosts = new ArrayList<>();httpPosts.add(httpHost);RestClientFactory restClientFactory = new RestClientFactory() { @Override public void configureRestClientBuilder(RestClientBuilder restClientBuilder) { CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCr...
Job 和 Trans 的参数详解
pan 调用示例1sh pan.sh -rep=initech_pdi_repo -user=pgibbons -pass=lumburghsux -trans=TPS_reports_2011 -param:db_ip=$db_ip -param:db_name=$db_name 参数 参数说明 rep 企业或数据库存储库名称 user 仓库用户名 pass 仓库密码 trans 要启动的转换的名称（在存储库中显示） dir 包含转换的存储库目录，包括前导斜杠 file 如果调用的是本地 KTR 文件，则为文件名，如果不在本地目录中，则包括路径 level 日志级别（Basic, Detailed, Debug, Rowlevel, Error, Nothing） logfile 将日志输出写入的本地文件名 listdir 列出指定存储库中的目录 listtrans 列出指定存储库目录中的转换 listrep 列出可用的存储库 exprep 将所有存储库对象导出到一个 XML 文件 norep 防止 Pan 登录到存...
Kettle 是什么
Kettle 中文网 免费开源的基于 Java 的企业级 ETL 工具ETL（Extract-Transform-Load 的缩写，即数据抽取、转换、装载的过程），对于企业或行业应用来说，我们经常会遇到各种数据的处理、转换、迁移，所以了解并掌握一种 ETL 工具的使用必不可少。 这里我介绍一个在工作中使用了 3 年左右的 ETL 工具 Kettle，本着好东西不独享的想法，跟大家分享碰撞交流一下！在使用中我感觉这个工具真的很强大，支持图形化的 GUI 设计界面，然后可以以工作流的形式流转，在做一些简单或复杂的数据抽取、质量检测、数据清洗、数据转换、数据过滤等方面有着比较稳定的表现。其中最主要的是，我们通过熟练地应用它，减少了非常多的研发工作量，提高了工作效率。
Kettle 开发中需要注意的问题点
注意点 过滤记录或者字段选择等组件，无法出现上面步骤中的字段的时候，尝试在上面步骤中再次点击确定。 两个数据流 join 前必须先进行排序操作。 数据流从计算器或 join 操作流出，并将流入另一 join 操作之前，需要插入一个字段选择插件。 表输入插件中，填写查询语句时，语句末尾不能加分号。Kettle 8.2 以上无此问题 SQL 脚本插件中，填写 SQL 语句时，语句末尾必须加分号，否则不会执行。Kettle 8.2 以上无此问题 引用变量时，采用 ${变量名} 的方式引用，以免出现问题。此种引用方式还可以在变量内容后接其他字符串，例如：${key}_20151217 会被解析为：value_20151217。 Job 中执行多个 trans 时，如何确定 trans 执行的先后顺序？根据连接线的先后顺序执行，如果需要优先执行则先连接从开始到该 trans 的连线即可。 如果源数据中有空值，但是目标数据中不希望出现空值： 直接设置目标数据库各列的默认值为空字符串是不行的，因为只有不对某列赋值才会填入默认值，而 NUL...
公告