vertica-->kafka-->mongodb数据流

川长思鸟来 2022-09-25 01:24 2阅读 0赞

此连接件为confluent修改过的jar包，这里不提供下载（涉及公司机密！！）

Kafka Connnect有两个核心概念：Source和Sink。 Source负责导入数据到Kafka，Sink负责从Kafka导出数据，它们都被称为Connector。

![Center][]

由于这里是kafka到mongo，所以这里使用Sink Connector

首先安装kafka

1，Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。

Kafka拓扑结构：

![Center 1][]

Broker  
Kafka集群包含一个或多个服务器，这种服务器被称为broker  
 Topic  
每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）  
 Partition  
Parition是物理上的概念，每个Topic包含一个或多个Partition.  
 Producer  
负责发布消息到Kafka broker  
 Consumer  
消息消费者，向Kafka broker读取消息的客户端。  
 Consumer Group  
每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。  
安装zookeeper  
配置环境变量，/etc/profile添加以下内容：  
\[root@dba06 kafka\]\# export ZOOKEEPER\_HOME=/opt/app/kafka/zookeeper-3.4.6  
\[root@dba06 kafka\]\# export PATH=$PATH:$ZOOKEEPER\_HOME/bin  
\[root@dba06 kafka\]\# cd zookeeper-3.4.6/conf/  
\[root@dba06 conf\]\# ls  
configuration.xsl  log4j.properties  zoo\_sample.cfg  
\[root@dba06 conf\]\# cat zoo\_sample.cfg | grep -v '\#' > zoo.cfg   
\[root@dba06 conf\]\# cat zoo.cfg   
tickTime=2000  
initLimit=10  
syncLimit=5  
dataDir=/tmp/zookeeper  
clientPort=2181  
\[root@dba06 conf\]\# ../bin/zkServer.sh start  
JMX enabled by default  
Using config: /opt/app/kafka/zookeeper-3.4.6/bin/../conf/zoo.cfg  
Starting zookeeper ... STARTED  
查看是否启动  
\[root@dba06 conf\]\# ../bin/zkServer.sh status  
JMX enabled by default  
Using config: /opt/app/kafka/zookeeper-3.4.6/bin/../conf/zoo.cfg  
Mode: standalone  
Server启动之后, 就可以启动client连接server了, 执行脚本:  
\[root@dba06 conf\]\# ../bin/zkCli.sh -server localhost:2181  
  
  
安装kafka  
安装kafka server之前需要单独安装zookeeper server，而且需要修改config/server.properties里面的IP信息  
\[root@dba06 kafka\]\# cd kafka\_2.11-0.9.0.0  
\[root@dba06 kafka\_2.11-0.9.0.0\]\# cd config/  
\[root@dba06 config\]\# vi server.properties   
zookeeper.connect=localhost:2181  
这里需要修改默认zookeeper.properties配置  
\[root@dba06 config\]\# vi zookeeper.properties   
保持默认设置  
先启动zookeeper，启动前先kill掉之前的zkServer.sh启动的zookeeper服务  
\[root@dba06 bin\]\# ./zookeeper-server-start.sh ../config/zookeeper.properties &  
启动kafka服务  
\[root@dba06 bin\]\# ./kafka-server-start.sh ../config/server.properties  &  
查看kafka进程是否启动：  
\[root@dba06 bin\]\# find /. -name 'java'  
重新安装Java  
\[root@dba06 bin\]\# yum install java-1.7.0  
\[root@dba06 bin\]\# yum install java-1.7.0-openjdk  java-1.7.0-openjdk-devel  
配置环境变量  
\[root@dba06 bin\]\# vi ~/.bash\_profile  
export PATH  
export JAVA\_HOME=/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.111.x86\_64/jre  
export CLASSPATH=.:$JAVA\_HOME/jre/lib/rt.jar:$JAVA\_HOME/lib/dt.jar:$JAVA\_HOME/lib/tools.jar  
export PATH=$PATH:$JAVA\_HOME/bin  
export LD\_LIBRARY\_PATH=$LD\_LIBRARY\_PATH:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.111.x86\_64/jre/lib/amd64/libjsig.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.111.x86\_64/jre/lib/amd64/server/libjvm.so:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.111.x86\_64/jre/lib/amd64/server:/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.111.x86\_64/jre/lib/amd64  
\[root@dba06 bin\]\# source ~/.bash\_profile  
\[root@dba06 bin\]\# jps  
835 Kafka  
803 QuorumPeerMain  
1679 Jps  
咱们先看看vertica，推数据到kafka  
dbadmin=> select version();  
              version                 
\------------------------------------  
 Vertica Analytic Database v7.2.3-0  
