Привет, я пытаюсь переместить данные в формате csv из kafka в hdfs, используя соединитель приемника hdfs. Ниже приведены свойства, которые я использовал
Connect.properties
name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
format.class=io.confluent.connect.hdfs.string.StringFormat
tasks.max=1
topics=topic_name
hadoop.conf.dir=/etc/hadoop/conf
hdfs.url=hdfs://nameservice1/dir
flush.size=3
hdfs.authentication.kerberos=true
connect.hdfs.principal=principal
connect.hdfs.keytab=principal.keytab
hdfs.namenode.principal=principal
partitioner.class=io.confluent.connect.hdfs.partitioner.TimeBasedPartitioner
partition.duration.ms=300000
path.format=path.format='year'=YYYY/'month'=MM/'day'=dd
locale=en
timezone=EST
свойства работника
bootstrap.servers=kafkaserver
plugin.path=/opt/confluent/share/java
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.file.filename=/tmp/connect.offsets
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
consumer.group.id=connect_group
consumer.auto.offset.reset=earliest
и я использую confluent-5.0.1
, но я получаю следующее исключение при запуске kafka connect
java.util.concurrent.ExecutionException: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: конфигурация соединителя недопустима и содержит следующие 1 ошибку: недопустимое значение io.confluent.connect.hdfs.string.StringFormat для конфигурации format.class: класс io.confluent.connect.hdfs.string.StringFormat не найден. ** Вы также можете найти приведенный выше список ошибок в конечной точке / {connectorType} / config / validate по адресу org.apache.kafka.connect.util.ConvertingFutureCallback.result (ConvertingFutureCallback.java:79) в org.apache.kafka.connect.util.ConvertingFutureCallback.get (ConvertingFutureCallback.java:66) в org.apache.kafka.connect.cli.ConnectStandalone.main (ConnectStandalone.java:110) Причина: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: недопустимая конфигурация коннектора и содержит следующие 1 ошибку: недопустимое значение io.confluent.connect.hdfs.string.StringFormat для конфигурации format.class: класс io.confluent.connect.hdfs.string.StringFormat не найден.Вы также можете найти приведенный выше список ошибок в конечной точке / {connectorType} / config / validate по адресу org.apache.kafka.connect.runtime.AbstractHerder.maybeAddConfigErrors (AbstractHerder.java:423) по адресу org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig (StandaloneHerder.java:189) в org.apache.kafka.connect.cli.ConnectStandalone.main (ConnectStandalone.java:107)