Hello,
using the HdfsWriter provided by samza package samza-hdfs we try to write
events to HDFS.
After the newest patches in version 0.10.1 the Bug regarding closing files
was fixed. But the Bucketer does not seem to work with any of the given
HdfsWriter implementations.
Every new event sent to the hdfs outputstream system is creating a new file
on HDFS. According to
documentation "systems.hdfsstream.producer.hdfs.write.batch.size.bytes=67108864"
should have taken care of appending events to an existing file until the
bytes threshold is reached.
Is this a known bug or have I missed sth in my implementation.
Code Snippet:
> # HDFS System
>
> systems.hdfsstream.samza.factory=org.apache.samza.system.hdfs.HdfsSystemFactory
>
> systems.hdfsstream.producer.hdfs.writer.class=org.apache.samza.system.hdfs.writer.BinarySequenceFileHdfsWriter
> systems.hdfsstream.producer.hdfs.write.batch.size.bytes=67108864
> systems.hdfsstream.samza.msg.serde=json
>
> # The base dir for HDFS output. The default Bucketer for SequenceFile
> HdfsWriters
>
> systems.hdfsstream.producer.hdfs.base.output.dir=/user/hive/warehouse/foobar
> # Bucket into following
>
> systems.hdfsstream.producer.hdfs.bucketer.class=org.apache.samza.system.hdfs.writer.JobNameDateTimeBucketer
> systems.hdfsstream.producer.hdfs.bucketer.date.path.format=yyyy
>
>
> private final SystemStream outputStream;
> outputStream = new SystemStream("hdfsstream", "foobarTask");
> >
>
>
@Override
> public void process(IncomingMessageEnvelope envelope, MessageCollector
> messageCollector, TaskCoordinator taskCoordinator) throws
> ClassNotFoundException, SQLException
> {
> final Timer.Context context = registry.timer("foobar").time();
> try
> {
> String incoming = (String) envelope.getMessage();
> GsonBuilder gsonBuilder = new
> GsonBuilder().registerTypeAdapter(DateTime.class, new DateTimeConverter());
> Gson gson = gsonBuilder.create();
> SomeClass message = gson.fromJson(incoming, SomeClass.class);
>
> try{
> messageCollector.send(new
> OutgoingMessageEnvelope(outputStream, gson.toJson(message)));
> registry.counter("foobar").inc();
> }
> catch (Exception e)
> {
> LOGGER.error("error with message: ", e);
> registry.counter("failedProcessCounter").inc();
> }
> }
> finally
> {
> context.stop();
> }
> }
Kind regards
Thees Gieselmann
--
Board of Directors: Jan-Niclaus Mewes, Claas Heiland
Commercial Register: HRB 110377
The information and attached file(s) (if any) contained in this email is
confidential and may be legally privileged. It is intended solely for the
addressee. Any access to this email by persons other than the addressee is
prohibited. If you are not the addressee for whom this email is intended,
you may not disclose, copy, distribute or store this email. If you receive
this in error, please delete and email confirmation to the sender. Thank
you for your cooperation.
