将Kafka和HDFS与码头集装箱连接起来
将Kafka和HDFS与码头集装箱连接起来
提问于 2020-01-21 16:39:36
你好，伙计们，我想把卡夫卡和HDFS和卡夫卡连接起来，但我仍然面临着一个我无法摆脱的问题。
我正在使用这个例子：
I首先用: docker-compose -d启动HDFS
然后，我用debezium网站上的图片启动了动物园管理员kafka和mysql。 。
码头运行-it -rm-名称动物园管理员-网络码头-hadoop-主机_默认-p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/动物园管理员:1.0
docker run -it -rm-名称kafka --网络停靠器-hadoop-master_ -e ZOOKEEPER_CONNECT=zookeeper -p 9092:9092 -链接动物园管理员:动物园管理员debezium/kafka:1.0
docker run -it -rm-name mysql --网络停靠-hadoop-master_ -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium /示例-mysql:1.0
在这些运行过程中，我使用网络，因为当我试图在docker上改变HDFS时--come.yml--资源管理器关闭了，而且无论我如何发现我如何能够再次启动并使其稳定。就这样直接添加到这些容器上，动物园管理员kafka和mysql.。
，这是最棘手的部分，卡夫卡连接，我在这个案例中使用了相同的网络，这很有意义。
docker run -it --rm -名称连接-网络对接-hadoop-主机默认-p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses -e BOOTSTRAP_SERVERS="172.18.0.10:9092“-e CORE_CONF_fs_defaultFS=hdfs://172.18.0.2:9000 -link namenode:namenode -链接动物园管理员-链接mysql:mysql debezium/ connect :1.0
将源代码(Mysql)链接到Kafka，我使用debezium教程中的连接器，如下所示.
curl -i -X POST -H "Accept:application/json“-H内容-Type:application/json”localhost:8083/连接器/ -d '{ "name"：“库存-连接器”，"config"：{ "connector.class"："io.debezium.connector.mysql.MySqlConnector"，"tasks.max"："1"，"database.hostname"："mysql"，"database.port"："3306"，"database.user"："debezium"，"database.password"："dbz"，"database.server.id"："184054"，"database.server.name"："dbserver1"，"database.whitelist"：“库存”，"database.history.kafka.bootstrap.servers"："kafka:9092"，"database.history.kafka.topic"："dbhistory.inventory“}‘
我测试了卡夫卡是否从源接收到任何事件，并且运行良好.
在设置了这个插件之后，我转到了插件的安装上，我从汇合网站下载了这个插件，并粘贴到我的本地机器Linux上，然后我安装了con连贯-集线器，然后在我的本地机器上安装了插件。然后我创建了用户kafka，并将插件目录中的所有内容更改为kafka:kafka.。
，在这之后，我使用了docker :/ Kafka /connect复制到Kafka.
然后检查它是否在那里，然后重新启动Kafka来安装它.
我们可以用它来检查是否已安装： curl -i -X GET -H "Accept:application/json“localhost:8083/连接器插件
您需要看到以下内容：[{"class":"io.confluent.connect.hdfs.HdfsSinkConnector"，“type”：“接收器”，"version":"5.4.0"}，…
在这一步之后，我相信我的问题就在这里: curl -i -X POST -H "Accept:application/json“-H内容-Type:application/json”localhost:8083/连接器/ -d‘{“name”：“hdfs-接收器”，-X "tasks.max":1，“dbserver1，dbserver1.inventory.products，dbserver1.inventory.products_on_hand，dbserver1.inventory.customers，dbserver1.inventory.orders，"hdfs.url":"hdfs://172.18.0.2:9000"，"flush.size":3，"logs.dir":"logs"，"topics.dir":"kafka"，"format.class":"io.confluent.connect.hdfs.parquet.ParquetFormat"，"partitioner.class":"io.confluent.connect.hdfs.partitioner.DefaultPartitioner"，“partition.field.name”：“day”}‘
我不知道如何说服Kafka我想从namenode获得一个特定的IP地址，他只保留了我的trowing消息，当期望的是hdfs://namenode:9000时，这些消息找到了不同的IP
还将这个 -e CORE_CONF_fs_defaultFS=hdfs://172.18.0.2:9000 添加到码头，在Kafka连接中运行我们的设置，当我发布hdfs的卷-接收器时，他向我发送以下消息。
卡夫卡连接的日志：
2020-01-21 15:22:09,597 INFO || Creating connector hdfs-sink of type io.confluent.connect.hdfs.HdfsSinkConnector [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,597 INFO || Instantiated connector hdfs-sink with version 5.4.0 of type class io.confluent.connect.hdfs.HdfsSinkConnector [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,598 INFO || HdfsSinkConnectorConfig values:
avro.codec = null
connect.hdfs.keytab =
connect.hdfs.principal =
connect.meta.data = true
enhanced.avro.schema.support = false
filename.offset.zero.pad.width = 10
flush.size = 3
format.class = class io.confluent.connect.hdfs.parquet.ParquetFormat
hadoop.conf.dir =
hadoop.home =
hdfs.authentication.kerberos = false
hdfs.namenode.principal =
hdfs.url = hdfs://172.18.0.2:9000
kerberos.ticket.renew.period.ms = 3600000
logs.dir = logs
retry.backoff.ms = 5000
rotate.interval.ms = -1
rotate.schedule.interval.ms = -1
schema.cache.size = 1000
schema.compatibility = NONE
shutdown.timeout.ms = 3000
[io.confluent.connect.hdfs.HdfsSinkConnectorConfig]
2020-01-21 15:22:09,599 INFO || StorageCommonConfig values:
directory.delim = /
file.delim = +
storage.class = class io.confluent.connect.hdfs.storage.HdfsStorage
store.url = null
topics.dir = kafka
[io.confluent.connect.storage.common.StorageCommonConfig]
2020-01-21 15:22:09,599 INFO || HiveConfig values:
hive.conf.dir =
hive.database = default
hive.home =
hive.integration = false
hive.metastore.uris =
[io.confluent.connect.storage.hive.HiveConfig]
2020-01-21 15:22:09,600 INFO || PartitionerConfig values:
locale =
partition.duration.ms = -1
partition.field.name = [day]
partitioner.class = class io.confluent.connect.hdfs.partitioner.DefaultPartitioner
path.format =
timestamp.extractor = Wallclock
timestamp.field = timestamp
timezone =
[io.confluent.connect.storage.partitioner.PartitionerConfig]
2020-01-21 15:22:09,601 INFO || Finished creating connector hdfs-sink [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,601 INFO || SinkConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.deadletterqueue.context.headers.enable = false
errors.deadletterqueue.topic.name =
errors.deadletterqueue.topic.replication.factor = 3
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
topics = [dbserver1, dbserver1.inventory.products, dbserver1.inventory.products_on_hand, dbserver1.inventory.customers, dbserver1.inventory.orders, dbserver1.inventory.geom, dbserver1.inventory.addresses]
topics.regex =
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.SinkConnectorConfig]
2020-01-21 15:22:09,602 INFO || EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.deadletterqueue.context.headers.enable = false
errors.deadletterqueue.topic.name =
errors.deadletterqueue.topic.replication.factor = 3
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
topics = [dbserver1, dbserver1.inventory.products, dbserver1.inventory.products_on_hand, dbserver1.inventory.customers, dbserver1.inventory.orders, dbserver1.inventory.geom, dbserver1.inventory.addresses]
topics.regex =
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig]
2020-01-21 15:22:09,604 INFO || [Worker clientId=connect-1, groupId=1] Starting task hdfs-sink-0 [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
2020-01-21 15:22:09,605 INFO || Creating task hdfs-sink-0 [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,606 INFO || ConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.ConnectorConfig]
2020-01-21 15:22:09,607 INFO || EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig]
2020-01-21 15:22:09,608 INFO || TaskConfig values:
task.class = class io.confluent.connect.hdfs.HdfsSinkTask
[org.apache.kafka.connect.runtime.TaskConfig]
2020-01-21 15:22:09,608 INFO || Instantiated task hdfs-sink-0 with version 5.4.0 of type io.confluent.connect.hdfs.HdfsSinkTask [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,609 INFO || JsonConverterConfig values:
converter.type = key
decimal.format = BASE64
schemas.cache.size = 1000
schemas.enable = true
[org.apache.kafka.connect.json.JsonConverterConfig]
2020-01-21 15:22:09,610 INFO || Set up the key converter class org.apache.kafka.connect.json.JsonConverter for task hdfs-sink-0 using the worker config [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,610 INFO || JsonConverterConfig values:
converter.type = value
decimal.format = BASE64
schemas.cache.size = 1000
schemas.enable = true
[org.apache.kafka.connect.json.JsonConverterConfig]
2020-01-21 15:22:09,611 INFO || Set up the value converter class org.apache.kafka.connect.json.JsonConverter for task hdfs-sink-0 using the worker config [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,611 INFO || Set up the header converter class org.apache.kafka.connect.storage.SimpleHeaderConverter for task hdfs-sink-0 using the worker config [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,613 INFO || Initializing: org.apache.kafka.connect.runtime.TransformationChain{} [org.apache.kafka.connect.runtime.Worker]
2020-01-21 15:22:09,614 INFO || SinkConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.deadletterqueue.context.headers.enable = false
errors.deadletterqueue.topic.name =
errors.deadletterqueue.topic.replication.factor = 3
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
topics = [dbserver1, dbserver1.inventory.products, dbserver1.inventory.products_on_hand, dbserver1.inventory.customers, dbserver1.inventory.orders, dbserver1.inventory.geom, dbserver1.inventory.addresses]
topics.regex =
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.SinkConnectorConfig]
2020-01-21 15:22:09,618 INFO || EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = io.confluent.connect.hdfs.HdfsSinkConnector
errors.deadletterqueue.context.headers.enable = false
errors.deadletterqueue.topic.name =
errors.deadletterqueue.topic.replication.factor = 3
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = hdfs-sink
tasks.max = 1
topics = [dbserver1, dbserver1.inventory.products, dbserver1.inventory.products_on_hand, dbserver1.inventory.customers, dbserver1.inventory.orders, dbserver1.inventory.geom, dbserver1.inventory.addresses]
topics.regex =
transforms = []
value.converter = null
[org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig]
2020-01-21 15:22:09,622 INFO || ConsumerConfig values:
allow.auto.create.topics = true
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [172.18.0.10:9092]
check.crcs = true
client.dns.lookup = default
client.id = connector-consumer-hdfs-sink-0
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = connect-hdfs-sink
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
[org.apache.kafka.clients.consumer.ConsumerConfig]
2020-01-21 15:22:09,653 INFO || Kafka version: 2.4.0 [org.apache.kafka.common.utils.AppInfoParser]
2020-01-21 15:22:09,653 INFO || Kafka commitId: 77a89fcf8d7fa018 [org.apache.kafka.common.utils.AppInfoParser]
2020-01-21 15:22:09,654 INFO || Kafka startTimeMs: 1579620129652 [org.apache.kafka.common.utils.AppInfoParser]
2020-01-21 15:22:09,659 INFO || [Worker clientId=connect-1, groupId=1] Finished starting connectors and tasks [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
2020-01-21 15:22:09,677 INFO || [Consumer clientId=connector-consumer-hdfs-sink-0, groupId=connect-hdfs-sink] Subscribed to topic(s): dbserver1, dbserver1.inventory.products, dbserver1.inventory.products_on_hand, dbserver1.inventory.customers, dbserver1.inventory.orders, dbserver1.inventory.geom, dbserver1.inventory.addresses [org.apache.kafka.clients.consumer.KafkaConsumer]
2020-01-21 15:22:09,678 INFO || HdfsSinkConnectorConfig values:
avro.codec = null
connect.hdfs.keytab =
connect.hdfs.principal =
connect.meta.data = true
enhanced.avro.schema.support = false
filename.offset.zero.pad.width = 10
flush.size = 3
format.class = class io.confluent.connect.hdfs.parquet.ParquetFormat
hadoop.conf.dir =
hadoop.home =
hdfs.authentication.kerberos = false
hdfs.namenode.principal =
hdfs.url = hdfs://172.18.0.2:9000
kerberos.ticket.renew.period.ms = 3600000
logs.dir = logs
retry.backoff.ms = 5000
rotate.interval.ms = -1
rotate.schedule.interval.ms = -1
schema.cache.size = 1000
schema.compatibility = NONE
shutdown.timeout.ms = 3000
[io.confluent.connect.hdfs.HdfsSinkConnectorConfig]
2020-01-21 15:22:09,679 INFO || StorageCommonConfig values:
directory.delim = /
file.delim = +
storage.class = class io.confluent.connect.hdfs.storage.HdfsStorage
store.url = null
topics.dir = kafka
[io.confluent.connect.storage.common.StorageCommonConfig]
2020-01-21 15:22:09,679 INFO || HiveConfig values:
hive.conf.dir =
hive.database = default
hive.home =
hive.integration = false
hive.metastore.uris =
[io.confluent.connect.storage.hive.HiveConfig]
2020-01-21 15:22:09,680 INFO || PartitionerConfig values:
locale =
partition.duration.ms = -1
partition.field.name = [day]
partitioner.class = class io.confluent.connect.hdfs.partitioner.DefaultPartitioner
path.format =
timestamp.extractor = Wallclock
timestamp.field = timestamp
timezone =
[io.confluent.connect.storage.partitioner.PartitionerConfig]
2020-01-21 15:22:09,681 INFO || AvroDataConfig values:
connect.meta.data = true
enhanced.avro.schema.support = false
schemas.cache.config = 1000
[io.confluent.connect.avro.AvroDataConfig]
2020-01-21 15:22:09,681 INFO || Hadoop configuration directory [io.confluent.connect.hdfs.DataWriter]
2020-01-21 15:22:09,757 ERROR || WorkerSinkTask{id=hdfs-sink-0} Task threw an uncaught and unrecoverable exception [org.apache.kafka.connect.runtime.WorkerTask]
java.lang.IllegalArgumentException: java.net.URISyntaxException: Illegal character in hostname at index 36: hdfs://namenode.docker-hadoop-master_default:9000
at org.apache.hadoop.net.NetUtils.getCanonicalUri(NetUtils.java:274)
at org.apache.hadoop.hdfs.DistributedFileSystem.canonicalizeUri(DistributedFileSystem.java:1577)
at org.apache.hadoop.fs.FileSystem.getCanonicalUri(FileSystem.java:235)
at org.apache.hadoop.fs.FileSystem.checkPath(FileSystem.java:623)
at org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:194)
at org.apache.hadoop.hdfs.DistributedFileSystem.access$000(DistributedFileSystem.java:106)
at org.apache.hadoop.hdfs.DistributedFileSystem$22.doCall(DistributedFileSystem.java:1305)
at org.apache.hadoop.hdfs.DistributedFileSystem$22.doCall(DistributedFileSystem.java:1301)
at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1317)
at org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:1426)
at io.confluent.connect.hdfs.storage.HdfsStorage.exists(HdfsStorage.java:149)
at io.confluent.connect.hdfs.DataWriter.createDir(DataWriter.java:548)
at io.confluent.connect.hdfs.DataWriter.<init>(DataWriter.java:222)
at io.confluent.connect.hdfs.DataWriter.<init>(DataWriter.java:102)
at io.confluent.connect.hdfs.HdfsSinkTask.start(HdfsSinkTask.java:84)
at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:301)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:189)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.net.URISyntaxException: Illegal character in hostname at index 36: hdfs://namenode.docker-hadoop-master_default:9000
at java.base/java.net.URI$Parser.fail(URI.java:2913)
at java.base/java.net.URI$Parser.parseHostname(URI.java:3448)
at java.base/java.net.URI$Parser.parseServer(URI.java:3297)
at java.base/java.net.URI$Parser.parseAuthority(URI.java:3216)
at java.base/java.net.URI$Parser.parseHierarchical(URI.java:3158)
at java.base/java.net.URI$Parser.parse(URI.java:3114)
at java.base/java.net.URI.<init>(URI.java:685)
at org.apache.hadoop.net.NetUtils.getCanonicalUri(NetUtils.java:272)
... 24 more
2020-01-21 15:22:09,759 ERROR || WorkerSinkTask{id=hdfs-sink-0} Task is being killed and will not recover until manually restarted [org.apache.kafka.connect.runtime.WorkerTask]IP地址: Namenode: 172.18.0.2 Kafka:172.18.0.10
我相信这可能与网络有关。
错误在下面，希望你们能帮我。提前感谢！
Stack Overflow用户
回答已采纳
发布于 2020-01-21 17:28:45
默认情况下，Docker compose添加下划线，在主机名中不允许运行命令下划线的目录。默认情况下，Hadoop更喜欢hdfs-site.xml配置文件中的主机名。
我不知道如何说服Kafka Connect我想从namenode获得一个特定的IP地址，他只保留了我的trowing消息，当预期的是hdfs://namenode:9000时，这些消息找到了不同的IP
理想情况下，您不会在Docker中使用IP，而是使用服务名称和公开端口。
对于HDFS，还需要定义1) HADOOP_CONF_DIR env-var 2)将您的XML信任挂载为远程客户端(如Connect )与Hadoop集群交互的卷，3)在连接器属性中定义hadoop.conf.dir。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/59845575
复制相关文章
相似问题
腾讯云开发者
