Невозможно отправить данные из kafka topi c вasticsearch - PullRequest
Новая
       66

Невозможно отправить данные из kafka topi c вasticsearch

0 голосов
/ 05 апреля 2020

Я пытаюсь построить конвейер данных, используя mon go, получить данные из mongoDB в мои kafka topi c db (в качестве источника) ,asticsearch (в качестве приемника) и kafka. Я успешно получил данные от mongoDB на мой kafka topi c. Это пример данных, собранных из mongoDB 

{"_id": {"_data": "825E88FED8000000012B022C0100296E5A10044D2CA180FAF94580B30CFA4B3CC80E1546645F696400645E88FED793AFA61A58411B2A0004"}, "operationType": "insert", "clusterTime": {"$timestamp": {"t": 1586036440, "i": 1}}, "fullDocument": {"_id": {"$oid": "5e88fed793afa61a58411b2a"}, "name": "Lefèvre Mathis", "phoneNumber": 87640262, "phoneNumber2": 98462768, "phoneNumber3": 50591075, "email": "LefèvreMathis@gmail.com", "websiteUrl": "www.LefèvreMathis.fr", "legalInformation": {"companyName": "Duval EI", "siren": 7.3887975858196E13, "nic": 28866, "siret": 7.3887975858196E13, "ape": "49.53", "tva": "FR-1173030343", "description": "Blanditiis et placeat voluptas hic et. Quae et autem inventore ut enim fugit. Nihil velit in ut magnam."}, "professionType": {"type": "Hotel", "category": "professionnel"}, "operator": {"name": "Orange"}, "address": [{"city": "Paris", "street": "Quartier Les Halles, Paris 1er Arrondissement, Paris, Île-de-France, France métropolitaine, 75001, France", "zipCode": 75001, "latitude": "48.86330665", "longitude": "2.348370623761905"}], "openingTimeSet": [{"day": "Lundi", "opening": "08:00", "closing": "18:00"}, {"day": "Mardi", "opening": "08:00", "closing": "18:00"}, {"day": "Mercredi", "opening": "08:00", "closing": "18:00"}, {"day": "Jeudi", "opening": "08:00", "closing": "18:00"}, {"day": "Vendredi", "opening": "08:00", "closing": "18:00"}, {"day": "Samedi", "opening": "08:00", "closing": "18:00"}, {"day": "Dimanche", "opening": "08:00", "closing": "18:00"}], "_class": "com.sofrecom.elasticsearch.model.Subscriber"}, "ns": {"db": "elasticsearchApp", "coll": "subscriber"}, "documentKey": {"_id": {"$oid": "5e88fed793afa61a58411b2a"}}}

Проблема заключается в том, что при запуске моего коннектора ES-приемника я получаю следующее исключение: 

Caused by: org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error: 
at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:355)
at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:86)
at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:485)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
... 13 more

Caused by: org.apache.kafka.common.errors.SerializationException: java.io.CharConversionException: Invalid UTF-32 character 0x658b027b (above 0x0010ffff) at char #1, byte #7)

Это моя конфигурация kafka-connect: 

 CONNECT_BOOTSTRAP_SERVERS: kafka:9092
  CONNECT_REST_ADVERTISED_HOST_NAME: connect
  CONNECT_REST_PORT: 8083
  CONNECT_GROUP_ID: compose-connect-group
  CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
  CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
  CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
  CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
  CONNECT_VALUE_CONVERTER:  org.apache.kafka.connect.json.JsonConverter
  CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
  CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
  CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR:  1
  CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR:  1
  CONNECT_STATUS_STORAGE_REPLICATION_FACTOR:  1
  CONNECT_PLUGIN_PATH: '/usr/share/java,/etc/kafka-connect/jars'
  CONNECT_CONFLUENT_TOPIC_REPLICATION_FACTOR: 1

мой разъем es-sink: 

{ "name": "sink", "config": { "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "connection.url": "http://172.21.0.4:9200", "type.name": "subscriber", "topics": "test5.elasticsearchApp.subscriber", "key.ignore": "false","value.converter.schemas.enable": "false","schema.ignore": "true","value.converter":"org.apache.kafka.connect.json.JsonConverter" } }

и mongodb-source-connector 

{ "name": "mongo-source", "config": { "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector","tasks.max":1,"connection.uri":"mongodb://mongo1:27017,mongo2:27017","database":"elasticsearchApp","collection":"subscriber", "topic.prefix":"test15","value.converter":"org.apache.kafka.connect.storage.StringConverter"} }

Когда Я попытался использовать конвертер json в моем mongoDBConnector. Я получаю формат String для своей полезной нагрузки при использовании kafka topi c

{"schema":{"type":"string","optional":false},"payload":"{\"_id\": {\"_data\": \"825E89EA94000000012B022C0100296E5A10044D2CA180FAF94580B30CFA4B3CC80E1546645F696400645E89EA94FC56002500157F490004\"}, \"operationType\": \"insert\", \"clusterTime\": {\"$timestamp\": {\"t\": 1586096788, \"i\": 1}}, \"fullDocument\": {\"_id\": {\"$oid\": \"5e89ea94fc56002500157f49\"}, \"name\": \"Lefèvre Mathis\", \"phoneNumber\": 87640262, \"phoneNumber2\": 98462768, \"phoneNumber3\": 50591075, \"email\": \"LefèvreMathis@gmail.com\", \"websiteUrl\": \"www.LefèvreMathis.fr\", \"legalInformation\": {\"companyName\": \"Duval EI\", \"siren\": 7.3887975858196E13, \"nic\": 28866, \"siret\": 7.3887975858196E13, \"ape\": \"49.53\", \"tva\": \"FR-1173030343\", \"description\": \"Blanditiis et placeat voluptas hic et. Quae et autem inventore ut enim fugit. Nihil velit in ut magnam.\"}, \"professionType\": {\"type\": \"Hotel\", \"category\": \"professionnel\"}, \"operator\": {\"name\": \"Orange\"}, \"address\": [{\"city\": \"Paris\", \"street\": \"Quartier Les Halles, Paris 1er Arrondissement, Paris, Île-de-France, France métropolitaine, 75001, France\", \"zipCode\": 75001, \"latitude\": \"48.86330665\", \"longitude\": \"2.348370623761905\"}], \"openingTimeSet\": [{\"day\": \"Lundi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Mardi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Mercredi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Jeudi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Vendredi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Samedi\", \"opening\": \"08:00\", \"closing\": \"18:00\"}, {\"day\": \"Dimanche\", \"opening\": \"08:00\", \"closing\": \"18:00\"}], \"_class\": \"com.sofrecom.elasticsearch.model.Subscriber\"}, \"ns\": {\"db\": \"elasticsearchApp\", \"coll\": \"subscriber\"}, \"documentKey\": {\"_id\": {\"$oid\": \"5e89ea94fc56002500157f49\"}}}"}

1 Ответ

0 голосов
/ 06 апреля 2020

  1. Не используйте это, если вы не хотите, чтобы соединитель Mon go генерировал полезную нагрузку строки 

    "value.converter":"org.apache.kafka.connect.storage.StringConverter"

  2. Вам понадобится это в стоке, потому что у вас есть и schema, и payload в вашем JSON в топи c

    "value.converter.schemas.enable": "true"

  3. Вам нужно будет использовать отображение индекса Elasticsearch разобрать строку, так как Connect не сделает этого за вас.

Я не уверен, есть ли ошибка в разъеме Mon go. Никогда не использовал его, но я хотел бы думать, что JSON Comverter должен работать, или, по крайней мере, Avro.

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...