0. 源码地址
https://github.com/wiselyman/kafka-in-battle
1. Operator Framework
Operator Framework是一个用来管理k8s原生应用(Operator)的开源工具。
Operator Framework支持的Operator分享地址:https://operatorhub.io。
如安装Kafka使用Strimzi Apache Kafka Operator,地址为:https://operatorhub.io/operator/strimzi-kafka-operator 。
打开Strimzi Apache Kafka Operator页面,右侧有install按钮,按照页面提示进行Operator安装。
2. 安装Operator Lifecycle Manager
Operator Lifecycle Manager是Operator Framework的一部分,OLM扩展了k8s提供声明式方法安装、管理、更新Operator以及他们的依赖。
点击页面上的install显示如何安装Strimzi Apache Kafka Operator,我们首先第一步要安装Operator Lifecycle Manager(不要执行下句命令):
curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/0.12.0/install.sh | bash -s 0.12.0
该命令需要使用quay.io的镜像,我们需采取从源码安装,并修改源码中的镜像地址加速。
源码地址:https://github.com/operator-framework/operator-lifecycle-manager/releases,当前最新版本为0.12.0
。
-
下载:https://github.com/operator-framework/operator-lifecycle-manager/releases/download/0.12.0/crds.yaml
-
下载:https://github.com/operator-framework/operator-lifecycle-manager/releases/download/0.12.0/olm.yaml
将olm.yml
中:
quay.io -> quay.azk8s.cn
执行安装:
kubectl apply -f crds.yaml
kubectl apply -f olm.yaml
3. 安装Strimzi Apache Kafka Operator
kubectl create -f https://operatorhub.io/install/strimzi-kafka-operator.yaml
使用下面命令观察Operator启动情况
kubectl get csv -n operators
显示如下则安装成功
wangyunfeis-MacBook-Pro:olm wangyunfei$ kubectl get csv -n operators
NAME DISPLAY VERSION REPLACES PHASE
strimzi-cluster-operator.v0.14.0 Strimzi Apache Kafka Operator 0.14.0 strimzi-cluster-operator.v0.13.0 Succeeded
4. 安装Kafka集群
下载https://raw.githubusercontent.com/strimzi/strimzi-kafka-operator/0.14.0/examples/kafka/kafka-persistent.yaml,主要修改的是所需存储空间为5Gi作为测试条件,这里的存储需要K8s集群中有默认的StorageClass
。
apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
name: my-cluster
spec:
kafka:
version: 2.3.0
replicas: 3
listeners:
plain: {}
tls: {}
config:
offsets.topic.replication.factor: 3
transaction.state.log.replication.factor: 3
delete.topic.enable: "true"
transaction.state.log.min.isr: 2
log.message.format.version: "2.3"
storage:
type: jbod
volumes:
- id: 0
type: persistent-claim
size: 5Gi
deleteClaim: false
zookeeper:
replicas: 3
storage:
type: persistent-claim
size: 5Gi
deleteClaim: false
entityOperator:
topicOperator: {}
userOperator: {}
kubectl apply -f kafka-persistent.yml -n kafka
- 发送消息测试
kubectl exec -i -n kafka my-cluster-kafka-0 -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic strimizi-my-topic
- 接受消息测试
kubectl exec -i -n kafka my-cluster-kafka-0 -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic strimizi-my-topic --from-beginning
- 显示集群Topic
kubectl exec -n kafka my-cluster-kafka-0 -- bin/kafka-topics.sh --list --zookeeper localhost:2181
5. Kafka Connect
本节将外部的SQL Server中的表person(字段只有id
和name
)通过Kafka Connect同步至K8s集群里的PostgreSQL中。
5.1 开启SQL Server数据库的CDC(Change Data Capture)功能
5.1.1 启用数据库CDC
USE bs_portal
EXEC sys.sp_cdc_enable_db;
bs_portal
为数据库名,此时会自动给我们创建cdc的schema和相关表:
captured_columns
change_tables
dbo_person_CT
ddl_history
index_columns
lsn_time_mapping
可使用下面sql语句查询已开启CDC的数据库:
select * from sys.databases where is_cdc_enabled = 1
5.1.2 启用表的CDC
USE bs_portal
EXEC sys.sp_cdc_enable_table
@source_schema = 'dbo',
@source_name = 'person',
@role_name = 'cdc_admin',
@supports_net_changes = 1;
@source_name
为表名,查询表开启CDS的sql语句:
select name, is_tracked_by_cdc from sys.tables where object_id = OBJECT_ID('dbo.person')
查看新增的job
SELECT job_id,name,enabled,date_created,date_modified FROM msdb.dbo.sysjobs ORDER BY date_created
确定用户有权限访问CDC表
EXEC sys.sp_cdc_help_change_data_capture;
5.1.3 开启“SQL Server 代理”
检查安装了SQL Server的操作系统中“服务”中是否开启了“SQL Server 代理”。
5.1.4 关闭CDC
关闭数据库的CDC
USE bs_portal
EXEC sys.sp_cdc_disable_db;
关闭表的CDC
USE bs_portal
EXEC sys.sp_cdc_disable_table
@source_schema = 'dbo',
@source_name = 'person',
@capture_instance = 'all';
5.2 SQL Server To PosgreSQL
5.2.1 准备Kafka Connect镜像
输入插件(source):下载SQL Server Connector plugin:http://central.maven.org/maven2/io/debezium/debezium-connector-sqlserver/;输出插件(sink):下载Kafka Connect JDBC:https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc。
新建Dockerfile文件,将debezium-connector-sqlserver-0.10.0.Final-plugin.zip
解压放置到Dockerfile相同目录下的plugins
目录;在plugins
目录下新建目录kafka-connect-jdbc
,解压confluentinc-kafka-connect-jdbc-5.3.1.zip
,将lib
下的kafka-connect-jdbc-5.3.1.jar
和postgresql-9.4.1212.jar
放置在kafka-connect-jdbc
。
编写Dockerfile
FROM strimzi/kafka:0.14.0-kafka-2.3.0
USER root:root
COPY ./plugins/ /opt/kafka/plugins/
USER 1001
MAINTAINER 285414629@qq.com
使用阿里云“容器镜像服务”(https://cr.console.aliyun.com/)编译镜像,目前我们的源码地址位于:https://github.com/wiselyman/kafka-in-battle。
-
“镜像仓库”->“创建镜像仓库”:
-
仓库名称:kafka-connect-form-sql-to-jdbc
-
仓库类型:公开
-
-
下一步后,选择“Github”标签页,使用自己的GitHub库,“构建设置”只勾选“海外机器构建”,然后点击“创建镜像仓库”。
-
点击镜像仓库列表中的“kafka-connect-mysql-postgres”->“构建”->“添加规则”:
-
类型:Branch
-
Branch/Tag:master
-
Dockerfile目录:/sqlserver-to-jdbc/
-
Dockfile文件名:Dockerfile
-
镜像版本:0.1
-
-
确认后,“构建规则设置”->“立即构建”,“构建日志”显示“构建状态”为“成功”即可。
5.2.2 安装Kafka Connect
编写Kafka Connect集群部署文件kafka-connect-sql-postgres.yml
:
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaConnect
metadata:
name: my-connect-cluster
spec:
version: 2.3.0
replicas: 1
bootstrapServers: 'my-cluster-kafka-bootstrap:9093'
image: registry.cn-hangzhou.aliyuncs.com/wiselyman/kafka-connect-from-sql-to-jdbc:0.1
tls:
trustedCertificates:
- secretName: my-cluster-cluster-ca-cert
certificate: ca.crt
执行安装
kubectl apply -f kafka-connect-sql-postgres.yml -n kafka
查询已安装的插件
kubectl exec -i -n kafka my-cluster-kafka-0 -- curl -X GET http://my-connect-cluster-connect-api:8083/connector-plugins
结果如:
[{
"class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"type": "sink",
"version": "5.3.1"
}, {
"class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"type": "source",
"version": "5.3.1"
}, {
"class": "io.debezium.connector.sqlserver.SqlServerConnector",
"type": "source",
"version": "0.10.0.Final"
}, {
"class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
"type": "sink",
"version": "2.3.0"
}, {
"class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
"type": "source",
"version": "2.3.0"
}]
5.2.3 使用Helm安装PostgreSQL
使用helm安装PostgreSQL,这里的PostgreSQL库来自于https://kubernetes.oss-cn-hangzhou.aliyuncs.com/charts/,可在Helm中配置。
对PostgreSQL的账号、密码、初始化数据库、服务类型进行定制后安装:
helm install --name my-pg --set global.storageClass=standard,postgresUser=wisely,postgresPassword=zzzzzz,postgresDatabase=center,service.type=NodePort,service.nodePort=5432 stable/postgresql
5.2.4 Kafka Connect Source配置
编写source配置:sql-server-source.json
{
"name": "sql-server-connector",
"config": {
"connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
"tasks.max" : "1",
"database.server.name" : "exam",
"database.hostname" : "172.16.8.221",
"database.port" : "1433",
"database.user" : "sa",
"database.password" : "sa",
"database.dbname" : "bs_portal",
"database.history.kafka.bootstrap.servers" : "my-cluster-kafka-bootstrap:9092",
"database.history.kafka.topic": "schema-changes.person",
"table.whitelist": "dbo.person"
}
}
编写sink配置:postgres-sink.json
{
"name": "postgres-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "1",
"topics": "exam.dbo.MH_YCZM",
"connection.url": "jdbc:postgresql://my-pg-postgresql.default.svc.cluster.local:5432/center?user=wisely&password=zzzzzz",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"auto.create": "true",
"insert.mode": "upsert",
"delete.enabled": "true",
"pk.fields": "IPDZ",
"pk.mode": "record_key"
}
}
5.2.5 使用
将配置文件提交到Kafka Connect
cat sql-server-source.json | kubectl exec -i -n kafka my-cluster-kafka-0 -- curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://my-connect-cluster-connect-api:8083/connectors -d @-
cat postgres-sink.json| kubectl exec -i -n kafka my-cluster-kafka-0 -- curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://my-connect-cluster-connect-api:8083/connectors -d @-
查看所有的Connector
kubectl exec -i -n kafka my-cluster-kafka-0 -- curl -X GET http://my-connect-cluster-connect-api:8083/connectors
删除Connect
kubectl exec -i -n kafka my-cluster-kafka-0 -- curl -X DELETE http://my-connect-cluster-connect-api:8083/connectors/postgres-sink
查看所有的topic
kubectl exec -n kafka my-cluster-kafka-0 -- bin/kafka-topics.sh --list --zookeeper localhost:2181
查看SQL Server Connector中的数据
kubectl exec -i -n kafka my-cluster-kafka-0 -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic exam.dbo.person --from-beginning
我们此时查看PostgreSQL数据库已经有了person表和数据,当对SQL Server新增、修改、删除数据时,PostgreSQL中也会同步更新。
相关推荐
图解 Kafka 之实战指南.pdf
strimzi-kafka-group-authorizer Strimzi Kafka Operator的简单授权者,可以基于模式为用户组配置ACL
k8s-kafka Kubernetes上的Kafka容器编辑controller.yaml并更改ZOOKEEPER_CONNECT环境以指向您已经拥有的Zookeeper实例。 Zookeeper的配置不在本文档范围内。 Zookeeper列表是逗号分隔的列表,例如host:port,host:...
图解 Kafka 之实战指南.7z
从zookeeper、kafka的安装,到kafka-connect的配置,有详细的步骤和参数的解释。
Apache Kafka实战.pdf..
kafka实战pdf
1. 高级-项目实战-日志收集系统kafka库实战 2. 高级-etcd、contex、kafka消费实例、logagent 3. 实战-商品秒杀架构设计与开发 4. 实战-商品秒杀开发与接入层实现 总共18课时,网上收集的资料,只共用于学习,不...
kafka-connect-ui 这是Kafka Connect的网络工具,用于设置和管理多个连接集群的连接器。现场演示与Docker独立运行docker run --rm -it -p 8000:8000 \ -e "CONNECT_URL=...
Kafka Connect JDBC连接器 kafka-connect-jdbc是一个用于与任何兼容JDBC的数据库之间加载数据。 可以在找到该连接器的文档。发展要构建开发版本,您需要Kafka的最新版本以及一系列上游Confluent项目,您必须从其相应...
kafka-connect-mqtt 此仓库包含用于Apache Kafka的MQTT源和接收器连接器。 已通过Kafka 2+进行了测试。 使用源连接器,您可以订阅MQTT主题,并将这些消息写到Kafka主题。 接收器连接器以相反的方式工作。 笔记: ...
Kafka Connect Redis 用于Redis的Kafka源和接收器连接器 连接器 来源 Kafka Connect Redis Source使用Redis发布/订阅订阅Redis通道/模式(包括),并将接收到的消息写入Kafka。 有关更多信息,请参见。 下沉 Kafka...
Kafka Streams是Kafka提供的一个用于构建流式处理程序的Java库,它与Storm、Spark等流式处理框架不同,是一个仅依赖于Kafka的Java库,而不是一个流式处理框架。除Kafka之外,Kafka Streams不需要额外的流式处理集群...
flink搭配kafka,构建流式采集框架,提供了docker部署方式脚本和k8s多副本方式部署脚本
kafka从入门基础,深入理解、核心设计,包括:kafka结构中的broker\topic\Partition\Producer\Consumer等
您可以通过使用kafka9-connect-mongodb分支将此连接器用于Kafka9。 用于Kafka Connect的MongoDB接收器连接器提供了从Kafka主题或主题集到MongoDB集合或多个集合的简单,连续的链接。 连接器使用Kafka消息,重命名...
The JDBC source and sink connectors allow you to exchange data between relational databases and Kafka. The JDBC source connector allows you to import data from any relational database with a JDBC ...
介绍通过安装该Kafka Connect连接器提供了监视目录的文件和在将新文件写入输入目录时读取数据的功能。 输入文件中的每个记录将根据用户提供的模式进行转换。 CSVRecordProcessor支持读取CSV或TSV文件。 它可以将CSV...
k8s-kafka Docker容器,用于通过环境中的配置设置在Ubuntu容器(java7)中运行kafka (0.8.1.1)。 环境 这些环境变量的名称经过精心选择,以与Kubernetes(主要是GKE)为定义的服务提供的内容兼容。 KAFKA_SERVER...
kafka监控工具KafkaOffsetMnitor 没有响应,需要修改offsetapp目录index.html ,进行本地化。压缩包文件就是所需要的文件。