Как передавать данные с консоли kafka avro на HDFS, используя kafka-connect-hdfs? - PullRequest
Новая
       15

Как передавать данные с консоли kafka avro на HDFS, используя kafka-connect-hdfs?

0 голосов
/ 27 апреля 2018

Я пытаюсь запустить kafka-connect-hdfs безуспешно.

Я добавил следующую строку в .bash_profile и запустил 'source ~ / .bash_profile' 

export LOG_DIR=~/logs

Файл конфигурации quickstart-hdfs.properties: 

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
hdfs.url=xxx.xxx.xxx.xxx:xxxx # placeholder
flush.size=3

hadoop.conf.dir = /etc/hadoop/conf/
logs.dir = ~/logs
topics.dir = ~/topics
topics=test_hdfs

Я следую инструкциям по быстрому старту, изложенным в https://docs.confluent.io/current/connect/connect-hdfs/docs/hdfs_connector.html

Содержимое файла connector-avro-stanalone.properties: 

bootstrap.servers=yyy.yyy.yyy.yyy:yyyy # This is the placeholder for the Kafka broker url with the appropriate port
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
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
offset.storage.file.filename=/tmp/connect.offsets

У меня в домашнем каталоге есть quickstart-hdfs.properties и connector-avro-stanalone.properties, и я запускаю: 

confluent load hdfs-sink -d quickstart-hdfs.properties

Я не уверен, как получить доступ к информации в файле connector-avro-stanalone.properties в моем домашнем каталоге.

Когда я запускаю: 'confluent log connect', я получаю следующую ошибку: 

[2018-04-26 17:36:00,217] INFO Couldn't start HdfsSinkConnector: (io.confluent.connect.hdfs.HdfsSinkTask:90)
org.apache.kafka.connect.errors.ConnectException: java.lang.reflect.InvocationTargetException
        at io.confluent.connect.storage.StorageFactory.createStorage(StorageFactory.java:56)
        at io.confluent.connect.hdfs.DataWriter.<init>(DataWriter.java:213)
        at io.confluent.connect.hdfs.DataWriter.<init>(DataWriter.java:101)
        at io.confluent.connect.hdfs.HdfsSinkTask.start(HdfsSinkTask.java:82)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:267)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:163)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.reflect.InvocationTargetException
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at io.confluent.connect.storage.StorageFactory.createStorage(StorageFactory.java:51)
        ... 12 more
Caused by: java.io.IOException: No FileSystem for scheme: xxx.xxx.xxx.xxx -> This is the hdfs_url in quickstart-hdfs.properties file without the port
        at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2660)
        at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2667)
        at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:94)
        at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
        at org.apache.hadoop.fs.FileSystem$Cache.getUnique(FileSystem.java:2691)
        at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:420)
        at io.confluent.connect.hdfs.storage.HdfsStorage.<init>(HdfsStorage.java:56)
        ... 17 more
[2018-04-26 17:36:00,217] INFO Shutting down HdfsSinkConnector. (io.confluent.connect.hdfs.HdfsSinkTask:91)

Любая помощь для исправления этой проблемы будет принята с благодарностью.

1 Ответ

0 голосов
/ 28 апреля 2018

Причина: java.io.IOException: Нет файловой системы для схемы: xxx.xxx.xxx.xxx

Вам нужно hdfs.url=hdfs://xxx.yyy.zzz.abc вместе с портом наменода

Кроме того, вы захотите удалить пробелы вокруг знаков равенства в файлах свойств

...