Medindo o desempenho do MongoDB Kafka Connector

1
curl -X POST -H "Content-Type: application/json" --data '
2
  {"name": "datagen-users",
3
   "config": { "connector.class":    "io.confluent.kafka.connect.datagen.DatagenConnector",
4
   "kafka.topic": "topic333",
5
   "quickstart": "users",
6
   "key.converter": "org.apache.kafka.connect.storage.StringConverter",
7
   "value.converter": "org.apache.kafka.connect.json.JsonConverter",
8
   "value.converter.schemas.enable": "false",
9
   "max.interval": 50,
10
   "iterations": 5000,
11
   "tasks.max": "2"
12
}}' http://localhost:8083/connectors -w "\n"

1
curl -X POST -H "Content-Type: application/json" --data '
2
{"name": "kafkametadata3",
3
  "config": {
4
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
5
    "topics": "topic333",
6
    "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
7
    "writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy",
8
    "database": "kafka",
9
    "collection": "datagen",
10
    "errors.log.include.messages": true,
11
    "errors.deadletterqueue.context.headers.enable": true,
12
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
13
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
14
    "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.KafkaMetaDataStrategy",
15
    "tasks.max": 2,
16
    "value.converter.schemas.enable":false,
17
    "transforms": "InsertField",
18
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
19
    "transforms.InsertField.offset.field": "offsetColumn",
20
    "transforms": "InsertField",
21
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
22
    "transforms.InsertField.timestamp.field": "timestampColumn"
23
}}' http://localhost:8083/connectors -w "\n"

1
use kafka
2
db.createView("SinkView","datagen",
3
[
4
	{
5
		"$sort" : {
6
			"_insertedTS" : 1,
7
			"timestampColumn" : 1
8
		}
9
	},
10
	{
11
		"$project" : {
12
			"_insertedTS" : 1,
13
			"timestampColumn" : 1,
14
			"_id" : 0
15
		}
16
	},
17
	{
18
		"$addFields" : {
19
			"diff" : {
20
				"$subtract" : [
21
					"$_insertedTS",
22
					{
23
						"$convert" : {
24
							"input" : "$timestampColumn",
25
							"to" : "date"
26
						}
27
					}
28
				]
29
			}
30
		}
31
	}
32
])

1
curl https://gist.githubusercontent.com/RWaltersMA/555b5f17791ecb58e6e683c54bafd381/raw/748301bcb7ae725af4051d40b2e17a8882ef2631/sink-chart-performance.charts -o sink-performance.charts

1
curl -X POST -H "Content-Type: application/json" --data '
2
{"name": "mongo-source-perf",
3
  "config": {
4
  "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
5
  "errors.log.enable": "true",
6
  "errors.log.include.messages": "true",
7
  "connection.uri": "mongodb+srv://MONGODB CONNECTION STRING HERE",
8
  "database": "kafka",
9
  "collection": "source-perf-test",
10
  "mongo.errors.log.enable": "true",
11
  "topic.prefix":"mdb",
12
  "output.json.formatter" : "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
13
    "output.format.value":"schema",
14
    "output.schema.infer.value":true,
15
    "output.format.key":"json",
16
    "publish.full.document.only": "false",
17
  "change.stream.full.document": "updateLookup"
18
}}' http://localhost:8083/connectors -w "\n"

1
{
2
   "_id" : ObjectId("59b99db4cfa9a34dcd7885b6"),
3
   "name" : "Ned Stark",
4
   "email" : "sean_bean@gameofthron.es",
5
   "password" : "$2b$12$UREFwsRUoyF0CRqGNK0LzO0HM/jLhgUCNNIJ9RJAqMUQ74crlJ1Vu"
6
}

1
docker run robwma/doc-gen:1.0 python doc-gen.py -s '{"name":"string","email":"string","password":"string"}' -c "MONGODB CONNECTION STRING GOES HERE" -t 1000 -db "kafka" -col "source-perf-test"

1
CREATE STREAM stats (
2
   clusterTime BIGINT
3
 ) WITH (
4
   KAFKA_TOPIC='kafka.source-perf-test',
5
       VALUE_FORMAT='AVRO'
6
 );

1
{
2
  _id: { <BSON Object> },
3
  "operationType": "<operation>",
4
  "fullDocument": { <document> },
5
  "ns": {
6
    "db": <database>,
7
    "coll": <collection>
8
  },
9
  "to": {
10
    "db": <database>,
11
    "coll": <collection>
12
  },
13
  "documentKey": {
14
    _id: <value>
15
  },
16
  "updateDescription": {
17
    "updatedFields": { <document> },
18
    "removedFields": [ <field>, ... ]
19
  },
20
  "clusterTime": <Timestamp>,
21
  "txnNumber": <NumberLong>,
22
  "lsid": {
23
    "id": <UUID>,
24
    "uid": <BinData>
25
  }
26
}

1
CREATE STREAM STATS2 AS
2
  select ROWTIME - CLUSTERTIME as diff, 1 AS ROW from STATS EMIT CHANGES;

1
SET 'ksql.suppress.enabled' = 'true';
2
3
CREATE TABLE STATSWINDOW2 AS
4
  SELECT  AVG( DIFF ) AS AVG, MAX(DIFF) AS MAX,  count(*) AS COUNT,  ROW  FROM STATS2
5
 WINDOW TUMBLING (SIZE 10 SECONDS)
6
 GROUP BY ROW
7
 EMIT FINAL;

1
curl -X POST -H "Content-Type: application/json" --data '
2
{
3
 "name": "MongoSource-SinkPerf",
4
 "config": {
5
   "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
6
   "tasks.max": "1",
7
   "errors.log.enable": true,
8
   "errors.log.include.messages": true,
9
   "topics": "STATSWINDOW2",
10
   "errors.deadletterqueue.context.headers.enable": true,
11
   "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
12
   "database": "kafka",
13
   "collection": "sourceStats",
14
   "mongo.errors.log.enable": true,
15
   "transforms": "InsertField",
16
   "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
17
   "transforms.InsertField.timestamp.field": "timestampColumn"
18
}}' http://localhost:8083/connectors -w "\n"

1
curl https://gist.githubusercontent.com/RWaltersMA/011f1473cf937badc61b752a6ab769d4/raw/bc180b9c2db533536e6c65f34c30b2d2145872f9/mongodb-source-performance.chart -o source-performance.charts