(1 row)  
  
  
dbadmin=> \\q  
这里采用7.2.3版本  
dbadmin=> SELECT KafkaExport(0, '1', '\{"EUTIME":"'||"EUTIME"||'"\}' USING PARAMETERS brokers='172.16.20.207:9092',topic='fafatest') OVER (PARTITION BEST) FROM tmp.DP\_TRADE\_PVCOUNT ;  
partition | key | message | failure\_reason   
\-----------+-----+---------+----------------  
(0 rows)  
说明，这里无需在对方创建topic，只需对方有zookeeper进程和kafka进程即可。  
报错详解：  
dbadmin=> SELECT KafkaExport(0, '1', '\{"EUTIME":"'||"STATDATE"||'"\}' USING PARAMETERS brokers='172.16.57.55:9092',topic='fafa01') OVER (PARTITION BEST) FROM tmp.DP\_TRADE\_PVCOUNT ;  
ERROR 5861:  Error calling processPartition() in User Function KafkaExport at \[src/KafkaExport.cpp:187\], error code: 0, message: Error reading topic metadata for topic fafa01: Broker: Leader not available  
dbadmin=> SELECT KafkaExport(0, '1', '\{"EUTIME":"'||"STATDATE"||'"\}' USING PARAMETERS brokers='172.16.57.55:9092',topic='fafa01') OVER () FROM tmp.DP\_TRADE\_PVCOUNT ;  
ERROR 5861:  Error calling processPartition() in User Function KafkaExport at \[src/KafkaExport.cpp:323\], error code: 0, message: Exception while processing partition row: \[0\] : \[Fatal kafka error: T2:9092/0: Failed to resolve 'T2:9092': Temporary failure in name resolution\]  
ERROR 5861问题是因为在双方的/etc/hosts下面要配置互相的ip，例如此时：  
172.16.57.55 T2  
172.16.57.208 bd-dev-vertica2-208 localhost  
在kafka上消费测试一下  
\[root@dba06 bin\]\# ./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic fafa01  
\{"EUTIME":"2016-04-30"\}  
\{"EUTIME":"2016-04-30"\}  
\{"EUTIME":"2016-05-01"\}  
\{"EUTIME":"2016-05-02"\}  
\{"EUTIME":"2016-05-02"\}  
\{"EUTIME":"2016-05-02"\}  
....................  
....................  
可以看到json格式数据已经消费。  
  
  
下面尝试kafka到mongo连接  
下载kafka到mongo连接件，这里下载的连接件需要再使用confluect上面的kafka才能使用  
http://www.confluent.io/product/connectors  
这里使用改写的单向的版本，使用普通的kafka  
\[root@dba06 libs\]\# rz  
?z waiting to receive.\*\*B0100000023be50  
即  
connect-mongodb-1.0-jar-with-dependencies.jar  
将jar报放在lib下面  
将配置文件放在config下面  
\[root@dba06 ~\]\# cd /opt/app/kafka/kafka\_2.11-0.9.0.0/config/  
connect-mongo160-sink.properties  
connect-standalone.properties  
配置kafka到mongo连接件  
\[root@T2 config\]\# vi connect-mongo160-sink.properties   
  
  
name=mongodb-sink-connector  
connector.class=org.apache.kafka.connect.mongodb.MongodbSinkConnector  
tasks.max=1  
host=172.16.57.160  
port=12001  
bulk.size=100  
mongodb.database=backbone  
mongodb.authorized.database=backbone  
skip.message.on.error=true  
mongodb.collections=test  
mongodb.user=backbone  
mongodb.password=Password$1  
topics=fafa01  
  
  
启动服务，连接服务  
这里先不管格式将connect-standalone.properties  
\[root@T2 bin\]\#  vi /opt/app/software/kafka\_2.11-0.10.0.0/config/connect-standalone.properties  
key.converter.schemas.enable=false  
value.converter.schemas.enable=false  
暂时关闭格式控制  
  
  
\[root@T2 bin\]\# ./connect-standalone.sh ../config/connect-standalone.properties ../config/connect-mongo160-sink.properties  
  
  
  
  
查看mongo端  
mongos> db.test.find()  
\{ "\_id" : NumberLong(2), "message" : "great", "EUTIME" : "2015-10-28" \}  
\{ "\_id" : NumberLong(3), "haoshibu" : "4", "EUTIME" : "2015-10-29" \}  
\{ "\_id" : NumberLong(4), "haoshibu" : "9", "EUTIME" : "2015-10-31" \}  
\{ "\_id" : NumberLong(5), "haoshibu" : "6", "EUTIME" : "2015-11-12" \}  
\{ "\_id" : NumberLong(6), "haoshibu" : "10", "EUTIME" : "2015-11-14" \}  
\{ "\_id" : NumberLong(7), "haoshibu" : "2", "EUTIME" : "2015-11-17" \}  
\{ "\_id" : NumberLong(8), "haoshibu" : "3", "EUTIME" : "2015-11-18" \}  
\{ "\_id" : NumberLong(9), "haoshibu" : "1", "EUTIME" : "2015-11-19" \}  
\{ "\_id" : NumberLong(10), "haoshibu" : "7", "EUTIME" : "2015-11-24" \}  
\{ "\_id" : NumberLong(11), "haoshibu" : "8", "EUTIME" : "2015-11-27" \}  
\{ "\_id" : NumberLong(12), "haoshibu" : "5", "EUTIME" : "2015-12-03" \}  
\{ "\_id" : NumberLong(13), "haoshibu" : "4", "EUTIME" : "2015-12-09" \}  
\{ "\_id" : NumberLong(14), "haoshibu" : "9", "EUTIME" : "2015-12-10" \}  
\{ "\_id" : NumberLong(15), "haoshibu" : "6", "EUTIME" : "2015-12-11" \}  
\{ "\_id" : NumberLong(16), "haoshibu" : "10", "EUTIME" : "2015-12-18" \}  
\{ "\_id" : NumberLong(17), "haoshibu" : "2", "EUTIME" : "2015-12-20" \}  
\{ "\_id" : NumberLong(18), "haoshibu" : "3", "EUTIME" : "2015-12-26" \}  
\{ "\_id" : NumberLong(19), "haoshibu" : "1", "EUTIME" : "2015-12-27" \}  
\{ "\_id" : NumberLong(20), "haoshibu" : "7", "EUTIME" : "2016-01-01" \}  
\{ "\_id" : NumberLong(21), "haoshibu" : "8", "EUTIME" : "2016-01-03" \}  
成功！

[Center]: /images/20220719/99ef243777bd4e3caf453eb79335fc0f.png
[Center 1]: /images/20220719/90685a5af087445b8376cb1f39da5925.png