基于MongoDB的实时数仓实现

1) 下载源码
    地址：https://github.com/debezium/debezium/archive/v0.10.0.Final.tar.gz
2) 业务需求
    在每条update/delete数据记录中增加oid标识，以提供数仓溯源使用。
3) 实现方法
    打开debezium/RecordMakers.java::createRecords() 中增加value.put("objectid", objId);
4) 编译
    命令：mvn install -pl debezium-connector-mongodb   -Ddocker.skip.build=true -Ddocker.skip.run=true  -DskipITs=true
5) 构建新docker镜像
    将编译后的包：debezium-connector-mongodb/target/debezium-connector-mongodb-0.10.0.Final.jar  拷贝到debezium/connect:0.10 Docker容器内。重新commit、push到测试环境。
6) 打包Sink功能
    将Mongo-Kafka 编译后的jar包(mongo-kafka-0.3-SNAPSHOT-all.jar) 拷贝到debezium/connect:0.10 Docker容器内/kafka/connect/mongodb-kafka-connect目录下。需要提前创建mongodb-kafka-connect目录。 重新commit、push image到测试环境。
7) 容器内目录结构
[kafka@deb-connect ~]$ ls -l connect/
total 8
drwxr-xr-x 1 kafka kafka   52 Dec  1 16:18 debezium-connector-mongodb
drwxr-xr-x 1 kafka kafka 4096 Oct  2 00:52 debezium-connector-mysql
drwxr-xr-x 1 kafka kafka  204 Oct  2 00:52 debezium-connector-oracle
drwxr-xr-x 1 kafka kafka  285 Oct  2 00:52 debezium-connector-postgres
drwxr-xr-x 1 kafka kafka  259 Oct  2 00:52 debezium-connector-sqlserver
drwxrwxr-x 1 kafka kafka   46 Nov 28 08:27 mongodb-kafka-connect
复制代码

# 由于需要提供Source和Sink功能，根据同步库的数量，适当的增加Docker数量，这样可以确保任务的正常高效执行。根据相同的GROUP_ID为一个集群，支持负载均衡。默认数据格式为：Avro。

# 依赖的环境变量如下：
      GROUP_ID: "DW-MongoToKafka"
      KAFKA_HEAP_OPTS: "-Xms2G -Xmx8G"
      SERVICE_28083_NAME: "dw-mongo-connect"
      SANITIZE_FIELD_NAMES: "true"
      CONNECT_PRODUCER_MAX_REQUEST_SIZE: 16777216
      KAFKA_PRODUCER_MAX_REQUEST_SIZE: 16777216
      STATUS_STORAGE_TOPIC: "debezium_connect_status"
      CONFIG_STORAGE_TOPIC: "debezium_connect_configs"
      OFFSET_STORAGE_TOPIC: "debezium_connect_offsets"
      KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://dw-schema-registry.com"
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://dw-schema-registry.com"
      BOOTSTRAP_SERVERS: "dn5.infra.app:9092, dn6.infra.app:9092, dn7.infra.app:9092"
复制代码

# 使用API方式创建source connector，开启实时同步MongoDB-Sharding数据到Kafka Topic

curl -X POST -H "Content-Type: application/json" --data
'{
  "name": "debezium-source-表名",
  "config": {
	  "connector.class":"io.debezium.connector.mongodb.MongoDbConnector",
	  "sanitize.field.names":"true",
	  "tasks.max":"1",
	  "mongodb.hosts":"mongos地址:端口",
	  "mongodb.user":"用户名",
	  "mongodb.password":"密码",
	  "mongodb.name":"datawarehouse.mongo.debezium",
	  "database.whitelist":"库名",
	  "collection.whitelist":"库名.表名",
	  "max.request.size":"16777216",
	  "database.history.kafka.bootstrap.servers":"dn5.infra.app:9092"
  }
}' http://dw-mongo-connect.com/connectors/
复制代码

# 使用API方式创建sink connector，开启实时增量同步Kafka数据到线下MongoDB-RS库。

curl -X POST -H "Content-Type: application/json" --data
'{
  "name": "debezium-sink-表名",
  "config": {
	  "tasks.max":"1",
	  "database":"目标库",
	  "topics":"填写source connector同步的topic",
	  "connection.uri":"mongodb://用户名:密码@IP:PORT/库名",
	  "collection":"表名",
	  "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
	  "change.data.capture.handler":"com.mongodb.kafka.connect.sink.cdc.debezium.mongodb.MongoDbHandler"
   }
}' http://dw-mongo-connect.com/connectors/
复制代码

# 由于kafka服务器存储受限，根据业务数据需求修改topic 保留失效为3天
kafka-topics --zookeeper zk地址:2181 --alter  --topic   TopicName --config retention.ms=259200000
复制代码

# 由于线上Mongo-Sharding集群对DataBase都有严格的权限管理，因此在创建connector后，一般会出现权限拒绝问题。错误信息如下

【2019-11-30 16:49:52,955 ERROR  MongoDB|datawarehouse.mongo.debezium|confrs  Error while attempting to get oplog position: Exception authenticating MongoCredential{mechanism=SCRAM-SHA-1, userName='同步用户', source='admin', password=<hidden>, mechanismProperties={}}   [io.debezium.connector.mongodb.Replicator]
com.mongodb.MongoSecurityException: Exception authenticating MongoCredential{mechanism=SCRAM-SHA-1, userName='同步用户', source='admin', password=<hidden>, mechanismProperties={}}】

使用Debezium Source connector 同步Mongo-sharding数据时，需要开启的权限为: mongos进入后admin库的read权限
mongos> show users;
{
	"_id" : "admin.同步用户",
	"userId" : UUID("fb982511-c779-41b8-8a9f-9ba492c30c28"),
	"user" : "同步用户",
	"db" : "admin",
	"roles" : [
		{
			"role" : "read",
			"db" : "risk"
		},
		{
			"role" : "read",
			"db" : "admin"
		},
		{
			"role" : "read",
			"db" : "config"
		}
	],
	"mechanisms" : [
		"SCRAM-SHA-1",
		"SCRAM-SHA-256"
	]
}

进入每个Replica下，创建 admin和local库的 read权限。
s5rs:PRIMARY> show users;
{
	"_id" : "admin.同步用户",
	"userId" : UUID("b99bd150-dc9c-4f67-8177-2580b78d63c1"),
	"user" : "同步用户",
	"db" : "admin",
	"roles" : [
		{
			"role" : "read",
			"db" : "local"
		},
		{
			"role" : "read",
			"db" : "admin"
		}
	],
	"mechanisms" : [
		"SCRAM-SHA-1",
		"SCRAM-SHA-256"
	]
}


使用Mongo-Kakfa Sink connector操作线下Mongodb时，需要开启权限：
riskPoolRs:PRIMARY> show users;
{
	"_id" : "risk.同步用户",
	"userId" : UUID("9f5e079f-a665-4664-830f-8b54f9848ea2"),
	"user" : "同步用户",
	"db" : "库名",
	"roles" : [
		{
			"role" : "readWrite",
			"db" : "risk"
		},
		{
			"role" : "read",
			"db" : "admin"
		},
		{
			"role" : "clusterAdmin",
			"db" : "admin"
		}
	],
	"mechanisms" : [
		"SCRAM-SHA-1",
		"SCRAM-SHA-256"
	]
}
复制代码

# 在etc/catalog下创建mongodb.properties
connector.name=mongodb
mongodb.seeds=IP:27017
mongodb.credentials=用户名:密码@库名
mongodb.schema-collection=presto_mongo
mongodb.socket-keep-alive=true
复制代码

bin/launcher stop
bin/launcher start
复制代码