Ошибка приемника Kafka «Этот соединитель требует, чтобы записи из Kafka содержали ключи для таблицы Cassandra» - PullRequest
Новая
       60

Ошибка приемника Kafka «Этот соединитель требует, чтобы записи из Kafka содержали ключи для таблицы Cassandra»

0 голосов
/ 25 октября 2018

Я пытаюсь синхронизировать все таблицы, прочитанные из Sap в cassandra, используя kafka, вот моя конфигурация cassandra 

{
    "name": "cassandra",
    "config": {
        "connector.class": "io.confluent.connect.cassandra.CassandraSinkConnector",
        "tasks.max": "5",
        "topics" :"sap_table1,sap_table2",
        "cassandra.keyspace": "sap",
        "cassandra.compression":"SNAPPY",
        "cassandra.consistency.level":"LOCAL_QUORUM",
        "cassandra.write.mode":"Update",
        "transforms":"prune", 
       "transforms.prune.type":"org.apache.kafka.connect.transforms.ReplaceField$Value",
        "transforms.prune.whitelist":"CreatedAt,Id,Text,Source,Truncated",
        "transforms.ValueToKey.fields":"ROWTIME"

    }
}

Я получаю эту ошибку 

Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. (org.apache.kafka.connect.runtime.WorkerSinkTask:584) org.apache.kafka.connect.errors.DataException: Record with a null key was encountered.  This connector requires that records from Kafka contain the keys for the Cassandra table. Please use a transformation like org.apache.kafka.connect.transforms.ValueToKey to create a key with the proper fields.

Все таблицы, сгенерированные из kafkaSap Connectior без ключа я не знаю, если это проблема

, дайте мне знать, если я делаю что-нибудь, wring

спасибо

Ответы [ 2 ]

0 голосов
/ 21 января 2019

эта ошибка означает, что ваши данные не сериализованы, поэтому они не в формате json или в словаре {ключ ':' значение '}.если вы читаете данные напрямую из брокера в качестве способа устранения неполадок, вы обнаружите, что ваши данные имеют только значения без каких-либо ключей:

используйте эту команду для чтения данных из брокера: 

/bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic your_topic_name--from-beginning

поэтому лучший способ решить эту проблему - добавить сериализатор в файл конфигурации вашего издателя.попробуйте этот файл в качестве исходного соединителя или издателя 

name=src-view
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
topic.prefix=test-
connection.url=jdbc:postgresql://127.0.0.1:5434/test?user=testuser&password=testpass
mode=incrementing
incrementing.column.name=id
table.types=table
table.whitelist=table_name
validate.non.null=false
batch.max.rows=10000
bootstrap.servers=localhost:9092

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schema.registry.url=http://localhost:8081

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

, а ниже - потребитель (sink.conf) для десериализации ваших данных: 

name=cas-dest
connector.class=io.confluent.connect.cassandra.CassandraSinkConnector
tasks.max=1
topics=your_topic_name
cassandra.contact.points=127.0.0.1
cassandra.port=9042
cassandra.keyspace=your_keyspace_name
cassandra.write.mode=Update
cassandra.keyspace.create.enabled=true
cassandra.table.manage.enabled=true
key.converter.schema.registry.url=http://localhost:8081
value.converter.schema.registry.url=http://localhost:8081
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
transforms=createKey
transforms.createKey.fields=id,timestamp
transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey

измените createKey.fields какследите за своими данными и будьте осторожны, так как это будут ключи вашего раздела, так что прочитайте о моделировании данных в cassandra, прежде чем выбирать ключи, и они должны существовать в вашем ключе данных.

0 голосов
/ 03 ноября 2018

"ROWTIME" существует только как концепция KSQL.На самом деле это не поле в вашем значении, поэтому ключ имеет значение null.

Кроме того, ValueToKey не указан в списке transforms, поэтому он даже не применяется.Вам также нужно будет добавить "transforms.ValueToKey.type".

Вам придется использовать другой метод преобразования, чтобы установить метку времени записи в качестве ключа сообщения ConnectRecord

...