Файлы потока данных Spring Cloud в HDFS - PullRequest
Новая
       28

Файлы потока данных Spring Cloud в HDFS

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

Я новичок в Spring Cloud Data Flow.Я использую v 1.7.3 и хочу создать простой поток, который сканирует каталог на наличие новых файлов и передает их в HDFS.У меня есть следующее определение:

file --cron='* * * * * *' --mode=ref --directory=/dir | hdfs --fs-uri=hdfs://myhdpmaster:8020

При развертывании потока у меня возникают две проблемы:

  1. Независимо от того, какой файл modeЯ использую только созданный hdfs-sink-0.txt, у которого вообще нет содержимого или строк, которые выводят вывод по умолчанию для toString () (например, '[B @ 7d5bfc85').

  2. Когда я помещаю новые файлы в каталог, сообщение не принимается приемником HDFS, хотя в журналах источника файлов я вижу, что сообщение было создано.

вывод моей раковины hdfs: 

2019-01-25 12:21:06.330  INFO 63 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 0.10.1.1
2019-01-25 12:21:06.330  INFO 63 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : f10ef2720b03b247
2019-01-25 12:21:06.338  INFO 63 --- [           main] s.i.k.i.KafkaMessageDrivenChannelAdapter : started org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter@67110f71
2019-01-25 12:21:06.338  INFO 63 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : Adding {message-handler:inbound.testhdfs1.file.testhdfs1} as a subscriber to the 'bridge.testhdfs1.file' channel
2019-01-25 12:21:06.338  INFO 63 --- [           main] o.s.i.endpoint.EventDrivenConsumer       : started inbound.testhdfs1.file.testhdfs1
2019-01-25 12:21:06.340  INFO 63 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 2147483647
2019-01-25 12:21:06.476  INFO 63 --- [           main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 47888 (http)
2019-01-25 12:21:06.483  INFO 63 --- [           main] s.c.s.a.h.s.k.HdfsSinkKafka10Application : Started HdfsSinkKafka10Application in 17.593 seconds (JVM running for 18.756)
2019-01-25 12:21:08.250  INFO 63 --- [           -C-1] o.a.k.c.c.internals.AbstractCoordinator  : Discovered coordinator kafka:9092 (id: 2147482646 rack: null) for group testhdfs1.
2019-01-25 12:21:08.256  INFO 63 --- [           -C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Revoking previously assigned partitions [] for group testhdfs1
2019-01-25 12:21:08.256  INFO 63 --- [           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions revoked:[]
2019-01-25 12:21:08.256  INFO 63 --- [           -C-1] o.a.k.c.c.internals.AbstractCoordinator  : (Re-)joining group testhdfs1
2019-01-25 12:21:08.522  INFO 63 --- [           -C-1] o.a.k.c.c.internals.AbstractCoordinator  : Successfully joined group testhdfs1 with generation 1
2019-01-25 12:21:08.526  INFO 63 --- [           -C-1] o.a.k.c.c.internals.ConsumerCoordinator  : Setting newly assigned partitions [testhdfs1.file-0] for group testhdfs1
2019-01-25 12:21:08.735  INFO 63 --- [           -C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$3  : partitions assigned:[testhdfs1.file-0]
2019-01-25 12:21:23.238  INFO 63 --- [           -L-1] o.s.i.codec.kryo.CompositeKryoRegistrar  : registering [40, java.io.File] with serializer org.springframework.integration.codec.kryo.FileSerializer
2019-01-25 12:21:23.353  INFO 63 --- [           -L-1] o.s.d.h.s.o.AbstractDataStreamWriter     : Creating output for path /data/hdfs-sink-0.txt

1 Ответ

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

Вы не можете копировать файлы в hdfs с приемником hdfs, так как он предназначен только для записи произвольных сообщений, полученных из источников.Причина, по которой вы видите нулевую длину этих файлов, заключается в том, что файл все еще открыт и не очищен. Файл hdfs readme содержит параметры конфигурации, и если вы, например, используете настройки idle-timeout или rollover, вы начинаете видеть записанные файлы.

...